Channels ▼


Custom Parallel Partitioning With .NET 4

Implementing an Enumerable Partitioner

Just as we did for IList<T>, we can implement a load-balancing partitioner for IEnumerable<T> that pulls one element at a time from the source and yields it from whichever partition gets there first. A full implementation is shown in Listing 2. This implementation shares a lot in common with our IList<T> partitioner, but also has a few key differences.

First, we can't use an atomically-incremented integer to progress through the data, as it's not an indexible data source and we have no notion a-priori of how many elements there will be. As can be seen in the GetEnumeratorCore method, we still employ a loop that iterates until no more elements are found, but instead of using Interlocked.Increment, we lock on a shared enumerator that all of the partitions pull from and loop until MoveNext returns False. While holding the lock, we MoveNext on the enumerator, and if it was successful, we yield the discovered element from this partition.

Second, we have to be careful (and use a fair bit of extra code) to correctly ensure that we clean up correctly when iteration through all partitions has completed. IEnumerator<T> is disposable, and thus by getting the shared enumerator from the provided enumerable, we need to ensure that we call Dispose on the shared enumerator when iteration has completed; otherwise, we could leak important resources, or at least delay their cleanup inappropriately. For static numbers of partition, we can and should clean up when all of the partitions have been disposed (i.e. when the consumer calls Dispose on all of the enumerators we hand back from GetOrderablePartitions); that accounts for the finally block in GetEnumeratorCore, where we atomically decrement a counter tracking the number of outstanding partitions, and only dispose of the shared enumerator when all of the partitions have completed. With a dynamic number of partitions, things get a bit trickier. The shared enumerator shouldn't be cleaned up until the object returned from GetOrderableDynamicPartitions (the enumerable instance) is itself disposed; otherwise, there could be a race between creating additional partitions and all partitions created thus far completing. As such, we build our own IEnumerable<T> -derived type (called DynamicGenerator in Listing 2), which participates in the reference counting on the remaining number of partitions.

public sealed class SingleItemIEnumerablePartitioner<T> : OrderablePartitioner<T>
    private readonly IEnumerable<T> _source;

    public SingleItemEnumerablePartitioner(IEnumerable<T> source) 
        : base(keysOrderedInEachPartition:true, 
        _source = source; 

    public override bool SupportsDynamicPartitions { get { return true; } }

        GetOrderablePartitions(int partitionCount)
        if (partitionCount < 1) 
            throw new ArgumentOutOfRangeException("partitionCount");
        var dynamicPartitioner = new DynamicGenerator(_source.GetEnumerator(), false);
        return (from i in Enumerable.Range(0, partitionCount) 
                select dynamicPartitioner.GetEnumerator()).ToList();

    public override IEnumerable<KeyValuePair<long, T>> GetOrderableDynamicPartitions()
        return new DynamicGenerator(_source.GetEnumerator(), true);

    private class DynamicGenerator : IEnumerable<KeyValuePair<long, T>>, IDisposable
        private readonly IEnumerator<T> _sharedEnumerator;
        private long _nextAvailablePosition;
        private int _remainingPartitions;
        private bool _disposed;

        public DynamicGenerator(IEnumerator<T> sharedEnumerator, bool requiresDisposal)
            _sharedEnumerator = sharedEnumerator;
            _nextAvailablePosition = -1;
            _remainingPartitions = requiresDisposal ? 1 : 0;

        void IDisposable.Dispose()
            if (!_disposed && Interlocked.Decrement(ref _remainingPartitions) == 0)
                _disposed = true;

        public IEnumerator<KeyValuePair<long, T>> GetEnumerator()
            Interlocked.Increment(ref _remainingPartitions);
            return GetEnumeratorCore();
        IEnumerator IEnumerable.GetEnumerator() { return GetEnumerator(); }

        private IEnumerator<KeyValuePair<long, T>> GetEnumeratorCore()
                while (true)
                    T nextItem;
                    long position;
                    lock (_sharedEnumerator)
                        if (_sharedEnumerator.MoveNext())
                            position = _nextAvailablePosition++;
                            nextItem = _sharedEnumerator.Current;
                        else yield break;
                    yield return new KeyValuePair<long,T>(position, nextItem);
            finally { if (Interlocked.Decrement(ref _remainingPartitions) == 0) 
                          _sharedEnumerator.Dispose(); }

Listing 2: A Load-Balancing Orderable Partitioner for IEnumerable<T>

Related Reading

More Insights

Currently we allow the following HTML tags in comments:

Single tags


<br> Defines a single line break

<hr> Defines a horizontal line

Matching tags


<a> Defines an anchor

<b> Defines bold text

<big> Defines big text

<blockquote> Defines a long quotation

<caption> Defines a table caption

<cite> Defines a citation

<code> Defines computer code text

<em> Defines emphasized text

<fieldset> Defines a border around elements in a form

<h1> This is heading 1

<h2> This is heading 2

<h3> This is heading 3

<h4> This is heading 4

<h5> This is heading 5

<h6> This is heading 6

<i> Defines italic text

<p> Defines a paragraph

<pre> Defines preformatted text

<q> Defines a short quotation

<samp> Defines sample computer code text

<small> Defines small text

<span> Defines a section in a document

<s> Defines strikethrough text

<strike> Defines strikethrough text

<strong> Defines strong text

<sub> Defines subscripted text

<sup> Defines superscripted text

<u> Defines underlined text

