ThreadPool.QueueUserWorkItem and checking if finished

MrEyes

Hello all, I am currently using ThreadPool.QueueUserWorkItem to queue items onto the thread pool, the code I am using is relatively simple and looks something like this:

private static void StartQueue
{
StringReader sr = new StringReader(rawData);
string line = string.Empty;
while (((line = sr.ReadLine()) != null) && (this.allowQueuing))
{
ThreadPool.QueueUserWorkItem(new WaitCallback(DoMagicalThings), line);
}
}

private static void DoMagicalThings(object data)
{
Console.WriteLine(data.ToString())
}

What I need to work out is when the last item in the queue has completed processing. At first I came up with a fairly heath robinsion solution which was after the While in StartQueue I would add a null object onto the queue and then check this in DoMagicalThings and set a completed flag. The problem here is that in the real app DoMagicalThings does alot more than just a simple console write, so I have found that in some cases the last item added to the queue is not always the last item whose execution finishes. So the question is does anybody know how I can validate that all items I have added to the thread pool have completed

Leslie Sanford

Keep a running count of how many threads you have going. Each time you queue up a job, increment a counter. In order to wait for all of the jobs to have completed, do something like this:

lock(lockObject)
{
while(jobCount > 0)
{
Monitor.Wait(lockObject);
}
}

And in each of threads, do this:

private void ThreadJobProcedure(object data)
{
// Do stuff...

lock(lockObject)
{
    jobCount--;

    Monitor.Pulse(lockObject);
}

}

MrEyes

That looks like it would do what I need, however considering I am a bit thick I am not sure what to use as the lockObject. Using the example in my first post the code has changed to something like this:

private static void StartQueue
{
StringReader sr = new StringReader(rawData);
string line = string.Empty;
int jobCount = 0;
while (((line = sr.ReadLine()) != null) && (this.allowQueuing))
{
ThreadPool.QueueUserWorkItem(new WaitCallback(DoMagicalThings), line);
jobCount++
}

lock(lockObject)
{
while(jobCount > 0)
{
Monitor.Wait(lockObject);
}
}

Console.WriteLine("All done");
}

private static void DoMagicalThings(object data)
{
Console.WriteLine(data.ToString())

lock(lockObject)
{
jobCount--;
Monitor.Pulse(lockObject);
}
}

An additional problem is that externally to this a user could click a Stop button, this should stop all threads and stop any further items being processed. Could I somehow use this Lock and Monitor functionality to do this? In my original code, I had a simple class var bool that would be checked at the top of the worker method. If the value was false then the worker method would immediately return

Leslie Sanford

You'll need to make the jobCount variable a member of the class instead of using a local variable as you did above. If you're using static methods, the variable will need to be a static variable. As far as a lock object, what I usually do is just create a variable of type object:

private readonly object lockObject = new object();

It will need to be static if you're using it at the class level, i.e. in static methods.

MrEyes wrote:

An additional problem is that externally to this a user could click a Stop button, this should stop all threads and stop any further items being processed. Could I somehow use this Lock and Monitor functionality to do this? In my original code, I had a simple class var bool that would be checked at the top of the worker method. If the value was false then the worker method would immediately return

This works. The structure of your thread method could be something like this:

private void ThreadMethod(object data)
{
while(notStopped)
{
// Do stuff...
}

lock(lockObject)
{
    jobCount--;
    Monitor.Pulse(lockObject);
}

}

This way if the user clicks "Stop" and the notStopped variable is set to false, all of the threads break out of their loop, decrement the job counter, and signal that they're done. You may want to make the jobCount and notStopped variables volatile so that when one thread updates them, their values are immediately updated, i.e. the compiler won't try to do any optimization in delaying their update.