diff --git a/SystemExtensions/Collections/PriorityQueue.cs b/SystemExtensions/Collections/PriorityQueue.cs
index dbb5b32..998f892 100644
--- a/SystemExtensions/Collections/PriorityQueue.cs
+++ b/SystemExtensions/Collections/PriorityQueue.cs
@@ -51,6 +51,11 @@ namespace SystemExtensions.Collections
{
return binaryHeap.Min;
}
+
+ public void Clear()
+ {
+ binaryHeap.Clear();
+ }
#endregion
#region Private Methods
diff --git a/SystemExtensions/Threading/PriorityThreadPool.cs b/SystemExtensions/Threading/PriorityThreadPool.cs
index 0b355d2..a624309 100644
--- a/SystemExtensions/Threading/PriorityThreadPool.cs
+++ b/SystemExtensions/Threading/PriorityThreadPool.cs
@@ -4,16 +4,386 @@ using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
+using SystemExtensions.Collections;
namespace SystemExtensions.Threading
{
- class PriorityThreadPool
+ public class PriorityThreadPool : IDisposable
{
- public enum Priority
+ #region Enum
+ ///
+ /// Enum for priority of task scheduled
+ ///
+ public enum TaskPriority
{
-
+ Highest,
+ AboveNormal,
+ Normal,
+ BelowNormal,
+ Lowest
}
-
+ #endregion
+ #region Fields
+ ///
+ /// List that contains current running threads and their status.
+ ///
+ private List threadpool;
+ private PriorityQueue> tasks;
+ private Thread observer;
+ private int maxThreads;
+ private object tasksLock = new object();
+ private struct WorkerThread
+ {
+ public Thread thread;
+ public bool running, working;
+ }
+ #endregion
+ #region Properties
+ public int NumberOfThreads
+ {
+ get
+ {
+ return threadpool.Count;
+ }
+ }
+ #endregion
+
+ #region Constructors
+ ///
+ /// Constructor that initializes a threadpool using default values. All threads run at the same priority.
+ /// The maximum number of threads is equal to System.Environment.ProcessorCount
+ ///
+ public PriorityThreadPool()
+ {
+ threadpool = new List();
+ tasks = new PriorityQueue>(PriorityCompare);
+ int nrThreads = (int)Math.Ceiling((double)System.Environment.ProcessorCount / 4);
+ maxThreads = System.Environment.ProcessorCount;
+ for (int i = 0; i < nrThreads; i++)
+ {
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Name = "ThreadPool WorkerThread";
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ observer = new Thread(() =>
+ {
+ ObserverLoop();
+ });
+ observer.Priority = ThreadPriority.Normal;
+ observer.Start();
+ }
+ ///
+ /// Constructor that initializes a threadpool with a number of threads, all of the same priority.
+ ///
+ /// Maximum number of threads
+ public PriorityThreadPool(int maxThreads)
+ {
+ threadpool = new List();
+ tasks = new PriorityQueue>(PriorityCompare);
+ this.maxThreads = Math.Max(maxThreads, 1);
+ int nrThreads = (int)Math.Ceiling((double)this.maxThreads / 4);
+ for (int i = 0; i < nrThreads; i++)
+ {
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Name = "ThreadPool WorkerThread";
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ observer = new Thread(() =>
+ {
+ ObserverLoop();
+ });
+ observer.Priority = ThreadPriority.Normal;
+ observer.Start();
+ }
+ ///
+ /// Constructor for a manually-configured threadpool that runs a specified number of threads.
+ /// This threadpool with not be automatically managed and will always run the specified amount of threads
+ ///
+ /// Number of threads with Lowest priority.
+ /// Number of threads with BelowNormal priority.
+ /// Number of threads with Normal priority.
+ /// Number of threads with AboveNormal priority.
+ /// Number of threads with Highest priority.
+ public PriorityThreadPool(int lowest, int belowNormal, int normal, int aboveNormal, int highest)
+ {
+ threadpool = new List();
+ tasks = new PriorityQueue>(PriorityCompare);
+ for(int i = 0; i < lowest; i++)
+ {
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Name = "ThreadPool WorkerThread";
+ worker.thread.Priority = ThreadPriority.Lowest;
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ for (int i = 0; i < belowNormal; i++)
+ {
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Name = "ThreadPool WorkerThread";
+ worker.thread.Priority = ThreadPriority.BelowNormal;
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ for (int i = 0; i < normal; i++)
+ {
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Name = "ThreadPool WorkerThread";
+ worker.thread.Priority = ThreadPriority.Normal;
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ for (int i = 0; i < aboveNormal; i++)
+ {
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Name = "ThreadPool WorkerThread";
+ worker.thread.Priority = ThreadPriority.AboveNormal;
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ for (int i = 0; i < highest; i++)
+ {
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Name = "ThreadPool WorkerThread";
+ worker.thread.Priority = ThreadPriority.Highest;
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ }
+ #endregion
+
+ #region Public Methods
+ public void QueueUserWorkItem(WaitCallback waitCallback, object callbackState, TaskPriority taskPriority)
+ {
+ while (!Monitor.TryEnter(tasksLock));
+ tasks.Enqueue(new Tuple(taskPriority, waitCallback, callbackState));
+ Monitor.Exit(tasksLock);
+ }
+
+ public void QueueUserWorkItem(WaitCallback waitCallback, object callbackState)
+ {
+ QueueUserWorkItem(waitCallback, callbackState, TaskPriority.Normal);
+ }
+ #endregion
+
+ #region Private Methods
+ ///
+ /// Function that compares two tasks based on their priorities
+ ///
+ /// First tuple to compare
+ /// Second tuple to compare
+ /// 1 if tp1 is bigger. 0 if equal and -1 if tp1 is lower.
+ private static int PriorityCompare(Tuple tp1, Tuple tp2)
+ {
+ int priority1 = 0, priority2 = 0;
+ switch (tp1.Item1)
+ {
+ case TaskPriority.Highest:
+ priority1 = 4;
+ break;
+ case TaskPriority.AboveNormal:
+ priority1 = 3;
+ break;
+ case TaskPriority.Normal:
+ priority1 = 2;
+ break;
+ case TaskPriority.BelowNormal:
+ priority1 = 1;
+ break;
+ case TaskPriority.Lowest:
+ priority1 = 0;
+ break;
+ }
+ switch (tp2.Item1)
+ {
+ case TaskPriority.Highest:
+ priority2 = 4;
+ break;
+ case TaskPriority.AboveNormal:
+ priority2 = 3;
+ break;
+ case TaskPriority.Normal:
+ priority2 = 2;
+ break;
+ case TaskPriority.BelowNormal:
+ priority2 = 1;
+ break;
+ case TaskPriority.Lowest:
+ priority2 = 0;
+ break;
+ }
+ if (priority1 == priority2)
+ {
+ return 0;
+ }
+ else if(priority1 > priority2)
+ {
+ return -1;
+ }
+ else
+ {
+ return 1;
+ }
+ }
+ ///
+ /// Main loop that a thread from the pool is running.
+ ///
+ /// Id of thread
+ private void ThreadMainLoop(WorkerThread thisWorkerThread)
+ {
+ thisWorkerThread.running = true;
+ while (thisWorkerThread.running)
+ {
+ //Check if there are tasks
+
+ if (tasks.Count > 0)
+ {
+ //If there are tasks, acquire a lock onto the list
+ //and try to dequeue the highest priority task.
+ //Finally, release the lock and invoke the task.
+
+ thisWorkerThread.working = true;
+ Tuple task = null;
+ while (!Monitor.TryEnter(tasksLock)) ;
+ if (tasks.Count > 0)
+ {
+ task = tasks.Dequeue();
+ }
+ Monitor.Exit(tasksLock);
+ if (task != null)
+ {
+ WaitCallback waitCallback = task.Item2;
+ waitCallback.Invoke(task.Item3);
+ }
+ thisWorkerThread.working = false;
+ }
+ }
+ }
+ ///
+ /// Loop of the thread that adjusts and manipulates the threadpool
+ ///
+ private void ObserverLoop()
+ {
+ int counter = 0;
+ while (true)
+ {
+ //Observer operates on a 100ms loop.
+ //Checks if the queue is empty. If yes, counter is decremented, else, counter is incremented.
+ //If counter exceeds 5, it will try to add another thread to the thread, unless the threadpool has reached max size.
+ //If counter is under -10, it will try to remove a thread from the threadpool, unless the threadpool has reached less than
+ //max size / 4.
+ Thread.Sleep(100);
+ if(tasks.Count > 0)
+ {
+ counter++;
+ if(counter > 5)
+ {
+ counter = 5;
+ }
+ }
+ else
+ {
+ counter--;
+ if(counter < -10)
+ {
+ counter = -10;
+ }
+ }
+ if(counter >= 5 && threadpool.Count < this.maxThreads)
+ {
+ //Add a thread to the threadpool.
+ //Reset counter to 0.
+ counter = 0;
+ WorkerThread worker = new WorkerThread();
+ worker.thread = new Thread(() =>
+ {
+ ThreadMainLoop(worker);
+ });
+ worker.thread.Start();
+ threadpool.Add(worker);
+ }
+ if(counter <= -10 && threadpool.Count > maxThreads / 4)
+ {
+ //Remove the last thread in the threadpool.
+ //If thread is currently working, notify it to close.
+ //Else, abort the thread.
+ //Reset counter to 0.
+ WorkerThread worker = threadpool[threadpool.Count - 1];
+ threadpool.RemoveAt(threadpool.Count - 1);
+ if (worker.working)
+ {
+ worker.running = false;
+ }
+ else
+ {
+ worker.thread.Abort();
+ }
+ counter = 0;
+ }
+ }
+ }
+ #endregion
+ #region IDisposable Support
+ private bool disposedValue = false;
+
+ protected virtual void Dispose(bool disposing)
+ {
+ if (!disposedValue)
+ {
+ if (disposing)
+ {
+ foreach(WorkerThread worker in threadpool)
+ {
+ worker.thread.Abort();
+ }
+ threadpool.Clear();
+ tasks.Clear();
+ if (observer != null)
+ {
+ observer.Abort();
+ }
+ }
+ threadpool = null;
+ tasks = null;
+ observer = null;
+ disposedValue = true;
+ }
+ }
+
+ public void Dispose()
+ {
+ Dispose(true);
+ }
+ #endregion
}
}
diff --git a/SystemExtensionsTests/SystemExtensionsTests.csproj b/SystemExtensionsTests/SystemExtensionsTests.csproj
index a228070..f8ff112 100644
--- a/SystemExtensionsTests/SystemExtensionsTests.csproj
+++ b/SystemExtensionsTests/SystemExtensionsTests.csproj
@@ -59,6 +59,7 @@
+
diff --git a/SystemExtensionsTests/Threading/PriorityThreadPoolTests.cs b/SystemExtensionsTests/Threading/PriorityThreadPoolTests.cs
new file mode 100644
index 0000000..8091ccc
--- /dev/null
+++ b/SystemExtensionsTests/Threading/PriorityThreadPoolTests.cs
@@ -0,0 +1,222 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using SystemExtensions.Threading;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using System.Threading;
+using System.Diagnostics;
+using static SystemExtensions.Threading.PriorityThreadPool;
+
+namespace SystemExtensions.Threading.Tests
+{
+ [TestClass()]
+ public class PriorityThreadPoolTests
+ {
+ private static PriorityThreadPool threadPool = new PriorityThreadPool();
+ private static int lowest = 0, belowAverage = 0,
+ normal = 0, aboveAverage = 0, highest = 0;
+ [TestMethod()]
+ public void PriorityThreadPoolTest()
+ {
+ threadPool.Dispose();
+ threadPool = new PriorityThreadPool();
+ if (threadPool.NumberOfThreads != System.Environment.ProcessorCount / 4)
+ {
+ Assert.Fail();
+ }
+ }
+
+ [TestMethod()]
+ public void PriorityThreadPoolTest1()
+ {
+ threadPool.Dispose();
+ threadPool = new PriorityThreadPool(4);
+ if (threadPool.NumberOfThreads != 4 / 4)
+ {
+ Assert.Fail();
+ }
+ }
+
+ [TestMethod()]
+ public void PriorityThreadPoolTest2()
+ {
+ threadPool.Dispose();
+ threadPool = new PriorityThreadPool(1, 1, 1, 1, 1);
+ if (threadPool.NumberOfThreads != 5)
+ {
+ Assert.Fail();
+ }
+ }
+
+ [TestMethod()]
+ public void DisposeTest()
+ {
+ threadPool.Dispose();
+ Thread.Sleep(100);
+ try
+ {
+ int x = threadPool.NumberOfThreads;
+ Assert.Fail();
+ }
+ catch(Exception e)
+ {
+
+ }
+ }
+
+ [TestMethod()]
+ public void DefaultQueueTest()
+ {
+ threadPool.Dispose();
+ threadPool = new PriorityThreadPool();
+ for (int i = 0; i < 10; i++)
+ {
+ threadPool.QueueUserWorkItem(o =>
+ {
+ for (int x = 0; x < 100; x++)
+ {
+ System.Diagnostics.Debug.WriteLine(Thread.CurrentThread.ManagedThreadId + " - " + x);
+ }
+ }, null);
+ }
+ for (int i = 0; i < 10; i++)
+ {
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ System.Diagnostics.Debug.WriteLine("Current threads in threadpool: " + threadPool.NumberOfThreads);
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ Thread.Sleep(100);
+ }
+ while (threadPool.NumberOfThreads != System.Environment.ProcessorCount / 4)
+ {
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ System.Diagnostics.Debug.WriteLine("Current threads in threadpool: " + threadPool.NumberOfThreads);
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ Thread.Sleep(100);
+ }
+ }
+
+ [TestMethod()]
+ public void QueueTestWith4MaxThreads()
+ {
+ threadPool.Dispose();
+ threadPool = new PriorityThreadPool(4);
+ if (threadPool.NumberOfThreads != 4 / 4)
+ {
+ Assert.Fail();
+ }
+ for (int i = 0; i < 10; i++)
+ {
+ threadPool.QueueUserWorkItem(o =>
+ {
+ for (int x = 0; x < 100; x++)
+ {
+ System.Diagnostics.Debug.WriteLine(Thread.CurrentThread.ManagedThreadId + " - " + x);
+ }
+ }, null);
+ }
+ for (int i = 0; i < 10; i++)
+ {
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ System.Diagnostics.Debug.WriteLine("Current threads in threadpool: " + threadPool.NumberOfThreads);
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ Thread.Sleep(100);
+ }
+ while (threadPool.NumberOfThreads != 1)
+ {
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ System.Diagnostics.Debug.WriteLine("Current threads in threadpool: " + threadPool.NumberOfThreads);
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ Thread.Sleep(100);
+ }
+ }
+
+ [TestMethod()]
+ public void QueueTestWith5ManualThreads()
+ {
+ threadPool.Dispose();
+ threadPool = new PriorityThreadPool(1, 1, 1, 1, 1);
+ if (threadPool.NumberOfThreads != 5)
+ {
+ Assert.Fail();
+ }
+ for (int i = 0; i < 10; i++)
+ {
+ threadPool.QueueUserWorkItem(o =>
+ {
+ for (int x = 0; x < 100; x++)
+ {
+ switch (Thread.CurrentThread.Priority)
+ {
+ case ThreadPriority.Lowest:
+ lowest++;
+ break;
+ case ThreadPriority.BelowNormal:
+ belowAverage++;
+ break;
+ case ThreadPriority.Normal:
+ normal++;
+ break;
+ case ThreadPriority.AboveNormal:
+ aboveAverage++;
+ break;
+ case ThreadPriority.Highest:
+ highest++;
+ break;
+ }
+ }
+ }, null);
+ }
+ for (int i = 0; i < 50; i++)
+ {
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ System.Diagnostics.Debug.WriteLine("Current threads in threadpool: " + threadPool.NumberOfThreads);
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ Thread.Sleep(100);
+ }
+ System.Diagnostics.Debug.WriteLine("\nL: " + lowest + "\nBN: " + belowAverage + "\nN: " + normal + "\nAN:" + aboveAverage + "\nH: " + highest);
+ }
+
+ [TestMethod()]
+ public void QueueTestWithDifferentPriorityTasks()
+ {
+ threadPool.Dispose();
+ threadPool = new PriorityThreadPool();
+ for (int i = 0; i < 1000; i++)
+ {
+ TaskPriority taskPriority = TaskPriority.Lowest;
+ if(i % 10 == 0)
+ {
+ taskPriority = TaskPriority.Highest;
+ }
+ else if(i % 5 == 0)
+ {
+ taskPriority = TaskPriority.Normal;
+ }
+ else if(i % 2 == 0)
+ {
+ taskPriority = TaskPriority.BelowNormal;
+ }
+ threadPool.QueueUserWorkItem(o =>
+ {
+ System.Diagnostics.Debug.WriteLine(Thread.CurrentThread.ManagedThreadId + " - " + taskPriority);
+ }, null, taskPriority);
+ }
+ for (int i = 0; i < 10; i++)
+ {
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ System.Diagnostics.Debug.WriteLine("Current threads in threadpool: " + threadPool.NumberOfThreads);
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ Thread.Sleep(100);
+ }
+ while (threadPool.NumberOfThreads != System.Environment.ProcessorCount / 4)
+ {
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ System.Diagnostics.Debug.WriteLine("Current threads in threadpool: " + threadPool.NumberOfThreads);
+ System.Diagnostics.Debug.WriteLine("=====================================");
+ Thread.Sleep(100);
+ }
+ }
+ }
+}
\ No newline at end of file