异步生成器，以前的迭代等待未来的迭代？

Question

我想生成一个可枚举的任务，这些任务将在不同的时间完成。

如何在 C# 中制作一个生成器：

• 产生任务
• 每隔几次迭代，就会解决以前产生的任务，结果现在才知道

我想要这样做的原因是因为我正在处理一个很长的可迭代输入，而且我经常从这些输入中积累足够的数据来发送批处理 API 请求并最终确定我的输出。

伪代码：

IEnumerable<Task<Output>> Process(IEnumerable<Input> inputs)
{
    var queuedInputs = Queue<Input>();
    var cumulativeLength = 0;
    foreach (var input in inputs)
    {
        yield return waiting task for this input
        queuedInputs.Enqueue(input);
        cumulativeLength += input.Length;
        if (cumulativeLength > 10)
        {
            cumulativeLength = 0
            GetFromAPI(queue).ContinueWith((apiTask) => {
                Queue<BatchResult> batchResults = apiTask.Result;
                while (queuedInputs.Count > 0)
                {
                    batchResult = batchResults.Dequeue();
                    historicalInput = queuedInputs.Dequeue();
                    var output = MakeOutput(historicalInput, batchResult);
                    resolve earlier input's task with this output
                }
            });
        }
    }
}

Answer 1

解决方案的形式将取决于问题的形式。我有几个问题，因为您的问题域似乎很奇怪：

1. 一开始就知道您的所有输入吗？ （同步）IEnumerable<Input> 表示它们是。
2. 您确定要在发送 any 查询之前等待一批输入吗？如果您按 10 个批处理但有 55 个输入，那么“余数”呢？

假设您确实有同步输入，并且您想对余数进行批处理，您可以立即累积所有输入，对它们进行批处理，然后遍历批处理，异步提供输出：

async IAsyncEnumerable<Output> Process(IEnumerable<Input> inputs)
{
  foreach (var batchedInput in inputs.Batch(10))
  {
    var batchResults = await GetFromAPI(batchedInput);
    for (int i = 0; i != batchedInput.Count; ++i)
      yield return MakeOutput(batchedInput[i], batchResults[i]);
  }
}

public static IEnumerable<IReadOnlyList<TSource>> Batch<TSource>(this IEnumerable<TSource> source, int size)
{
  List<TSource>? batch = null;
  foreach (var item in source)
  {
    batch ??= new List<TSource>(capacity: size);
    batch.Add(item);
    if (batch.Count == size)
    {
      yield return batch;
      batch = null;
    }
  }

  if (batch?.Count > 0)
    yield return batch;
}

更新：

如果您想立即启动 API 调用，可以将它们移出循环：

async IAsyncEnumerable<Output> Process(IEnumerable<Input> inputs)
{
  var batchedInputs = inputs.Batch(10).ToList();
  var apiCallTasks = batchedInputs.Select(GetFromAPI).ToList();
  foreach (int i = 0; i != apiCallTasks.Count; ++i)
  {
    var batchResults = await apiCallTasks[i];
    var batchedInput = batchedInputs[i];
    for (int j = 0; j != batchedInput.Count; ++j)
      yield return MakeOutput(batchedInput[j], batchResults[j]);
  }
}

Answer 2

一种方法是使用TPL Dataflow 库。该库提供了各种名为“块”的组件（TransformBlock、ActionBlock 等），其中每个块都在处理其输入数据，然后将结果传播到下一个块。这些块被链接在一起，以便管道中前一个块的完成触发下一个块的完成等，直到最后一个块通常是一个没有输出的ActionBlock<T>。这是一个例子：

var block1 = new TransformBlock<int, string>(item =>
{
    Thread.Sleep(1000); // Simulate synchronous work
    return item.ToString();
}, new()
{
    MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded,
    EnsureOrdered = false
});

var block2 = new BatchBlock<string>(batchSize: 10);

var block3 = new ActionBlock<string[]>(async batch =>
{
    await Task.Delay(1000); // Simulate asynchronous work
}); // The default MaxDegreeOfParallelism is 1

block1.LinkTo(block2, new() { PropagateCompletion = true });
block2.LinkTo(block3, new() { PropagateCompletion = true });

// Provide some input in the pipeline
block1.Post(1);
block1.Post(2);
block1.Post(3);
block1.Post(4);
block1.Post(5);

block1.Complete(); // Mark the first block as completed
await block3.Completion; // Await the completion of the last block

TPL 数据流库功能强大且灵活，但在异常传播方面存在弱点。如果block3 失败，没有内置方法可以指示block1 停止工作。您可以阅读有关此问题的更多信息here。如果您不希望您的块经常失败，这可能不是一个严重的问题。

Answer 3

假设MyGenerator()返回List<Task<T>>，并且任务数量相对较少（即使数百个也可能没问题），那么您可以使用Task.WhenAny()，它返回第一个完成的Task。然后从列表中删除Task，处理结果，然后继续下一个：

var tasks = MyGenerator();
while (tasks.Count > 0) {
    var t = Task.WhenAny(tasks);
    tasks.Remove(t);

    var result = await t; // this won't actually wait since the task is already done
    // Do something with result
}

Stephen Toub 的一篇文章对此进行了很好的讨论，该文章进行了更详细的解释，并在您的任务列表有数千个时提供了替代方案：Processing tasks as they complete

还有这篇文章，但我认为斯蒂芬的写得更好：Process asynchronous tasks as they complete (C#)

Answer 4

使用TaskCompletionSource:

IEnumerable<Task<Output>> Process(IEnumerable<Input> inputs)
{
    var tcss = new List<TaskCompletionSource<Output>>();
    var queue = new Queue<(Input, TaskCompletionSource<Output>)>();
    var cumulativeLength = 0;
    foreach (var input in inputs)
    {
        var tcs = new TaskCompletionSource<Output>();
        queue.Enqueue((input, tcs));
        tcss.Add(tcs);
        cumulativeLength += input.Length;
        if (cumulativeLength > 10)
        {
            cumulativeLength = 0
            var queueClone = Queue<(Input, TaskCompletionSource<Input>)>(queue);
            queue.Clear();
            GetFromAPI(queueClone.Select(x => x.Item1)).ContinueWith((apiTask) => {
                Queue<BatchResult> batchResults = apiTask.Result;
                while (queueClone.Count > 0)
                {
                    var batchResult = batchResults.Dequeue();
                    var (queuedInput, queuedTcs) = queueClone.Dequeue();
                    var output = MakeOutput(queuedInput, batchResult);
                    queuedTcs.SetResult(output)
                }
            });
        }
    }
    GetFromAPI(queue.Select(x => x.Item1)).ContinueWith((apiTask) => {
        Queue<BatchResult> batchResults = apiTask.Result;
        while (queue.Count > 0)
        {
            var batchResult = batchResults.Dequeue();
            var (queuedInput, queuedTcs) = queue.Dequeue();
            var output = MakeOutput(queuedInput, batchResult);
            queuedTcs.SetResult(output)
        }
    });
    foreach (var tcs in tcss)
    {
        yield return tcs.Task;
    }
}