以最大并发运行多个操作-不执行最后2个任务

c#

我创建了一个类，它允许我同时运行多个操作以及设置最大并发限制的选项。即，如果我有 100 个操作要做，并且我设置maxCurrency为 10，那么在任何给定时间，最多应该同时运行 10 个操作。最终，所有的操作都应该被执行。

这是代码：

public async Task<IReadOnlyCollection<T>> Run<T>(IEnumerable<Func<CancellationToken, Task<T>>> operations, int maxConcurrency, CancellationToken ct)
{
    using var semaphore = new SemaphoreSlim(maxConcurrency, maxConcurrency);
    var results = new ConcurrentBag<T>();
    var tasks = new List<Task>();
    foreach (var operation in operations)
    {
        await semaphore.WaitAsync(ct).ConfigureAwait(false);

        var task = Task.Factory.StartNew(async () =>
        {
            try
            {
                Debug.WriteLine($"Adding new result");
                var singleResult = await operation(ct).ConfigureAwait(false);
                results.Add(singleResult);
                Debug.WriteLine($"Added {singleResult}");
            }
            finally
            {
                semaphore.Release();
            }
        }, ct);
        tasks.Add(task);
    }
    await Task.WhenAll(tasks).ConfigureAwait(false);

    Debug.WriteLine($"Completed tasks: {tasks.Count(t => t.IsCompleted)}");
    Debug.WriteLine($"Calculated results: {results.Count}");
    
    return results.ToList().AsReadOnly();
}

这是我如何使用它的示例：

var operations = Enumerable.Range(1, 10)
    .Select<int, Func<CancellationToken, Task<int>>>(n => async ct =>
    {
        await Task.Delay(100, ct);
        return n;
    });

var data = await _sut.Run(operations, 2, CancellationToken.None);

每次执行此操作时，data集合只有 8 个结果。我希望有 10 个结果。

这是调试日志：

Adding new
Adding new
Added 1
Added 2
Adding new
Adding new
Added 3
Added 4
Adding new
Adding new
Added 5
Adding new
Added 6
Adding new
Added 7
Adding new
Added 8
Adding new
Completed tasks: 10
Calculated results: 8

如你看到的：

• 完成10个任务
• “添加新”被记录 10 次
• “添加的 x”被记录了 8 次

我不明白为什么最后两个操作没有完成。所有任务都IsComplete设置为true，据我所知，这应该意味着所有任务都被执行到最后。

回答

var task = ....
    ....
}, ct).Unwrap();

...
Added 9
Added 10
Completed tasks: 10
Calculated results: 10