TPL Dataflow中IO绑定任务如何维持MaxDegreeOfParallelism？

当在TPL Dataflow块中执行IO绑定（IO-bound）操作时，如何确保块维持设定的MaxDegreeOfParallelism？

场景描述 #

需向应用服务器发送10万+ Web请求并等待结果，该场景属于IO绑定（延迟为主，非CPU绑定）。使用TPL Dataflow库并将MaxDegreeOfParallelism设置为服务器可承受的请求数，但实际运行中无法保证任何时刻的任务数都达到该最大值。

通过网络捕获监测发现：任务初期速度快，能达到设定的最大并行度，但后期明显变慢，并行度下降。且TPL库无内置方法监测当前活动线程数，无法进一步排查。

官方文档说明 #

官方仅给出模糊描述：

由于MaxDegreeOfParallelism属性代表最大并行度，数据流块的实际并行度可能低于指定值。数据流块可能为满足功能需求或应对系统资源不足而降低并行度，但绝不会超过指定值。

原本期望它能识别任务因IO而非系统资源阻塞，从而在整个操作期间维持最大并行度，但实际不符。

实现代码 #

internal static ExecutionDataflowBlockOptions GetDefaultBlockOptions(int maxDegreeOfParallelism,
    CancellationToken token) => new()
    {
        MaxDegreeOfParallelism = maxDegreeOfParallelism,
        CancellationToken = token,
        SingleProducerConstrained = true,
        EnsureOrdered = false
    };


private static async ValueTask<T?> ReceiveAsync<T>(this ISourceBlock<T?> block, bool configureAwait, CancellationToken token)
{
    try
    {
        return await block.ReceiveAsync(token).ConfigureAwait(configureAwait);
    }
    catch (InvalidOperationException)
    {
        return default;
    }
}

internal static async IAsyncEnumerable<T> YieldResults<T>(this ISourceBlock<T?> block, bool configureAwait,
    [EnumeratorCancellation] CancellationToken token)
{
    while (await block.OutputAvailableAsync(token).ConfigureAwait(configureAwait))
        if (await block.ReceiveAsync(configureAwait, token).ConfigureAwait(configureAwait) is T result)
            yield return result;

    // by the time OutputAvailableAsync returns false, the block is guaranteed to be complete. However,
    // we want to await it anyway, since this will propagate any exception thrown to the consumer.
    // we don't simply await the completed task, because that wouldn't return all aggregate exceptions,
    // just the last to occur
    if (block.Completion.Exception != null)
        throw block.Completion.Exception;
}

public static IAsyncEnumerable<TResult> ParallelSelectAsync<T, TResult>(this IEnumerable<T> source, Func<T, Task<TResult?>> body,
    int maxDegreeOfParallelism = DataflowBlockOptions.Unbounded, TaskScheduler? scheduler = null, CancellationToken token = default)
{
    var options = GetDefaultBlockOptions(maxDegreeOfParallelism, token);

    if (scheduler != null)
        options.TaskScheduler = scheduler;

    var block = new TransformBlock<T, TResult?>(body, options);

    foreach (var item in source)
        block.Post(item);

    block.Complete();

    return block.YieldResults(scheduler != null && scheduler != TaskScheduler.Default, token);
}

解决方案 #

1. 限制输入缓冲区大小 #

默认情况下TransformBlock的输入缓冲区无界，一次性提交10万+任务会导致缓冲区堆积大量未处理项，干扰内部调度逻辑。修改GetDefaultBlockOptions，添加BoundedCapacity设置，建议设为MaxDegreeOfParallelism的2倍左右，避免过度堆积：

internal static ExecutionDataflowBlockOptions GetDefaultBlockOptions(int maxDegreeOfParallelism,
    CancellationToken token) => new()
{
    MaxDegreeOfParallelism = maxDegreeOfParallelism,
    CancellationToken = token,
    SingleProducerConstrained = true,
    EnsureOrdered = false,
    BoundedCapacity = maxDegreeOfParallelism * 2 // 添加缓冲区限制
};

2. 异步提交输入项 #

改用SendAsync替代Post提交输入项，当缓冲区满时异步等待，确保输入平稳进入块，避免一次性压入大量任务导致调度异常：

public static async IAsyncEnumerable<TResult> ParallelSelectAsync<T, TResult>(this IEnumerable<T> source, Func<T, Task<TResult?>> body,
    int maxDegreeOfParallelism = DataflowBlockOptions.Unbounded, TaskScheduler? scheduler = null, [EnumeratorCancellation] CancellationToken token = default)
{
    var options = GetDefaultBlockOptions(maxDegreeOfParallelism, token);

    if (scheduler != null)
        options.TaskScheduler = scheduler;

    var block = new TransformBlock<T, TResult?>(body, options);

    // 异步提交输入，缓冲区满时等待
    foreach (var item in source)
    {
        await block.SendAsync(item, token).ConfigureAwait(false);
    }

    block.Complete();

    await foreach (var result in block.YieldResults(scheduler != null && scheduler != TaskScheduler.Default, token).ConfigureAwait(false))
    {
        yield return result;
    }
}

3. 移除不必要的单生产者优化 #

SingleProducerConstrained是针对单线程生产者的优化选项，若输入项提交并非严格单线程，可移除该设置，提升调度灵活性：

internal static ExecutionDataflowBlockOptions GetDefaultBlockOptions(int maxDegreeOfParallelism,
    CancellationToken token) => new()
{
    MaxDegreeOfParallelism = maxDegreeOfParallelism,
    CancellationToken = token,
    // 移除单生产者优化
    EnsureOrdered = false,
    BoundedCapacity = maxDegreeOfParallelism * 2
};

4. 自定义并行度监测 #

TPL Dataflow无内置监测方法，可在任务委托中添加线程安全计数逻辑，跟踪当前活跃IO任务数，辅助排查问题：

// 线程安全计数器实现
public class ConcurrentCounter
{
    private int _value;
    public int Value => Volatile.Read(ref _value);
    public void Increment() => Interlocked.Increment(ref _value);
    public void Decrement() => Interlocked.Decrement(ref _value);
}

// 使用时包装原任务委托
var activeTasksCounter = new ConcurrentCounter();
var monitoredBody = async (T item) =>
{
    activeTasksCounter.Increment();
    try
    {
        return await body(item);
    }
    finally
    {
        activeTasksCounter.Decrement();
        // 定期输出当前活跃任务数
        if (activeTasksCounter.Value % 100 == 0)
        {
            Console.WriteLine($"当前活跃IO任务数：{activeTasksCounter.Value}");
        }
    }
};

// 调用ParallelSelectAsync时传入包装后的委托
await foreach (var result in source.ParallelSelectAsync(monitoredBody, maxDegreeOfParallelism, token: token))
{
    // 处理结果
}

原理说明 #

TPL Dataflow的MaxDegreeOfParallelism基于正在执行的任务数而非等待IO的任务数。IO绑定任务在await时会释放线程，理论上块应立即启动新任务填补空缺，但输入缓冲区过大、提前标记块完成等操作会干扰调度逻辑。通过限制缓冲区、异步提交输入、优化块选项，可让调度器更稳定维持最大并行度。

内容的提问来源于stack exchange，提问作者thomasrea0113