Producer/Consumer programs

Question

I am studying mutual exclusion in college, and we just covered the producer/consumer problem. The class does not involve writing code, but I decided to implement a bounded buffer version of this problem. I have never written a multi-threaded program before, nor have I written a program with mutual exclusion before, so I decided to request a review here.

I implemented three variations, a busy-waiting variation, a Semaphore variation, and a Monitor variation. All of these reside in a class named Program, which is needed for the threading. The Monitor variation looks as if there should be a simpler solution with fewer variables. Is this so?

This is the part of the code that never changes:

const int buffSize = 10;
static char[] buffer = new char[buffSize];
static int valuesToProduce = 95;

static void Main(string[] args)
{
    Thread p = new Thread(new ThreadStart(Program.produce));
    Thread c = new Thread(new ThreadStart(Program.consume));
    p.Start();
    c.Start();
}

This is the busy-waiting producer/consumer and their related global variable:

static int avail = 0;

static void produce()
{
    for(int i=0; i<valuesToProduce; i++)
    {
        while (avail == buffSize) { };
        buffer[i % buffSize] = (char)(32 + i % 95);
        Console.WriteLine("Produced: {0}", buffer[i % buffSize]);
        avail++;
    }
}

static void consume()
{
    for (int i = 0; i < valuesToProduce; i++)
    {
        while (avail < 1) { };
        char c = buffer[i % buffSize];
        Console.WriteLine("Consumed: {0}", buffer[i % buffSize]);
        avail--;
    }
}

This is the Semaphore implementation:

private static Semaphore isFull = new Semaphore(buffSize, buffSize);
private static Semaphore isEmpty = new Semaphore(0, buffSize);

static void produce()
{
    for (int i = 0; i < valuesToProduce; i++)
    {
        isFull.WaitOne();
        buffer[i % buffSize] = (char)(32 + i % 95);
        Console.WriteLine("Produced: {0}", buffer[i % buffSize]);
        isEmpty.Release(1);
    }
}

static void consume()
{
    for (int i = 0; i < valuesToProduce; i++)
    {
        isEmpty.WaitOne();
        char c = buffer[i % buffSize];
        Console.WriteLine("Consumed: {0}", c);
        isFull.Release(1);
    }
}

And this is the Monitor implementation:

static int avail = 0;
private static object _buffer = new object();
private static object isFull = new object();
private static object isEmpty = new object();

static void produce()
{
    for (int i = 0; i < valuesToProduce; i++)
    {
        while (avail == buffSize)
        {
            Monitor.Enter(isFull);
            Monitor.Wait(isFull);
            Monitor.Exit(isFull);
        }

        Monitor.Enter(_buffer);
        buffer[i % buffSize] = (char)(32 + i % 95);
        avail++;
        Console.WriteLine("Produced: {0}", buffer[i % buffSize]);
        Monitor.Exit(_buffer);

        Monitor.Enter(isEmpty);
        Monitor.Pulse(isEmpty);
        Monitor.Exit(isEmpty);
    }
    avail++;
}

static void consume()
{
    for (int i = 0; i < valuesToProduce; i++)
    {
        while (avail < 1)
        {
            Monitor.Enter(isEmpty);
            Monitor.Wait(isEmpty);
            Monitor.Exit(isEmpty);
        }

        Monitor.Enter(_buffer);
        char c = buffer[i % buffSize];
        avail--;
        Console.WriteLine("Consumed: {0}", buffer[i % buffSize]);
        Monitor.Exit(_buffer);

        Monitor.Enter(isFull);
        Monitor.Pulse(isFull);
        Monitor.Exit(isFull);
    }
}

Answer 1

Style

Standard C# naming convention for methods is PascalCase. For static and instance members there are more variants around but often they are prefixed with _ and/or area also PascalCase so they can be easily distinguished from local variables and parameters. Following standard naming conventions makes your code look more familiar to other C# developers.

Structure

It is all static methods and global variables which seriously hurts maintenance and makes testing the code real painful as you probably have already discovered (I guess you have managed with a lot of commenting in and out code).

The easiest way to simplify this is to provide an abstract base class which the various implementations can derive from an implement. You should also pass the buffer size and values to produce/consume as parameters to break dependencies on global variables.

Something along these lines:

public class ProducerConsumerBase
{
    protected char[] _Buffer;
    protected int _TotalNumberOfValues;

    protected ProducerConsumerBase(int bufferSize, int totalNumberOfValues)
    {
        _Buffer = new char[buffSize];
        _TotalNumberOfValues = totalNumberOfValues;
    }

