The following code is a self-implementation of the static class ThreadPool in C#, only the QueueUserWorkItem (the simpler method), written to practice multithreading in the .NET environment and generally C# coding.
Implementation details:
The background threads are simulated using Thread objects.
Enqueuing and dequeuing of jobs is done with a ConcurrentQueue of Thread.
Once dequeued, a new worker thread is created with the job as its parameter. Each worker thread does its first job and then enters an infinite while loop, that keeps dequeuing jobs from the queue and process them until the queue is empty. Once that happens, that thread atomically increments a number variable that indicates the amount of threads that have finished running.
Once the background manager thread kicks in, it is able to respond to this by looking for a single thread that's stopped running, in order to either remove it from the list in case that there's no more jobs left in the queue, or replace the finished Thread object (causing it to be available for garbage collection) with the newly created thread with the latest dequeue'd job.
There's probably an excessive use of locks here and maybe some bugs.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
namespace ThreadPool
{
public delegate void WaitCallback(Object state);
public static class ThreadPool
{
private static int _maxThreads = 1000; // an assumption.
public static int FinishedThreads = 0;
public static int TotalQueued = 0;
public static int MaxThreadsUsed = 0;
private static ConcurrentQueue<WaitCallback> _queue = new ConcurrentQueue<WaitCallback>();
private static List<Thread> _workers = new List<Thread>(100);
private static Thread _backgroundThread = new Thread(() =>
{
while (true)
{
DequeueRun();
Thread.Sleep(50);
}
});
public static Boolean QueueUserWorkItem(WaitCallback callBack)
{
if (callBack == null)
{
throw new ArgumentNullException("callBack is null.");
}
//check for: NotSupportedException: The common language runtime (CLR) is hosted, and the host does not support this action.
lock (_queue)
{
_queue.Enqueue(callBack);
TotalQueued++;
}
lock (_backgroundThread)
{
if (_backgroundThread.ThreadState == ThreadState.Unstarted)
{
_backgroundThread.IsBackground = true;
_backgroundThread.Start();
}
}
return true;
}
private static void DequeueRun()
{
WaitCallback wcb = null;
Thread stoppedThread = null;
lock (_workers)
{
if (FinishedThreads > 0) //not an atomic read
{ //there may be threads need to be renewed / removed
for (int i = 0; i < _workers.Count; i++)
{
if (_workers[i].ThreadState == ThreadState.Stopped)
{
stoppedThread = _workers[i]; //get a reference to the stopped thread
}
}
}
if (_workers.Count > MaxThreadsUsed) //reassign the maximum threads used
{
MaxThreadsUsed = _workers.Count;
}
lock (_queue)
{
if (_queue.Count > 0 && _maxThreads - _workers.Count > 0)
{
_queue.TryDequeue(out wcb); //if there's still work, dequeue something
}
}
} //release everything while preparing a new independant thread
if (wcb == null && stoppedThread != null) //dequeued nothing and there exist a stopped thread
{
bool empty = false;
lock (_queue) //find out if the queue is empty
{
empty = _queue.Count == 0;
}
if (empty)
{
lock (_workers)
{
if (!_workers.Remove(stoppedThread))
{
throw new SystemException("Could not remove a thread from the list!");
}
else
{
Interlocked.Decrement(ref FinishedThreads);
}
}
}
}
else if (wcb != null)
{
Thread thread = new Thread(new ParameterizedThreadStart(ThreadFunc));
/* The threads in the managed thread pool are background threads.
* That is, their IsBackground properties are true.
* This means that a ThreadPool thread will not keep an application running after all foreground threads have exited. */
thread.IsBackground = true;
lock (_workers)
{
if (_maxThreads - _workers.Count > 0)
{
if (stoppedThread != null) //replace the stopped thread with a new one
{
_workers[_workers.IndexOf(stoppedThread)] = thread;
}
else
{
_workers.Add(thread);
}
}
else
{
lock (_queue) //let other active threads do the work.
{
_queue.Enqueue(wcb);
wcb = null;
}
}
}
if (wcb != null)
{
thread.Start(wcb);
}
}
}
public static void ThreadFunc(Object obj)
{
//when a thread is started, it already has a job assigned to it.
WaitCallback wcb = obj as WaitCallback;
if (wcb == null)
{
throw new ArgumentException("ThreadFunc must recieve WaitCallback as a parameter!");
}
wcb.Invoke(null);
while (true) //from now on, I'm dequeueing/invoking jobs from the queue.
{
wcb = null;
lock (_workers)
{
lock (_queue)
{
if (_queue.Count > 0) //help a non-empty queue to get rid of its load
{
_queue.TryDequeue(out wcb);
}
}
}
if (wcb != null)
{
wcb.Invoke(null);
}
else
{ //could not dequeue from the queue, terminate the thread
Interlocked.Increment(ref FinishedThreads);
return;
}
Thread.Sleep(0); //context switch
}
}
}
}
Usage example:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace ThreadPool
{
public sealed class Program
{
public static int NumberOfThreadsWrote = 0;
public static void Main(string[] args)
{
// Queue some tasks.
for (int i = 0; i < 5000; i++)
{
ThreadPool.QueueUserWorkItem(new WaitCallback(ThreadProc));
}
Console.WriteLine("Main thread does some work, then sleeps.");
// If you comment out the Sleep, the main thread exits before
// the thread pool task runs. The thread pool uses background
// threads, which do not keep the application running. (This
// is a simple example of a race condition.)
Thread.Sleep(1 * 7 * 1000);
Console.WriteLine("Completed {0} tasks using a maximum concurrency of {1} threads.", NumberOfThreadsWrote, ThreadPool.MaxThreadsUsed);
Console.WriteLine("Total items queued: {0} and now there are {1} threads in the inner list of the thread pool.", ThreadPool.TotalQueued, ThreadPool.FinishedThreads);
Console.WriteLine("Main thread exits.");
}
// This thread procedure performs the task.
public static void ThreadProc(Object stateInfo)
{
// No state object was passed to QueueUserWorkItem, so
// stateInfo is null.
int number = Interlocked.Increment(ref NumberOfThreadsWrote);
Console.WriteLine("Hello from the thread pool. Done {0} tasks.", number);
}
}
}
