using System.Collections.Generic; namespace System.Threading { /// /// Threadpool implementation that uses a priority queue as the queue for the tasks to execute. /// Features both an automated algorithm to calibrate the number of active threads and a Constructor for constructing /// a threadpool manually, without the auto-managed pool algorithm. /// public class PriorityThreadPool : IDisposable { private class QueueEntry : IComparable { public TaskPriority TaskPriority { get; } public WaitCallback WaitCallback { get; } public object Object { get; } public QueueEntry(TaskPriority priority, WaitCallback callback, object obj) { this.TaskPriority = priority; this.WaitCallback = callback; this.Object = obj; } public int CompareTo(QueueEntry other) { int priority1 = 0, priority2 = 0; switch (this.TaskPriority) { 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 (other.TaskPriority) { 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; } } } #region Enum /// /// Enum for priority of task scheduled /// public enum TaskPriority { /// /// Highest priority of a task. /// Highest, /// /// Priority level above the default level. /// AboveNormal, /// /// Default priority level of all tasks without a specified priority level. /// Normal, /// /// Priority level below the default level. /// BelowNormal, /// /// Lowest priority level. /// Lowest } #endregion #region Fields /// /// List that contains current running threads and their status. /// private volatile List threadpool; private volatile PriorityQueue tasks; private Thread observer; private int maxThreads; private readonly object tasksLock = new object(); private struct WorkerThread { public Thread thread; public bool running, working; } private struct Statistics { public bool Initialized; public int PerformanceCounter; public DateTime LastUpdate; public double LoopFrequency; } #endregion #region Properties /// /// Returns the number of active threads in the threadpool. /// public int NumberOfThreads { get { return threadpool.Count; } } /// /// Returns true if there are no tasks queued. /// public bool Empty { get { return tasks.Count == 0; } } /// /// Maximum amount of threads that the pool can utilize /// public int MaxThreads { set { maxThreads = value; } get { return maxThreads; } } #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(); maxThreads = System.Environment.ProcessorCount; for (int i = 0; i < maxThreads; i++) { WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } observer = new Thread(() => { ObserverLoop(); }); observer.Name = "ThreadPool ObserverThread"; observer.Priority = ThreadPriority.BelowNormal; 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(); this.maxThreads = Math.Max(maxThreads, 1); for (int i = 0; i < maxThreads; i++) { WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } observer = new Thread(() => { ObserverLoop(); }); observer.Name = "ThreadPool ObserverThread"; observer.Priority = ThreadPriority.BelowNormal; 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(); for (int i = 0; i < lowest; i++) { WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.thread.Priority = ThreadPriority.Lowest; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } for (int i = 0; i < belowNormal; i++) { WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.thread.Priority = ThreadPriority.BelowNormal; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } for (int i = 0; i < normal; i++) { WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.thread.Priority = ThreadPriority.Normal; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } for (int i = 0; i < aboveNormal; i++) { WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.thread.Priority = ThreadPriority.AboveNormal; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } for (int i = 0; i < highest; i++) { WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.thread.Priority = ThreadPriority.Highest; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } } #endregion #region Public Methods /// /// Add a work item into the queue. /// /// WaitCallBack delegate that will be invoked by the threads. /// State used as parameter during invoke. /// Priority of task. Affects its position into the queue. public void QueueUserWorkItem(WaitCallback waitCallback, object callbackState, TaskPriority taskPriority) { while (!Monitor.TryEnter(tasksLock)) ; tasks.Enqueue(new QueueEntry(taskPriority, waitCallback, callbackState)); Monitor.Exit(tasksLock); } /// /// Add a work item into the queue. /// /// WaitCallBack delegate that will be invoked by the threads. /// State used as parameter during invoke. public void QueueUserWorkItem(WaitCallback waitCallback, object callbackState) { QueueUserWorkItem(waitCallback, callbackState, TaskPriority.Normal); } #endregion #region Private Methods /// /// Main loop that a thread from the pool is running. /// /// Id of thread private void ThreadMainLoop(ref 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; QueueEntry task = null; while (!Monitor.TryEnter(tasksLock)) ; if (tasks.Count > 0) { task = tasks.Dequeue(); } Monitor.Exit(tasksLock); if (task != null) { System.Diagnostics.Debug.WriteLine(Thread.CurrentThread.Name + " - Running task!"); WaitCallback waitCallback = task.WaitCallback; waitCallback.Invoke(task.Object); } thisWorkerThread.working = false; } } } /// /// Loop of the thread that adjusts and manipulates the threadpool /// private void ObserverLoop() { Statistics statistics = new Statistics(); while (true) { //Observer operates on a 100ms loop. Due to the low priority of the thread itself, this loop will almost always take //considerably longer than 100ms. //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. //This part of code updates the statistics of the threadpool. if (statistics.Initialized) { double loopDuration = (DateTime.Now - statistics.LastUpdate).TotalMilliseconds; if (statistics.LoopFrequency == 0) { statistics.LoopFrequency = loopDuration; } else { statistics.LoopFrequency = (statistics.LoopFrequency + loopDuration) / 2; } } else { statistics.Initialized = true; } statistics.LastUpdate = DateTime.Now; Thread.Sleep(100); //This part of code adjusts the performance counter. Ideally, the value should be 0. if (tasks.Count > 0) { statistics.PerformanceCounter++; if (statistics.PerformanceCounter > 5) { statistics.PerformanceCounter = 5; } } else { statistics.PerformanceCounter--; if (statistics.PerformanceCounter < -10) { statistics.PerformanceCounter = -10; } } //This part of code adjusts thread priorities based on the current performance of the threadpool if (tasks.Count > 0) { //If there are tasks pending, find a thread with priority under Normal and upgrade its priority. Thread t = FindThreadWithLowPriority(); if (t != null) { UpgradeThreadPriority(t); } } else { //If there are no tasks pending, find a thread with priority above Lowest and downgrade its priority. Thread t = FindThreadWithAcceptablePriority(); if (t != null) { DowngradeThreadPriority(t); } } //This part of code modifies the number of active threads based on the current performance of the threadpool if (statistics.PerformanceCounter >= 5) { if (threadpool.Count < this.maxThreads) { //Add a thread to the threadpool. //Reset counter to 0. statistics.PerformanceCounter = 0; WorkerThread worker = new WorkerThread(); worker.thread = new Thread(() => { ThreadMainLoop(ref worker); }); worker.thread.Name = "ThreadPool WorkerThread"; worker.running = true; worker.thread.Start(); threadpool.Add(worker); } } else if (statistics.PerformanceCounter <= -10) { if (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]; if (worker.working) { worker.running = false; } else { worker.thread.Abort(); } threadpool.RemoveAt(threadpool.Count - 1); statistics.PerformanceCounter = 0; } } } } /// /// Find a thread with Lowest or BelowNormal priority. /// /// Thread with low priority. private Thread FindThreadWithLowPriority() { foreach (WorkerThread t in threadpool) { if (t.thread.Priority == ThreadPriority.Lowest || t.thread.Priority == ThreadPriority.BelowNormal) { return t.thread; } } return null; } /// /// Find a thread with BelowNormal or Normal priority. /// /// Thread with BelowNormal or Normal priority. private Thread FindThreadWithAcceptablePriority() { foreach (WorkerThread t in threadpool) { if (t.thread.Priority == ThreadPriority.Normal || t.thread.Priority == ThreadPriority.BelowNormal) { return t.thread; } } return null; } /// /// Downgrades the priority of a thread one level. /// /// Thread to have its priority level downgraded. /// private void DowngradeThreadPriority(Thread t) { switch (t.Priority) { case ThreadPriority.Highest: t.Priority = ThreadPriority.AboveNormal; break; case ThreadPriority.AboveNormal: t.Priority = ThreadPriority.Normal; break; case ThreadPriority.Normal: t.Priority = ThreadPriority.BelowNormal; break; case ThreadPriority.BelowNormal: t.Priority = ThreadPriority.Lowest; break; case ThreadPriority.Lowest: t.Priority = ThreadPriority.Lowest; break; } } /// /// Upgrades the priority of a thread one level. /// /// Thread to have its priority level upgraded. /// private void UpgradeThreadPriority(Thread t) { switch (t.Priority) { case ThreadPriority.Highest: t.Priority = ThreadPriority.Highest; break; case ThreadPriority.AboveNormal: t.Priority = ThreadPriority.Highest; break; case ThreadPriority.Normal: t.Priority = ThreadPriority.AboveNormal; break; case ThreadPriority.BelowNormal: t.Priority = ThreadPriority.Normal; break; case ThreadPriority.Lowest: t.Priority = ThreadPriority.BelowNormal; break; } } #endregion #region IDisposable Support private bool disposedValue = false; /// /// Disposes of the tasks as well as aborts all threads. Called by the public Dispose() method. /// /// protected virtual void Dispose(bool disposing) { if (!disposedValue) { if (disposing) { if (observer != null) { observer.Abort(); } foreach (WorkerThread worker in threadpool) { worker.thread.Abort(); } threadpool.Clear(); tasks.Clear(); } threadpool = null; tasks = null; observer = null; disposedValue = true; } } /// /// Disposes of the tasks as well as aborts all threads. /// public void Dispose() { Dispose(true); } #endregion } }