A custom thread-pool/queue class

Question

I wanted a class which executes any number of tasks but only a certain amount at the same time (e.g. to download various internet content and keep the overall download speed at a good level). The class I wrote seems to work but there are probably things that can be improved.

1. Do these locks make sense? Should there be other locks as well?
2. I have a method and an event that is only relevant if a specific constructor is used. Is there a way to improve that?
3. The class I use for tasks is ThreadStart. Is that a good idea?
4. There might be better method names/class names.
5. Are there any general errors (e.g. that more Threads than the max-count will be executed)?

There are probably more points that could be improved. Also, if anyone knows a good open source library (or native .NET class even) which does just what my class is supposed to do, I would be interested in that too.

public class ThreadQueue
{
    private readonly HashSet<ThreadStart> WorkingThreads = new HashSet<ThreadStart>();
    private readonly Queue<ThreadStart> Queue = new Queue<ThreadStart>();

    private bool RaiseCompleteEventIfQueueEmpty = false;

    private int ThreadsMaxCount;

    public ThreadQueue(int threadsMaxCount)
    {
        ThreadsMaxCount = threadsMaxCount;
    }

    /// <summary>
    /// Creates a new thread queue with a maximum number of threads and the tasks that should be executed.
    /// </summary>
    /// <param name="threadsMaxCount">The maximum number of currently active threads.</param>
    /// <param name="tasks">The tasks that should be executed by the queue.</param>
    public ThreadQueue(int threadsMaxCount, ThreadStart[] tasks) : this(threadsMaxCount)
    {
        RaiseCompleteEventIfQueueEmpty = true;
        foreach (ThreadStart task in tasks)
        {
            Queue.Enqueue(task);
        }
    }

    /// <summary>
    /// Starts to execute tasks. Used in conjunction with the constructor in which all tasks are provided.
    /// </summary>
    public void Start()
    {
        CheckQueue();
    }

    private readonly object addlock = new object();
    /// <summary>
    /// Adds a task and runs it if a execution slot is free. Otherwise it will be enqueued.
    /// </summary>
    /// <param name="task">The task that should be executed.</param>
    public void AddTask(ThreadStart task)
    {
        lock (addlock)
        {
            if (WorkingThreads.Count == ThreadsMaxCount)
            {
                Queue.Enqueue(task);
            }
            else
            {
                StartThread(task);
            }
        }
    }

    /// <summary>
    /// Starts the execution of a task.
    /// </summary>
    /// <param name="task">The task that should be executed.</param>
    private void StartThread(ThreadStart task)
    {
        WorkingThreads.Add(task);
        BackgroundWorker thread = new BackgroundWorker();
        thread.DoWork += delegate { task.Invoke(); };
        thread.RunWorkerCompleted += delegate { ThreadCompleted(task); };
        thread.RunWorkerAsync();
    }

    private void ThreadCompleted(ThreadStart start)
    {
        WorkingThreads.Remove(start);
        CheckQueue();
        if (Queue.Count == 0 && WorkingThreads.Count == 0 && RaiseCompleteEventIfQueueEmpty) OnCompleted();
    }

    private readonly object checklock = new object();
    /// <summary>
    /// Checks if the queue contains tasks and runs as many as there are free execution slots.
    /// </summary>
    private void CheckQueue()
    {
        lock (checklock)
        {
            while (Queue.Count > 0 && WorkingThreads.Count < ThreadsMaxCount)
            {
                StartThread(Queue.Dequeue());
            }
            if (Queue.Count == 0 && WorkingThreads.Count == 0 && RaiseCompleteEventIfQueueEmpty) OnCompleted();
        }
    }

    /// <summary>
    /// Raised when all tasks have been completed. Will only be used if the ThreadQueue has been initialized with all the tasks it should execute.
    /// </summary>
    public event EventHandler Completed;

    /// <summary>
    /// Raises the Completed event.
    /// </summary>
    protected void OnCompleted()
    {
        if (Completed != null)
        {
            Completed(this, null);
        }
    }
}

Answer 1

There is a lot that needs to be taken into consideration when implementing a threaded work queue and I wouldn't recommend doing it manually unless you've got a really good reason because existing solutions are tested and reliable and without very specific design needs and some hardcore coding to go with it you'll likely wind up with something less efficient/powerful than the existing options. That said, as an exercise this can be a very educational challenge and I'll try to review it as such; production relevant notes are below.

Technically it's just a matter of preference, but generally speaking, private member variables don't have capitalized first letters to help differentiate between public members and internal implementation details (typical .NET style is CamelCase for public members and pascalCase for fields/variables/parameters).

It seems you are using the Complete event for two different meanings; you should make this two separate events. I wouldn't bother with RaiseCompleteEventIfQueueEmpty, instead just have an ItemComplete event and a QueueComplete event or something like that and always raise the events; if the caller doesn't care about one or the other (or either) they simply don't attach handlers to them.

