Jon Skeet's ProducerConsumer

[Cool Guy]

I'm using a ProducerConsumer in a worker thread. If I want to end this
thread, and it's currently waiting for Consumer to return, what's the best
way to proceed?

Should I just make the thread a background thread? Or should I pass a
dummy object to ProducerConsumer.Produce to make Consume return? Or is
there a better way?

| public class ProducerConsumer
| {
| readonly object listLock = new object();
| Queue queue = new Queue();
|
| public void Produce(object o)
| {
| lock (listLock)
| {
| queue.Enqueue(o);
| if (queue.Count==1)
| {
| Monitor.Pulse(listLock);
| }
| }
| }
|
| public object Consume()
| {
| lock (listLock)
| {
| while (queue.Count==0)
| {
| Monitor.Wait(listLock);
| }
| return queue.Dequeue();
| }
| }
| }

 

[Jon Skeet [C# MVP]]

I'm using a ProducerConsumer in a worker thread. If I want to end this
thread, and it's currently waiting for Consumer to return, what's the best
way to proceed?

Should I just make the thread a background thread? Or should I pass a
dummy object to ProducerConsumer.Produce to make Consume return? Or is
there a better way?

You could put a flag in the class to say "I'm stopping" and have a Stop
method which sets the flag and pulses the monitor. The Consume method
should then check the flag before dequeuing anything (or waiting).

 

[Cool Guy]

You could put a flag in the class to say "I'm stopping" and have a Stop
method which sets the flag and pulses the monitor. The Consume method
should then check the flag before dequeuing anything (or waiting).

Something like this (added lines are prepended with '+', and removed lines
with '-')?

public class ProducerConsumer
{
readonly object listLock = new object();
Queue queue = new Queue();
+ volatile bool stopped = false;

public void Produce(object o)
{
lock (listLock)
{
queue.Enqueue(o);
if (queue.Count==1)
{
Monitor.Pulse(listLock);
}
}
}

public object Consume()
{
lock (listLock)
{
- while (queue.Count==0)
+ while (queue.Count==0 && !stopped)
{
Monitor.Wait(listLock);
}
- return queue.Dequeue();
+ return stopped ? null : queue.Dequeue();
}
}

+ public void Stop()
+ {
+ stopped = true;
+ Monitor.Pulse(listLock);
+ }
}

I believe that making stopped *volatile* should make it thread-safe here..?

 

[Jon Skeet [C# MVP]]

Something like this (added lines are prepended with '+', and removed lines
with '-')?

Yup, that looks okay. Personally I don't like using volatile,
preferring a property which takes care of the thread-safety, but that
would certainly work and would be thread-safe.

You'd need to make sure the consumer knew that null meant "the queue
might have stopped" of course. (You would probably want to expose the
stopped flag as a public property too, just in case you ever wanted to
*actually* produce a null.)

 

[Cool Guy]

Yup, that looks okay.
Right.

Personally I don't like using volatile,
preferring a property which takes care of the thread-safety, but that
would certainly work and would be thread-safe.
Right.

You'd need to make sure the consumer knew that null meant "the queue
might have stopped" of course. (You would probably want to expose the
stopped flag as a public property too, just in case you ever wanted to
*actually* produce a null.)

Right.

Thanks for your help!

 

[Cool Guy]

Something like this (added lines are prepended with '+', and removed lines
with '-')?

[snip]

Argh! I can't get it to work!

I get an exception when I try to call ProducerConsumer.Stop:

An unhandled exception of type
'System.Threading.SynchronizationLockException' occurred in
mscorlib.dll

Additional information: Object synchronization method was called from
an unsynchronized block of code.

I can't see how to get around this.

Full code:

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

public class Test
{
static ProducerConsumer queue = new ProducerConsumer();

static void Main()
{
new Thread(new ThreadStart(ConsumerJob)).Start();

queue.Produce(new object());
Thread.Sleep(2000);
queue.Stop();
}

static void ConsumerJob()
{
while (!queue.Stopped)
{
object o = queue.Consume();
Console.WriteLine("Consuming {0}.", o);
}
}
}

public class ProducerConsumer
{
readonly object listLock = new object();
Queue queue = new Queue();
volatile bool stopped = false;

public bool Stopped
{
get { return stopped; }
}

public void Produce(object o)
{
lock (listLock)
{
queue.Enqueue(o);
if (queue.Count==1)
{
Monitor.Pulse(listLock);
}
}
}

public object Consume()
{
lock (listLock)
{
while (queue.Count==0 && !stopped)
{
Monitor.Wait(listLock);
}
return stopped ? null : queue.Dequeue();
}
}

public void Stop()
{
stopped = true;
Monitor.Pulse(listLock);
}
}

 

[Jon Skeet [C# MVP]]

Something like this (added lines are prepended with '+', and removed lines
with '-')?

[snip]

Argh! I can't get it to work!

I get an exception when I try to call ProducerConsumer.Stop:

You're calling Monitor.Pulse on listLock without acquiring the monitor
first. Change it to:

lock (listLock)
{
Monitor.Pulse (listLock);
}

and it should be fine.

 

[Cool Guy]

You're calling Monitor.Pulse on listLock without acquiring the monitor
first.

Aha!