Producer / Consumer implementation for Parallel Processing in the TPL

Question

I've stripped this down to its bare bones removing stopping/cancellation logic etc... to keep it simple.

The Producer is a very simple class containing a timer. At regular intervals the TimerOnElapsed will let the host now that it has another batch of items available. The onus is on the host to pull that next batch using GetNextBatch().

public class Producer
{
    public event EventHandler BatchAvailable; 
    private readonly Timer timer;
    private int i;

    public Producer()
    {
        i = 1;
        timer = new Timer(5000);
        timer.Elapsed += TimerOnElapsed;
    }

    public void Start()
    {
        timer.Enabled = true;
        timer.Start();
    }

    private void TimerOnElapsed(object sender, ElapsedEventArgs e)
    {
        if (BatchAvailable != null)
            BatchAvailable(sender , e);
    }

    public IEnumerable<int> GetNextBatch()
    {
        var range = Enumerable.Range(i, i + 50).ToList();
        i = i + 50;
        return range;
    }
}

The Consumer class uses a Concurrent.BlockingCollection<T> to pass objects to a Parallel ForEach loop. The Consumer is READY for the next batch each time it empties the blocking collection of the existing batch.

public class Consumer
{
    private TaskFactory _factory;
    private readonly BlockingCollection<int> _entries;

    public Consumer()
    {
        _entries = new BlockingCollection<int>();
    }

    public void Start()
    {

        _factory = new TaskFactory();

        try
        {
            _factory.StartNew(() =>
            {
                Parallel.ForEach(
                    _entries.GetConsumingEnumerable(),
                    new ParallelOptions() { MaxDegreeOfParallelism = 5 },
                    ProcessEntry
                    );
            });
        }
        catch (OperationCanceledException oce) { }
    }

    public void Add(int entry)
    {
        _entries.Add(entry);
    }

    public bool Ready
    {
        get { return (_entries.Count == 0); }
    }

    private void ProcessEntry(int entry)
    {
        Console.WriteLine("Processing {0}", entry);
        Thread.Sleep(3000);
    }
}

The Host is a class containing each of the above. It orchestrates communication between the two. Each time the Producer says it has another batch available, the host checks to see if the Consumer is ready, and if so, retrieves the batch and passes it on.

public class Host
{
    private Producer _producer;
    private Consumer _consumer;

    public Host()
    {
        _producer = new Producer();
        _producer.BatchAvailable += (s,e) => ProducerOnBatchAvailable();
        _consumer = new Consumer();
    }

    public void Start()
    {
        _producer.Start();
        _consumer.Start();
    }

    private void ProducerOnBatchAvailable()
    {
        if (!_consumer.Ready) return;
        Console.WriteLine("Producer is ready for another Batch...");
        var batch = _producer.GetNextBatch().ToList();
        batch.ForEach(_consumer.Add);
    }
}

I've tested this a couple of times and it behaves as I'd expect. The throttling of batch sizes and the max parallelism in the P-ForEach loop also works. And in my larger example I have a number of other Players that hand entry along to the next step in the pipeline by way of more events and blocking collections.

However, I'm a bit irked by the fact that I never need to call the BlockCollection<T>.CompletedAdding(). Is this bad practice? Do I have any potential problems leaving the blocking collection in the WAIT state for long periods of time?

In the real example the producer will be querying a DB queue which could potentially have no work in it, so the BlockingCollection could sit there for hours overnight.

Answer 1

I have never added a second answer before, but I feel it's somewhat different from the other one and a bit more complete to address your questions. I've added Stop() methods and process those as such in the both the producer, consumer and host as such:

Producer:

public sealed class Producer : IDisposable
{
    private readonly Timer timer = new Timer(5000);

    private int i = 1;

    private bool disposed;

    public event EventHandler BatchAvailable;

    public void Start()
    {
        this.timer.Elapsed += this.TimerOnElapsed;
        this.timer.Start();
    }