There are synchronization issues due to your locking strategy. Because your AddTask and CheckQueue methods use different lock objects it's possible to be adding to and accessing the queue at the same time. Because Queue is not inherently thread-safe this could cause issues. You're better off using a single lock between the two. If you're on framework 4.0 you may want to consider using ConcurrentQueue instead.

BackgroundWorker contains provisions intended to make it easy to setup simple background tasks separate from a UI thread. BackgroundWorker is also a Component, which technically means you should Dispose of it when you're done (not from its own thread) although I don't think its implementation depends on it currently, that could change and you want to follow proper practices regarding IDisposable. It doesn't look like you need much of the convenience it offers and are mainly using it for the completion event, which you could implement yourself pretty easily. You would have more control and less overhead using Thread which should also keep you conscious of everything that's happening in something this low level. Here's a brief discussion on Stack Overflow. Also worth noting: BackgroundWorker uses BeginInvoke internally which executes the work on a ThreadPool thread.

In regard to using ThreadStart, it's just a delegate like any other, but if you are wanting to parallelize work, it's likely that you will want to provide a context or workload to each thread. It's possible to work around this a few ways, including using closures when defining the delegate but that's messy and not suitable for all situations. The Thread class, which uses ThreadStart, also supports ParameterizedThreadStart. Since you're using Invoke you can probably accept any delegate and any number of parameters, perhaps like so:

private readonly Queue<KeyValuePair<Delegate, object[]>> Queue = new Queue<KeyValuePair<Delegate, object[]>>();
private readonly object queueSync = new object();

public void AddTask<F>(F task, params object[] parameters) where F : Delegate
{
    lock (queueSync)
    {
        if (WorkingThreads.Count == ThreadsMaxCount)
        {
            Queue.Enqueue(new KeyValuePair<Delegate, object[]>(task, parameters));
        }
        else
        {
            StartThread(task, parameters);
        }
    }
}

private void StartThread(Delegate task, object[] parameters)
{
    WorkingThreads.Add(task);
    BackgroundWorker thread = new BackgroundWorker();
    thread.DoWork += delegate { task.Invoke(parameters); };
    thread.RunWorkerCompleted += delegate { ThreadCompleted(task); };
    thread.RunWorkerAsync();
}

Warnings: this code doesn't take into account my other suggestions and was written ad-hoc in the text editor here so it has not been compiled or tested; the idea is all I'm trying to demonstrate. There are several better ways to make use of generics or the delegate choice in general and you would probably also want to provide the context data in your completion event.

You should probably provide a little more information in your completion event so that the attached code can see what completed (consider one event handler can be used for many threads and even many pools). Use an object that inherits from EventArgs to contain your task (and the parameters if you take that bit of advice).

---

In regard to Darragh's answer and your comments there: if you would like to leverage the ThreadPool but need to know when work is complete you can use Task Parallel Library which now ships with 4.0 to use Task objects which by default are scheduled using the ThreadPool. Alternatively, if your tasks will be very long running and you still want to run on your own threads or specify your own concurrency regulations you can implement a Task Scheduler and continue to leverage the existing patterns/classes. This will promote usability and maintenance and if at some point you wanted to use a different scheduling mechanism it would require minimal change. It should be noted (as can be see in the documentation above) that there is already support for long running tasks not using the ThreadPool even in the default scheduler.

The following free "books" have a lot of good information, both on the new parallelism extensions to the framework (TPL, PLINQ, Parallel.For, etc.) as well as information on synchronization primitives and the raw patterns used to manage multi-threaded workloads:

Patterns for Parallel Programming: Understanding and Applying Parallel Patterns with the .NET Framework 4

Parallel Programming with Microsoft .NET

Threading in C#

Most of these contain very similar information, but each has their own style as well as different juicy nuggets of wisdom.

Answer 2

I know you have already marked one of the answers as the answer, and rightly so, it is very complete and helpful. I am adding this answer as well as no one mentioned this and it is part of your first question: 1. Do those locks make sense? Should there be other locks as well?

In regards to the second half of that, Should there be other locks as well. Yes. You are using a HashSet to keep track of working threads. You are adding to, removing from, and getting the count of this set without any locking. That is a disaster waiting to happen. You need to mutex around your reads and writes to the HashSet. As has already been mentioned, if you can use .Net 4.0 then there is a ConcurrentBag collection that you can use. Otherwise you need to add another lock object and lock around access to your HashSet.

Answer 3

We use Smart Thread Pool. It is simple and tested. You should be able to implement your requirements with less effort and more confidence by building on it rather than implementing your own.

Answer 4

In ThreadCompleted looks like you're raising OnCompleted twice - once in this method itself and another one in CheckQueue method

Answer 5

I noticed that ThreadStart is not such a good delegate even if you just want to pass a simple method, in that case the Action delegate is to be preferred because the framework can implicitly convert lambda expressions to it, which makes it more convenient to create a simple task. So one task constructor/add-task-method which takes an action as parameter would always be good.