2
\$\begingroup\$

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);
        }
    }
}
\$\endgroup\$
2
\$\begingroup\$
public static int FinishedThreads = 0;
public static int TotalQueued = 0;
public static int MaxThreadsUsed = 0;

Public fields are usually not a good idea, you should make them private.

while (true)
{
    DequeueRun();
    Thread.Sleep(50);
}

You shouldn't busy wait like this. Instead, you should use some synchronization that tells this thread it has work to do.

lock (_queue)
{
    _queue.Enqueue(callBack);
    TotalQueued++;
}

The whole point of ConcurrentQueue is that you don't need locking to access it. If you needed to rely on TotalQueued being in sync with _queue, then the lock would make sense, but in that case, you should use a normal Queue instead.


Also, I think ThreadFunc() should wait for a while even when it has no work to do. This way, if the user enqueues a new work item right after previous work item finishes, you're not going to destroy and create thread unnecessarily.

\$\endgroup\$
  • \$\begingroup\$ I think that the main reason I used ConcurrentQueue and also put a lock on it was in cases you want to check if the queue Count is some positive value and in the same breathe be able to manipulate it, like in these lines: if (_queue.Count > 0 && _maxThreads - _workers.Count > 0) { _queue.TryDequeue(out wcb); //if there's still work, dequeue something } What'll happen when you remove the surrounding lock there? Is it possible to get a positive value from _queue.Count then another thread dequeues right after it, \$\endgroup\$ – rycle May 5 '13 at 22:36
  • \$\begingroup\$ causing the next call to TryDequeue from this thread to find nothing inside? Or is the compiler smart enough to optimize both the Count check and the TryDequeue operations in the same critical code? Guess there's only one way to verify (IL). \$\endgroup\$ – rycle May 5 '13 at 22:37
  • \$\begingroup\$ @rycle No, the compiler certainly doesn't do anything like that. A better way is to write your code in a way that doesn't rely on checking Count at all. \$\endgroup\$ – svick May 6 '13 at 0:27
  • \$\begingroup\$ i'm wrong or if you do something like: public static int FinishedThreads { get; private set; } the public would be ok? \$\endgroup\$ – Marco Acierno Jun 12 '14 at 21:01
  • \$\begingroup\$ @MarcoAcierno Yeah, if you need to expose it, that's exactly the right way to do it. \$\endgroup\$ – svick Jun 12 '14 at 21:21
4
\$\begingroup\$

I would probably change your thread pool class to work off of Semaphores. This simplifies a bunch of the logic and removes all of the sleep statements from the code.

I would also change to have threads exit once there is no more work to do (although that is easy to change in the code below). The following never got to more than 1 thread spawned to handle the 5000 units of Console.Writeline so I also had to increase the amount of time spent in the delegates.

using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;

namespace ThreadPool
{
    public delegate void WaitCallback(Object state);

    public static class MyThreadPool
    {
        private static int _maxThreads = 1000; // an assumption.
        private static long _workingThreadCount = 0;
        private static long _runningThreadCount = 0;
        public static long FinishedThreads = 0;
        public static long MaxThreadsUsed = 0;
        public static long TotalQueued = 0;
        private static Queue<WaitCallback> _queue = new Queue<WaitCallback>();
        private static List<Thread> _workers = new List<Thread>();
        static Semaphore deQueueSemaphore = new Semaphore(0, 100000); //max 100000 items in the queue at a time.

        public static Boolean QueueUserWorkItem(WaitCallback callBack)
        {
            if (callBack == null)
            {
                throw new ArgumentNullException("callBack is null.");
            }

            lock (_queue)
            {
                ++TotalQueued;
                _queue.Enqueue(callBack);

                //only spawn a thread if we have not hit the max.
                if (_maxThreads > _workers.Count)
                {
                    long busyThreads = Interlocked.Read(ref _workingThreadCount);

                    //Only spawn a thread if you dont have idle background threads.
                    if (busyThreads >= _workers.Count)
                    {
                        System.Diagnostics.Debug.WriteLine("Starting Thread.");
                        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;

                        _workers.Add(thread);
                        thread.Start(null);
                        ++MaxThreadsUsed;
                        Interlocked.Increment(ref _runningThreadCount);
                    }
                }

                deQueueSemaphore.Release();
            }

            return true;
        }

        public static bool HasRunningJobs()
        {
            bool running = (0 != Interlocked.Read(ref _workingThreadCount));

            if (running)
                return true;

            lock (_queue)
            {
                return (0 != _queue.Count && 0 != _runningThreadCount);
            }
        }

        public static void ThreadFunc(Object obj)
        {
            while (true) //from now on, I'm dequeueing/invoking jobs from the queue.
            {
                try
                {
                    deQueueSemaphore.WaitOne();

                    Interlocked.Increment(ref _workingThreadCount);

                    WaitCallback wcb = null;

                    lock (_queue)
                    {
                        if (_queue.Count > 0) //help a non-empty queue to get rid of its load
                        {
                            wcb = _queue.Dequeue();
                        }
                    }

                    if (wcb != null)
                    {
                        wcb.Invoke(null);
                    }
                    else
                    {
                        System.Diagnostics.Debug.WriteLine("Exiting Thread");
                        //could not dequeue from the queue, terminate the thread
                        Interlocked.Increment(ref FinishedThreads);
                        Interlocked.Decrement(ref _runningThreadCount);
                        return;
                    }
                }
                finally
                {
                    deQueueSemaphore.Release();
                    Interlocked.Decrement(ref _workingThreadCount);
                }
            }

        }
    }
}

Here is the code for the main class:

public class Program
{
    public static int NumberOfThreadsWrote = 0;
    public static Random rand = new Random();

    public static void Main(string[] args)
    {
        // Queue some tasks.
        for (int i = 0; i < 5000; i++)

        {
            MyThreadPool.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.)
        while (MyThreadPool.HasRunningJobs())
        {
            Thread.Sleep(1000);
        }

        //NOTE: threads will exit on their own when there is nothing in the queue for them to do.

        Console.WriteLine("Completed {0} tasks using a maximum concurrency of {1} threads.", NumberOfThreadsWrote, MyThreadPool.MaxThreadsUsed);
        Console.WriteLine("Total items queued: {0} and now there are {1} threads in the inner list of the thread pool.", MyThreadPool.TotalQueued, MyThreadPool.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.
        System.Threading.Thread.Sleep(rand.Next() % 100);
        int number = Interlocked.Increment(ref NumberOfThreadsWrote);
        Console.WriteLine("Hello from the thread pool. Done {0} tasks.", number);
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.