    public abstract void ProduceAll();
    public abstract void ConsumeAll();
}

this can then be derived like this:

public class BusyWaitingProducerConsumer : ProducerConsumerBase
{
    public BusyWaitingProducerConsumer(int bufferSize, int totalNumberOfValues)
        : base(bufferSize, totalNumberOfValues)
    {
    }

    public void ProduceAll()
    {
        ...
    }

    public void ConsumeAll()
    {
        ...
    }
}


public class SemaphoreProducerConsumer : ProducerConsumerBase
{
    public SemaphoreProducerConsumer (int bufferSize, int totalNumberOfValues)
        : base(bufferSize, totalNumberOfValues)
    {
    }

    public void ProduceAll()
    {
        ...
    }

    public void ConsumeAll()
    {
        ...
    }
}

The main method of your program then would be:

static const int BuffSize = 10;
static const int ValuesToProduce = 95;

static void Main(string[] args)
{
    var producerConsumer = new BusyWaitingProducerConsumer(BuffSize, ValuesToProduce);
    Thread p = new Thread(new ThreadStart(producerConsumer.Produce));
    Thread c = new Thread(new ThreadStart(producerConsumer.Consume));
    p.Start();
    c.Start();
}

And you can easily swap out the concrete implementation you want to test with without having to change or comment out a lot of code.

Bugs

Your busy-waiting implementation is broken in that you will need to at least call Interlocked.Increment and Interlocked.Decrement on avail because ++ and -- are not atomic operations - they consist of a read, an increment/decrement and a store. For example if avail is 1 and avail++ and avail-- are executed on two different threads simultaneously then the end result could be 0, 1 or 2.

Answer 2

static int avail = 0;

...

while (avail == buffSize) { };

...

avail++;

You can't just access the same field from multiple threads at the same time and assume it will work correctly.

As already pointed out by ChrisWue, incrementing a value is not an atomic operation, and you could fix that by using Interlocked.Increment() instead.

Another problem is that when you're reading the value of avail in a loop, the compiler is allowed to notice that the field doesn't change on this thread and optimize the repeated checks into something like:

bool condition = avail == buffSize;
while (condition) { }

This is especially problematic, because the compiler might make this optimization only in the Release build, so it the issue won't show while you're debugging.

To fix this, you would need to mark avail as volatile, which (among other things) disallows that optimization.

But all of this is pretty hard to get right. The safe way to access shared state (especially if you're just starting with multi-threaded programming) is to use a lock whenever you're accessing the shared field.

The disadvantage of this approach is that locks can be relatively slow.

---

isEmpty.Release(1);

You don't need to specify that you want to release the semaphone once, that's the default:

isEmpty.Release();

Also, consider using SemaphoreSlim instead of Semaphore. It's a newer and faster version of the same concept.

---

Monitor.Enter(isFull);
Monitor.Wait(isFull);
Monitor.Exit(isFull);

C# has a special syntax for to write the Enter-Exit pair for Monitor:

lock (isFull)
{
    Monitor.Wait(isFull);
}

The advantage of this is that it handles exceptions: if the code inside the lock throws, the lock won't stay locked, stalling your code. (Though some consider this behavior dangerous: if your code throws, it means the data can be in an invalid state and letting another thread in could cause even more damage.)

---

You should decide whether to spell out private always or never, to stay consistent.

Answer 3

Hand implementing this scenario is good education, and certainly worthwhile. That said, it's also good to realize that this is a solved problem - in particular the BlockingCollection type does many of the hard parts for you. I happened to have just put together a solution using that, so I thought I would offer it up as another alternative.

This is a simplified copy/paste from my code, so it doesn't do exactly what yours does.

//The work item queue that sits between the producer and the consumer
var blockingCollection = new BlockingCollection<int>();

var producerTask = Task.Run(() => {
    try{
        foreach(var i in Enumerable.Range(0, 10000)){
            blockingCollection.Add(i);
            Console.Out.WriteLine("Produced " + i);
        }
    } finally {
        //the try/finally is overkill in this particular case, but a good general practice to prevent hangs

        //signal the consumer that we're done
        blockingCollection.CompleteAdding(); 
    }
});

var consumerTask = Task.Run(() => {
    foreach (var i in blockingCollection.GetConsumingEnumerable()){
        Console.Out.WriteLine("Consumed " + i);
    }
});

producerTask.Wait();
consumerTask.Wait();

and here's a DotNetFiddle of it.