17
\$\begingroup\$

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);
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ Well there is a .Net built-in class System.Threading.ThreadPool which is probably worth looking at.... msdn.microsoft.com/en-us/library/… \$\endgroup\$ – user1516 Feb 6 '11 at 14:58
  • 2
    \$\begingroup\$ I am aware of that class and i looked at it a few times but i don't quite get how to use it, also it is static which is problematic if you need multiple pools/queues. \$\endgroup\$ – H.B. Feb 6 '11 at 15:10
  • \$\begingroup\$ I added an answer with a good, imho, open source library. \$\endgroup\$ – quentin-starin Feb 9 '11 at 22:12
19
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ Oh wow, what a response, thank you very much for all the effort! Shows me once again that i have a lot left to learn, and all the links you provided will certainly help and keep me busy for a while. Thanks again! \$\endgroup\$ – H.B. Feb 9 '11 at 18:07
  • \$\begingroup\$ Not a problem and remember: we all always have a lot left to learn. Forgetting that is the first step to getting old! ;P \$\endgroup\$ – TheXenocide Feb 10 '11 at 21:55
4
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ Thank you for your input as well, i am quite a novice when it comes to threading (and many other things too sadly) so i appreciate any feedback which may help me improve :) \$\endgroup\$ – H.B. Feb 9 '11 at 22:40
  • 2
    \$\begingroup\$ Also about the marked answer, i do mot think answers like TheXenocide's are all too commonplace so for a site like this were the goal is not to solve a specific problem it sure would be nice if one could accept all the answers that bring the pieces together... \$\endgroup\$ – H.B. Feb 9 '11 at 22:45
  • \$\begingroup\$ @H.B. Glad to be of help. The only reason I caught the lack of mutex was I was looking for it. I look for it because I made a subtle error like that in production code that made it into customer hands and caused our software (which is supposed to run in a server room 24/7 unattended) to randomly dead lock. Turns out we had a collection that almost all access to it was read access and I had believed was only written to in a static initializer, well I missed a code path that caused it to be written to at a later time. When two threads collided one reading, one writing the whole thing deadlocked. \$\endgroup\$ – pstrjds Feb 10 '11 at 13:57
3
\$\begingroup\$

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.

\$\endgroup\$
  • 2
    \$\begingroup\$ Thank you for the link, i wrote my own class for two reasons, the first is that I only needed it for a small throw-away console application and the second is that i thought it'd be some interesting practice to try my hands at writing my own thread-pool. \$\endgroup\$ – H.B. Feb 9 '11 at 22:39
2
\$\begingroup\$

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

\$\endgroup\$
  • \$\begingroup\$ You are right! I added that to have the event being raised if there are no tasks at all. I should move that from CheckQueue to the constructor i think. \$\endgroup\$ – H.B. Feb 6 '11 at 14:00
  • \$\begingroup\$ No wait, that's a bad idea, of course the event needs to be attached first, Start would be a better destination. \$\endgroup\$ – H.B. Feb 6 '11 at 14:54
0
\$\begingroup\$

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.

\$\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.