Parallel invocation of elements of an IEnumerable

Question

I have an IEnumerable<IEnumerable<T>> method called Batch that works like

var list = new List<int>() { 1, 2, 4, 8, 10, -4, 3 }; 
var batches = list.Batch(2); 
foreach(var batch in batches)
    Console.WriteLine(string.Join(",", batch));

-->

1,2
4,8
10,-4
3

The problem I've having is that I'm to optimize something like

foreach(var batch in batches)
    ExecuteBatch(batch);

by

Task[] tasks = batches.Select(batch => Task.Factory.StartNew(() => ExecuteBatch(batch))).ToArray();
Task.WaitAll(tasks);

or

Action[] executions = batches.Select(batch => new Action(() => ExecuteBatch(batch))).ToArray();
var options = new ParallelOptions { MaxDegreeOfParallelism = 4 };
Parallel.Invoke(options, executions);

(because ExecuteBatch is a long-running operation involving IO)

then I notice that each batch gets screwed up, is only 1 element which is default(int). Any idea what's happening or how to fix it?

Batch:

public static IEnumerable<IEnumerable<T>> Batch<T>(this IEnumerable<T> source, int size)
{
    for(var mover = source.GetEnumerator(); ;)
    {
        if(!mover.MoveNext())
            yield break;
        yield return LimitMoves(mover, size);
    }
}
private static IEnumerable<T> LimitMoves<T>(IEnumerator<T> mover, int limit)
{
    do yield return mover.Current;
    while(--limit > 0 && mover.MoveNext());
}

Answer 1

正如评论中所指出的，您的实际问题是您的实施Batch。

这段代码：

for(var mover = source.GetEnumerator(); ;)
{
    if(!mover.MoveNext())
        yield break;
    yield return LimitMoves(mover, size);
}

当Batch具体化时，该代码将不断调用，MoveNext()直到枚举耗尽。LimitMoves()使用相同的迭代器，并且被延迟调用。由于Batch耗尽了可枚举性，LimitMoves()因此永远不会发出项目。（实际上，它只会发出default(T)，因为它总是返回mover.Current，这将default(T)在枚举完成后返回）。

这是一个Batch在具体化时（因此在并行时）可以工作的实现。

public static IEnumerable<IEnumerable<T>> Batch<T>(this IEnumerable<T> source, int size)
{
    var mover = source.GetEnumerator();
    var currentSet = new List<T>();
    while (mover.MoveNext())
    {
        currentSet.Add(mover.Current);
        if (currentSet.Count >= size)
        {   
            yield return currentSet;
            currentSet = new List<T>();
        }
    }
    if (currentSet.Count > 0)
        yield return currentSet;
}

或者，您可以使用MoreLINQ - 它带有一个Batch实现。你可以在这里看到他们的实现