Producer / Consumer implementation for Parallel Processing in the TPL 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&lt;int&gt; GetNextBatch()
    {
        var range = Enumerable.Range(i, i + 50).ToList();
        i = i + 50;
        return range;
    }
}


The Consumer class uses a Concurrent.BlockingCollection&lt;T&gt; 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&lt;int&gt; _entries;

    public Consumer()
    {
        _entries = new BlockingCollection&lt;int&gt;();
    }

    public void Start()
    {

        _factory = new TaskFactory();

        try
        {
            _factory.StartNew(() =&gt;
            {
                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) =&gt; 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&lt;T&gt;.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.
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.