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.