Gracefully closing a thread

[David]

Hi All (and Peter),

I have my WIA stuff working now with info I eventually found on the net.

I have sucessfully put it into a producer/consumer thread that Peter
mentioned earlier this month. I went to one of the sites mentioned in the
message (actually, I went to both)...

http://www.yoda.arachsys.com/csharp/threads/deadlocks.shtml

About half way down is "More monitor deadlocks" which has an example of the
producer / consumer.

I am using that, but I am finding when I close my application, it is still
running...

Basically, in my consumer thread, I have a loop...

while (AppRunning)
{
object o = queue.Consume();

// Rest of my code.
}

The AppRunning bool is set to false when I close my app, but as the consumer
is holding at the queue.Consume(); the AppRunning can never be checked.

So, here is what I am thinking...

Can I make
Queue queue = new Queue();

in the ProducerConsumer a public item? If I can, then in my close, I can
check if the queue.Count is zero and if it is, I should be able to run
Thread.Abort. I guess that I will have to loop through the queue.Count ...
while (!queue.Count = 0) until it has completed.

Alternatively, is there a better way of doing this?


As I expect the consumer to take a while to clear, I want to ensure it is
empty before it is finally destroyed.

--
Best regards,
Dave Colliver.
http://www.AshfieldFOCUS.com
~~
http://www.FOCUSPortals.com - Local franchises available

 

[Peter Duniho]

[...]
Can I make
Queue queue = new Queue();

in the ProducerConsumer a public item? If I can, then in my close, I can
check if the queue.Count is zero and if it is, I should be able to run
Thread.Abort. I guess that I will have to loop through the queue.Count ...
while (!queue.Count = 0) until it has completed.

Alternatively, is there a better way of doing this?

Don't abort the thread. Even if you were careful, and only aborted the
thread while the thread calling Thread.Abort() has the lock, to ensure
no one else tries to queue more items while you're aborting the consumer
thread, you could find that you're accidentally aborting the thread
while it's processing something it's already removed from the queue
(i.e. making the count 0).

It's also just not a good habit to get into. It's too easy to create
data consistency/corruption bugs by aborting threads.

There's also the issue of exposing the queue instance, which isn't a
very good idea (it's encapsulated into the ProducerConsumer class for a
reason), but of course you could address that by just wrapping the
inspection of the count in a property or something.

Jon's sample code is really just that: sample code. I don't think he
intended it to be used as full production code, but rather just to
illustrate the basic concept. So it may be missing features useful to you.

In this case, you have at least a couple of other options that are much
better than aborting the thread:

-- Keep a flag, and wake the consumer up when you need it to check it
-- Enqueue a sentinel value (e.g. "null") that the consumer can watch for

