Thread pool

ThreadPool/ThreadPool.sln

@@ -0,0 +1,22 @@
+
+Microsoft Visual Studio Solution File, Format Version 12.00
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ThreadPool", "ThreadPool\ThreadPool.csproj", "{E0D93E78-090C-4C9B-B545-DBB1534BC37D}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ThreadPoolTest", "ThreadPoolTest\ThreadPoolTest.csproj", "{AE21AEB8-EC3E-4D26-A154-E8C11BA4F93E}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{E0D93E78-090C-4C9B-B545-DBB1534BC37D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{E0D93E78-090C-4C9B-B545-DBB1534BC37D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{E0D93E78-090C-4C9B-B545-DBB1534BC37D}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{E0D93E78-090C-4C9B-B545-DBB1534BC37D}.Release|Any CPU.Build.0 = Release|Any CPU
+		{AE21AEB8-EC3E-4D26-A154-E8C11BA4F93E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{AE21AEB8-EC3E-4D26-A154-E8C11BA4F93E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{AE21AEB8-EC3E-4D26-A154-E8C11BA4F93E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{AE21AEB8-EC3E-4D26-A154-E8C11BA4F93E}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+EndGlobal

ThreadPool/ThreadPool/IMyTask.cs

@@ -0,0 +1,25 @@
+namespace ThreadPool;
+
+using System;
+
+/// <summary>
+/// interface for object from MyThreadPool
+/// </summary>
+public interface IMyTask<out TResult>
+{
+    /// <summary>
+    /// shows whether the task is completed or not
+    /// </summary>
+    public bool IsCompleted { get; set; }
+
+    /// <summary>
+    /// returns the result of the task
+    /// </summary>
+    public TResult Result { get; }
+
+    /// <summary>
+    ///  continues calculating
+    /// </summary>
+    public IMyTask<TNewResult> ContinueWith<TNewResult>(Func<TResult, TNewResult> func);
+}
+

ThreadPool/ThreadPool/MyThreadPool.cs

@@ -0,0 +1,208 @@
+namespace ThreadPool;
+
+using System.Collections.Generic;
+using System.Threading;
+using System.Collections.Concurrent;
+using System;
+
+/// <summary>
+/// pool of threads that can be used to execute tasks
+/// </summary>
+public class MyThreadPool
+{
+    private readonly Thread[] _threads;
+    private readonly CancellationTokenSource _token = new();
+    private readonly ConcurrentQueue<Action> _tasksQueue = new();
+    private readonly AutoResetEvent _waiterNewTask = new AutoResetEvent(false);
+    private readonly AutoResetEvent _waiterTaskDone = new AutoResetEvent(false);
+    private volatile int _freeThreadsCount = 0;
+    private readonly object _lockObject = new();
+
+    /// <summary>
+    /// constructor
+    /// </summary>
+    public MyThreadPool(int countThreads)
+    {
+        if (countThreads < 1)
+        {
+            throw new ArgumentOutOfRangeException(nameof(countThreads));
+        }
+
+        _threads = new Thread[countThreads];
+        for (int i = 0; i < countThreads; i++)
+        {
+            _threads[i] = new Thread(() =>
+            {
+                while (!_token.IsCancellationRequested || !_tasksQueue.IsEmpty)
+                {
+                    if (_tasksQueue.TryDequeue(out Action action))
+                    {
+                        action();
+                    }
+                    else
+                    {
+                        _waiterNewTask.WaitOne();
+                        if (!_tasksQueue.IsEmpty)
+                        {
+                            _waiterNewTask.Set();
+                        }
+                    }
+                }
+
+                Interlocked.Increment(ref _freeThreadsCount);
+                _waiterTaskDone.Set();
+            });
+
+            _threads[i].Start();
+        }
+    }
+
+    /// <summary>
+    /// add task 
+    /// </summary>
+    public IMyTask<TResult> AddTask<TResult>(Func<TResult> function)
+    {
+        ReadyToAddNewTask();
+        lock (_lockObject)
+        {
+            ReadyToAddNewTask();
+            var task = new MyTask<TResult>(function, this);
+            AddInTaskQueue(task.Count);
+            return task;
+        }
+    }
+
+    private void ReadyToAddNewTask()
+    {
+        if (_token.IsCancellationRequested)
+        {
+            throw new InvalidOperationException();
+        }
+    }
+
+    private void AddInTaskQueue(Action task)
+    {
+        _tasksQueue.Enqueue(task);
+        _waiterNewTask.Set();
+    }
+
+    /// <summary>
+    /// close threads
+    /// </summary>
+    public void Shutdown()
+    {
+        lock (_lockObject)
+        {
+            _token.Cancel();
+        }
+
+        while (_freeThreadsCount != _threads.Length)
+        {
+            _waiterNewTask.Set();
+            _waiterTaskDone.WaitOne();
+        }
+    }
+
+    /// <summary>
+    /// class for parallel tasks
+    /// </summary>
+    private class MyTask<TResult> : IMyTask<TResult>
+    {
+        private Func<TResult> _func;
+        private TResult _result;
+        private Exception _resultException = null;
+        private readonly object _locker = new();
+        private readonly ManualResetEvent _waiterManual = new(false);
+        private Queue<Action> _continueWithTasksQueue = new();
+        private readonly MyThreadPool _myThreadPool;
+
+        /// <summary>
+        /// shows whether the task is completed or not
+        /// </summary>
+        public bool IsCompleted { get; set; }
+
+        /// <summary>
+        /// returns the result of the task
+        /// </summary>
+        public TResult Result
+        {
+            get
+            {
+                _waiterManual.WaitOne();
+                if (_resultException != null)
+                {
+                    throw new AggregateException(_resultException);
+                }
+
+                return _result;
+            }
+        }
+
+        /// <summary>
+        ///  constructor
+        /// </summary>
+        public MyTask(Func<TResult> function, MyThreadPool myThreadPool)
+        {
+            ArgumentNullException.ThrowIfNull(function);
+            _func = function;
+            _myThreadPool = myThreadPool;
+        }
+
+        /// <summary>
+        /// counting result of task
+        /// </summary>
+        public void Count()
+        {
+            try
+            {
+                _result = _func();
+            }
+            catch (Exception funcException)
+            {
+                _resultException = funcException;
+            }
+
+            lock (_locker)
+            {
+                _func = null;
+                IsCompleted = true;
+                _waiterManual.Set();
+            }
+
+            lock (_locker)
+            {
+                while (_continueWithTasksQueue.Count > 0)
+                {
+                    var action = _continueWithTasksQueue.Dequeue();
+                    lock (_myThreadPool._lockObject)
+                    {
+                        _myThreadPool.AddInTaskQueue(action);
+                    }
+                }
+            }
+        }
+
+        /// <summary>
+        ///  continues calculating
+        /// </summary>
+        public IMyTask<TNewResult> ContinueWith<TNewResult>(Func<TResult, TNewResult> func)
+        {
+            lock (_myThreadPool._lockObject)
+            {
+                _myThreadPool.ReadyToAddNewTask();
+                var task = new MyTask<TNewResult>(() => func(Result), _myThreadPool);
+                _continueWithTasksQueue.Enqueue(task.Count);
+                if (IsCompleted)
+                {
+                    while (_continueWithTasksQueue.Count > 0)
+                    {
+                        var action = _continueWithTasksQueue.Dequeue();
+                        _myThreadPool._tasksQueue.Enqueue(action);
+                        _myThreadPool._waiterNewTask.Set();
+                    }
+                }
+                return task;
+            }
+        }
+    }
+}

ThreadPool/ThreadPool/ThreadPool.csproj

@@ -0,0 +1,10 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net6.0</TargetFramework>
+        <DockerDefaultTargetOS>Windows</DockerDefaultTargetOS>
+        <LangVersion>10</LangVersion>
+        <Nullable>disable</Nullable>
+    </PropertyGroup>
+
+</Project>