    public void Stop()
    {
        this.timer.Stop();
        this.timer.Elapsed -= this.TimerOnElapsed;
    }

    public void Dispose()
    {
        if (this.disposed)
        {
            return;
        }

        this.Stop();
        this.timer.Dispose();
        this.disposed = true;
    }

    public IEnumerable<int> GetNextBatch()
    {
        var range = Enumerable.Range(this.i, this.i + 50);

        this.i += 50;
        return range;
    }

    private void TimerOnElapsed(object sender, ElapsedEventArgs e)
    {
        var batchAvailable = this.BatchAvailable;

        if (batchAvailable != null)
        {
            batchAvailable(sender, e);
        }
    }
}

Consumer:

public sealed class Consumer : IDisposable
{
    private readonly BlockingCollection<int> entries = new BlockingCollection<int>();

    private readonly TaskFactory factory = new TaskFactory();

    private CancellationTokenSource tokenSource;

    private Task task;

    public void Start()
    {
        try
        {
            this.tokenSource = new CancellationTokenSource();
            this.task = this.factory.StartNew(
                () =>
                {
                    Parallel.ForEach(
                        this.entries.GetConsumingEnumerable(),
                        new ParallelOptions { MaxDegreeOfParallelism = 5, CancellationToken = tokenSource.Token },
                        (i, loopState) =>
                        {
                            if (!this.tokenSource.IsCancellationRequested)
                            {
                                ProcessEntry(i);
                            }
                            else
                            {
                                this.entries.CompleteAdding();
                                loopState.Stop();
                            }
                        });
                },
                this.tokenSource.Token);
        }
        catch (OperationCanceledException oce)
        {
            System.Diagnostics.Debug.WriteLine(oce);
        }
    }

    public void Stop()
    {
        this.Dispose();
    }

    public void Add(int entry)
    {
        this.entries.Add(entry);
    }

    public void Dispose()
    {
        if (this.task == null)
        {
            return;
        }

        this.tokenSource.Cancel();
        while (!this.task.IsCanceled)
        {
        }

        this.task.Dispose();
        this.tokenSource.Dispose();
        this.task = null;
    }

    public bool Ready
    {
        get
        {
            return this.entries.Count == 0;
        }
    }

    private static void ProcessEntry(int entry)
    {
        Console.WriteLine("Processing {0}", entry);
        Thread.Sleep(3000);
    }
}

and Host:

public sealed class Host : IDisposable
{
    private readonly Producer producer = new Producer();

    private readonly Consumer consumer = new Consumer();

    private bool disposed;

    public Host()
    {
        this.producer.BatchAvailable += (s, e) => this.ProducerOnBatchAvailable();
    }

    public void Start()
    {
        this.producer.Start();
        this.consumer.Start();
    }

    public void Stop()
    {
        this.producer.Stop();
        this.consumer.Stop();
    }

    public void Dispose()
    {
        if (this.disposed)
        {
            return;
        }

        this.Stop();
        this.producer.Dispose();
        this.consumer.Dispose();
        this.disposed = true;
    }

    private void ProducerOnBatchAvailable()
    {
        if (!this.consumer.Ready)
        {
            return;
        }

        Console.WriteLine("Producer is ready for another Batch...");

        var batch = this.producer.GetNextBatch().ToList();

        batch.ForEach(this.consumer.Add);
    }
}

Answer 2

As a matter of being thread-safe, you should replace this:

    private void TimerOnElapsed(object sender, ElapsedEventArgs e)
    {
        if (BatchAvailable != null)
            BatchAvailable(sender , e);
    }

with this:

    private void TimerOnElapsed(object sender, ElapsedEventArgs e)
    {
        var batchAvailable = this.BatchAvailable;

        if (batchAvailable != null)
        {
            batchAvailable(sender, e);
        }
    }

Reason being, accessing the event field "raw", you may wind up with it going null between the if and the invocation itself.