Either should be fine. The latter is probably easier to incorporate
into your code if you've already followed Jon's example. The former is
potentially more reliable, because it provides a flag that producers can
also check and generate an error if they try to enqueue something after
shutdown has started (which may or may not be an issue in your case...if
you are only shutting down after there's no possibility of a producer
adding something, it's not an issue).

As I expect the consumer to take a while to clear, I want to ensure it is
empty before it is finally destroyed.

With either of the above suggestions, you can leave it up to the
consumer thread to terminate itself (by simply exiting when it's done).

Pete

 

[David]

Thanks, I think I have got it...

I have created a public bool property in the ProducerConsumer. This is
called Terminate and is set to false by default.

From my form closing, I set the Terminate to true, I then do
queue.Produce(null) which triggers the pulse. In my consumer, I have added
to the while statement... while (queue.Count == 0 && !Terminated) (I think I
need to change this actually, I will work on the logic.)

In my uploader code (which is what the consumer thread is handling) I am
checking for null before I do an upload (just in case).

I think what I have done is the first option you have said, though it sounds
like a combination.
--
Best regards,
Dave Colliver.
http://www.AshfieldFOCUS.com
~~
http://www.FOCUSPortals.com - Local franchises available

[...]
Can I make
Queue queue = new Queue();

in the ProducerConsumer a public item? If I can, then in my close, I can
check if the queue.Count is zero and if it is, I should be able to run
Thread.Abort. I guess that I will have to loop through the queue.Count
... while (!queue.Count = 0) until it has completed.

Alternatively, is there a better way of doing this?

Don't abort the thread. Even if you were careful, and only aborted the
thread while the thread calling Thread.Abort() has the lock, to ensure no
one else tries to queue more items while you're aborting the consumer
thread, you could find that you're accidentally aborting the thread while
it's processing something it's already removed from the queue (i.e. making
the count 0).

It's also just not a good habit to get into. It's too easy to create data
consistency/corruption bugs by aborting threads.

There's also the issue of exposing the queue instance, which isn't a very
good idea (it's encapsulated into the ProducerConsumer class for a
reason), but of course you could address that by just wrapping the
inspection of the count in a property or something.

Jon's sample code is really just that: sample code. I don't think he
intended it to be used as full production code, but rather just to
illustrate the basic concept. So it may be missing features useful to
you.

In this case, you have at least a couple of other options that are much
better than aborting the thread:

-- Keep a flag, and wake the consumer up when you need it to check it
-- Enqueue a sentinel value (e.g. "null") that the consumer can watch
for

Either should be fine. The latter is probably easier to incorporate into
your code if you've already followed Jon's example. The former is
potentially more reliable, because it provides a flag that producers can
also check and generate an error if they try to enqueue something after
shutdown has started (which may or may not be an issue in your case...if
you are only shutting down after there's no possibility of a producer
adding something, it's not an issue).

As I expect the consumer to take a while to clear, I want to ensure it is
empty before it is finally destroyed.

With either of the above suggestions, you can leave it up to the consumer
thread to terminate itself (by simply exiting when it's done).

Pete

 

[Peter Duniho]

Thanks, I think I have got it...

I have created a public bool property in the ProducerConsumer. This is
called Terminate and is set to false by default.

From my form closing, I set the Terminate to true, I then do
queue.Produce(null) which triggers the pulse. In my consumer, I have added
to the while statement... while (queue.Count == 0 && !Terminated) (I think I
need to change this actually, I will work on the logic.)

In my uploader code (which is what the consumer thread is handling) I am
checking for null before I do an upload (just in case).

I think what I have done is the first option you have said, though it sounds
like a combination.

The description of your code makes it seem like it may be overly
complicated, and perhaps also not quite correct. Did you synchronize
the assignment to the boolean property in any way (e.g. "volatile", or
an actual lock)? In Jon's code, the consumer thread is actually
implemented by the client, not the ProducerConsumer class; does your
client code correctly detect the termination condition and exit?

Using my second suggestion, the _only_ change you should need to make is
in your own consumer thread code (i.e. the code that calls Jon's
ProducerConsumer.Consume() method), check for a null return value and
exit the thread when it's null (otherwise, consume the value as normal).
You shouldn't need the sentinel _and_ a flag.

If for some reason, you see the use of a flag more desirable or
important (e.g. you need to block producers from adding to work after
shutdown has initiated), then IMHO it makes more sense to move the
consumer thread into the ProducerConsumer class itself, because of the
additional logic that is required to be in each consumer thread.

With that in mind, see below for a modification of Jon's sample that you
might find a useful example to work from if you would prefer to use my
first suggestion.

I suspect he wrote the example before C# had generics and anonymous
methods...in any case, I've modified it to take advantage of both. I've
also made some tweaks to the testing code to help illustrate better the
specific behavior of the implementation:

-- Increasing the max time for a consumer to work on an item
from 1000 ms to 2000 ms (twice the max for the producer) helps
show that the consumer may lag behind the producer without
problem.

-- Adding a second consumer helps show that, with variations
between processing time, tasks are not necessarily completed
in the same order in which they are enqueued (there is in fact
a very tiny chance they won't even be started in the same
order in which they are enqueued, but this is so dependent on
the exact thread scheduling that without a bunch of extra
synchronization code to force the issue, you'd probably almost
never see it happen).

-- Move the consumer's console output line to the end of its
"processing" rather than the beginning, to better illustrate
the above two points (especially the second).

Finally, the biggest change I made to Jon's example is to, as I
suggested above, have the ProducerConsumer class itself manage the
consumer threading, with the client code only providing an event handler
for the actual consumption of an item in the queue.

Here's the code:


using System;
using System.Collections.Generic;
using System.Threading;

public class Test
{
static void Main()
{
Random rng = new Random(0);

using (ProducerConsumer<int> queue =
new ProducerConsumer<int>())
{
queue.Consume += (i) =>
{
// Note: the call to Thread.Sleep() simply
// simulates a consumer that takes variable
// time to do some work. An actual consumer
// would not include that.
Thread.Sleep(rng.Next(2000));
Console.WriteLine("\t\t\t\tConsumed {0}", i.ToString());
};

// Call this multiple times for multiple consumers,
// or just once for only a single consumer. Creating
// multiple consumers can be very helpful if the data
// for each queue task is 100% isolated, without
// shared resources between tasks. Otherwise, be sure
// to mind the synchronization, and the use of resources
// to minimize contention (e.g. don't create more
// consumers for uploading files than you have bandwidth
// to support).
queue.StartConsumer();
queue.StartConsumer();

for (int i = 0; i < 10; i++)
{
Console.WriteLine("Producing {0}", i);
queue.Produce(i);
Thread.Sleep(rng.Next(1000));
}
}

Console.ReadLine();
}
}

public class ProducerConsumer<T> : IDisposable
{
readonly object listLock = new object();
readonly Queue<T> queue = new Queue<T>();
bool fTerminate;

public event Action<T> Consume = delegate { };

public void StartConsumer()
{
new Thread(_ConsumerThread).Start();
}

public void StopAllConsumers()
{
lock (listLock)
{
fTerminate = true;
Monitor.PulseAll(listLock);
}
}

public void Produce(T t)
{
lock (listLock)
{
if (fTerminate)
{
throw new InvalidOperationException(
"Consumers are shutting down");
}

queue.Enqueue(t);

Monitor.Pulse(listLock);
}
}

public void _ConsumerThread()
{
while (true)
{
T t = default(T);

lock (listLock)
{
while (queue.Count == 0)
{
if (fTerminate)
{
return;
}
Monitor.Wait(listLock);
}

t = queue.Dequeue();
}

Consume(t);
}
}

#region IDisposable Members

public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}

protected virtual void Dispose(bool fDisposing)
{
if (fDisposing)
{
// dispose any managed resources here.
// for now, none are present in this
// implementation
}

StopAllConsumers();
}

~ProducerConsumer()
{
Dispose(false);
}

#endregion
}

 

[David]

Right, I think I have it now...

I have put the "using" inside my button click to only start the thread when
the button is clicked.

I have made my ScanAndSaveOnePage return a string (was a void), the string
contains the file details that I want to push to the webserver. I have
modified my PushToWebServer to receive a string.

So, inside my button click is...

using (ProducerConsumer<string> queue = new ProducerConsumer<string>())
{
queue.Consume += (i) =>
{
PushToWebServer(i);
};

queue.StartConsumer();

WantsToScan = true;
while (WantsToScan)
{
queue.Produce(ScanAndSaveOnePage());
}
}

(Because my app is for scanning, the gui becomes unresponsive, which is
fine, as I don't want the users clicking other buttons to try and scan
something else whilst it is scanning)

I noted that yours was inside the Main event, which I understand is for a
console application.

I have never used => and at the moment, I don't understand the queue.Consume
+= (i) => { do action }; (actually, just laying it out like that makes it
look like an array, so this must be a generics thing, but what does =>
mean)?

This threading stuff is very frustrating, each time I apply threading, my
app breaks. I have to undo it and it is still broken, so I fix the problem
then re-apply threading. It is highlighting other issues, but it is only the
threading that is highlighting it. It seems like a whole different
discipline in programming.

--
Best regards,
Dave Colliver.
http://www.AshfieldFOCUS.com
~~
http://www.FOCUSPortals.com - Local franchises available

 

[Peter Duniho]

[...]
(Because my app is for scanning, the gui becomes unresponsive, which is
fine, as I don't want the users clicking other buttons to try and scan
something else whilst it is scanning)

IMHO, you should be thinking about fixing that in the future. For now,
it's probably fine. It seems like you have enough other stuff to worry
about. But more generally, an unresponsive UI means more than just the
inability for the user to be able to provide input and should be avoided.

The usual way to do that would be to use a different thread for the
processing, which in this case is the "using" statement and the block
within. And of course, that block winds up creating yet another thread.
You may feel that if the threading is already complicated, adding yet
another thread is just plain unreasonable. But in reality, it's the
best approach. It would be good to keep that in mind for future
improvements.

I noted that yours was inside the Main event, which I understand is for a
console application.

Your Forms-based application has a Main() _method_ too. It's not just
for console applications. But yes, for that particular example, the
significance of the location of that code is related to the fact that
it's a console application.

I have never used => and at the moment, I don't understand the queue.Consume
+= (i) => { do action }; (actually, just laying it out like that makes it
look like an array, so this must be a generics thing, but what does =>
mean)?

It's not an array nor a generics thing. The => operator is documented here:
http://msdn.microsoft.com/en-us/library/bb397687.aspx

and here:
http://msdn.microsoft.com/en-us/library/bb311046.aspx

In this context, it's a way of representing an anonymous method. Given
a ProducerConsumer instance named "queue", as in the example code, the
following are all equivalent:

void ConsumeHandler(string str)
{
// do stuff
}

void button1_Click(object sender, EventArgs e)
{
using(...)
{
queue.Consume += ConsumeHandler;
}
}

and...

void button1_Click(object sender, EventArgs e)
{
using(...)
{
queue.Consume += delegate(string str)
{
// do stuff
};
}
}

and...

void button1_Click(object sender, EventArgs e)
{
using(...)
{
queue.Consume += (str) =>
{
// do stuff
};
}
}

The latter example is the one I used, mainly for its convenience. I
like the anonymous method because it keeps all the relevant
implementation details together, and I like the lambda expression
because it's a bit shorter, especially since the compiler will infer the
type of the argument "str" for me, so I don't have to type it in the
declaration.

All three versions will work the same however.

This threading stuff is very frustrating, each time I apply threading, my
app breaks. I have to undo it and it is still broken, so I fix the problem
then re-apply threading. It is highlighting other issues, but it is only the
threading that is highlighting it. It seems like a whole different
discipline in programming.

Well, every language and framework feature is in some respects "a whole
different discipline". In that sense, yes...threading is too.

On the other hand, as you've seen, while threading is in and of itself a
bit of its own discipline, it's also very good at exposing weaknesses in
one's discipline in other areas. Sloppy coding practices that might
have been "safe enough" in a single-threaded design are revealed in
their deficiencies when one tries to combine them with multi-threaded code.

To some extent, this is the sort of thing that just requires practice.
But getting the threading right will help you get _all_ of your code
right. The problems that threading may be exposing are problems
regardless, and _all_ of your code will be better as you practice and
learn to be more disciplined in the way you write the code, even
independent of any threaded behavior.

Pete