I'm hoping that you kind people could cast an eye over my code and let me know what you think... Have I missed something obvious? Are there any possible race conditions? Is there an entirely different and/or better way to do this? Good and bad, everything is appreciated!
So, I have to produce and consume lots of something, however the producers are extremely slow and the consumption of the items extremely fast. slow producers, fast consumer
Thankfully my current problem can have the production of items parallelized to some extent, so I have written a class that implements IEnumerable<T>, takes a bunch of IEnumerable<T>s in it's constructor and exposes their combined output. The idea being that I am breaking the problem into parallelizable chunks, and feeding each in to be (behind-the-scenes) enumerated in parallel.
The code is probably more instructive than my babbling sentences above:
// The ability to limit the number of cached items is in case we have the unexpected
// case of producing items faster than we can consume
class ParallelProducer<T> : IEnumerable<T>
{
#region Fields
private readonly IEnumerable<T>[] enumerables;
private readonly int maxItemsToCache;
#endregion
#region Constructors
private ParallelProducer(int maxItemsToCache, IEnumerable<T>[] enumerables)
{
this.maxItemsToCache = maxItemsToCache;
this.enumerables = enumerables;
}
#endregion
#region Properties
#endregion
#region Methods
public IEnumerator<T> GetEnumerator()
{
return new ParallelProducerEnumerator<T>(this.maxItemsToCache, this.enumerables);
}
System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
{
return this.GetEnumerator();
}
#endregion
#region Nested types
private class ParallelProducerEnumerator<U> : IEnumerator<U>
{
#region Fields
private readonly List<Thread> producerThreads;
int threadsStillEnumerating;
private bool producersStarted = false;
private readonly BlockingCollection<U> cachedItems;
private U currentItem;
#endregion
#region Constructors
public ParallelProducerEnumerator(int maxItemsToCache, IEnumerable<U>[] slowEnumerables)
{
this.cachedItems = new BlockingCollection<U>(maxItemsToCache);
this.producerThreads = new List<Thread>();
this.threadsStillEnumerating = slowEnumerables.Length;
// this variable will be captured by all of the thread methods
foreach (var slowEnumerable in slowEnumerables)
{
// to avoid a reference to the iterator variable being captured
var enumerableToCapture = slowEnumerable;
var thread = new Thread(() => ProducerMethod(enumerableToCapture));
this.producerThreads.Add(thread);
}
}
#endregion
#region Properties
public U Current
{
get { return this.currentItem; }
}
object System.Collections.IEnumerator.Current
{
get { return this.Current; }
}
#endregion
#region Methods
private void ProducerMethod(IEnumerable<U> enumerable)
{
foreach (var item in enumerable)
{
this.cachedItems.Add(item);
}
int postDecrementValue = Interlocked.Decrement(ref this.threadsStillEnumerating);
if (postDecrementValue == 0)
{
cachedItems.CompleteAdding();
}
}
public bool MoveNext()
{
if (!producersStarted)
{
producersStarted = true;
foreach (var thread in producerThreads)
{
thread.Start();
}
}
if (!cachedItems.TryTake(out this.currentItem, Timeout.Infinite))
return false;
return true;
}
public void Reset()
{
throw new NotSupportedException();
}
public void Dispose()
{
}
#endregion
}
#endregion
}
And a little example usage for good measure...
// a slow producer...
static IEnumerable<int> SlowGetNumbers(int start, int end)
{
for (int i = start; i < end; i++)
{
Thread.Sleep(100);
yield return i;
}
}
// a fast consumer
static void Consume(IEnumerable<int> enumerable)
{
foreach (var item in enumerable)
{
Console.WriteLine(item);
}
}
var vanillaProducer = SlowGetNumbers(1, 30);
var parallelProducer = new ParallelProducer<int>(
SlowGetNumbers(1, 10),
SlowGetNumbers(10, 20),
SlowGetNumbers(20, 30));
// This will take 3 seconds
Consume(vanillaProducer);
// This will take ~1 second
Consume(parallelProducer);
Edit
On svick's advice, I'm just creating a IEnumerator<T>-returning method, and it's far simpler and has far less boilerplate...
static class ParallelProducer
{
public static IEnumerable<T> Create<T>(int maxItemsToCache, params IEnumerable<T>[] enumerables)
{
BlockingCollection<T> cachedItems = new BlockingCollection<T>(maxItemsToCache);
int threadsStillEnumerating = enumerables.Length;
foreach (var slowEnumerable in enumerables)
{
// to avoid a reference to the iterator variable being captured
var enumerableToCapture = slowEnumerable;
var thread = new Thread(() =>
{
foreach (var item in enumerableToCapture)
{
cachedItems.Add(item);
}
int postDecrementValue = Interlocked.Decrement(ref threadsStillEnumerating);
if (postDecrementValue == 0)
{
cachedItems.CompleteAdding();
}
});
thread.Start();
}
return cachedItems.GetConsumingEnumerable();
}
}
