Блок потока данных TPL потребляет всю доступную память

Вы, кажется, неправильно понимаете, как работает поток данных TPL.

BoundedCapacity ограничивает количество элементов, которые вы можете разместить в блоке. В вашем случае это означает один char в TransformManyBlock и один string в ActionBlock.

Таким образом, вы отправляете один элемент в TransformManyBlock, который затем возвращает 1024*1024 строк и пытается передать их ActionBlock, который принимает только одну за раз. Остальные строки просто останутся в очереди вывода TransformManyBlock.

Что вы, вероятно, захотите сделать, так это создать один блок и отправлять в него элементы в потоковом режиме, ожидая (синхронно или иным образом), когда его емкость будет достигнута:

private static void Main()
{
    MainAsync().Wait();
}

private static async Task MainAsync()
{
    var block = new ActionBlock<string>(async item =>
    {
        Console.WriteLine(item.Substring(0, 10));
        await Task.Delay(1000);
    }, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });

    foreach (var item in GetSequence('A'))
    {
        await block.SendAsync(item);
    }

    block.Complete();
    await block.Completion;
}

Кажется, что для создания ограниченного вывода TransformManyBlock необходимы три внутренних блока:

1. TransformBlock, который получает входные данные и производит IEnumerable, потенциально работающие параллельно.
2. Непараллельный ActionBlock, который перечисляет созданные IEnumerable и распространяет окончательные результаты.
3. BufferBlock, где хранятся окончательные результаты с соблюдением желаемого BoundedCapacity.

Немного сложная часть заключается в том, как распространять завершение второго блока, потому что он не связан напрямую с третьим блоком. В приведенной ниже реализации метод PropagateCompletion написан в соответствии с исходный код библиотеки.

public static IPropagatorBlock<TInput, TOutput>
    CreateOutputBoundedTransformManyBlock<TInput, TOutput>(
    Func<TInput, Task<IEnumerable<TOutput>>> transform,
    ExecutionDataflowBlockOptions dataflowBlockOptions)
{
    if (transform == null) throw new ArgumentNullException(nameof(transform));
    if (dataflowBlockOptions == null)
        throw new ArgumentNullException(nameof(dataflowBlockOptions));

    var input = new TransformBlock<TInput, IEnumerable<TOutput>>(transform,
        dataflowBlockOptions);
    var output = new BufferBlock<TOutput>(dataflowBlockOptions);
    var middle = new ActionBlock<IEnumerable<TOutput>>(async results =>
    {
        if (results == null) return;
        foreach (var result in results)
        {
            var accepted = await output.SendAsync(result).ConfigureAwait(false);
            if (!accepted) break; // If one is rejected, the rest will be rejected too
        }
    }, new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = 1,
        BoundedCapacity = dataflowBlockOptions.MaxDegreeOfParallelism,
        CancellationToken = dataflowBlockOptions.CancellationToken,
        SingleProducerConstrained = true,
    });

    input.LinkTo(middle, new DataflowLinkOptions() { PropagateCompletion = true });
    PropagateCompletion(middle, output);

    return DataflowBlock.Encapsulate(input, output);

    async void PropagateCompletion(IDataflowBlock source, IDataflowBlock target)
    {
        try
        {
            await source.Completion.ConfigureAwait(false);
        }
        catch { }

        var exception = source.Completion.IsFaulted ? source.Completion.Exception : null;
        if (exception != null) target.Fault(exception); else target.Complete();
    }
}

// Overload with synchronous delegate
public static IPropagatorBlock<TInput, TOutput>
    CreateOutputBoundedTransformManyBlock<TInput, TOutput>(
    Func<TInput, IEnumerable<TOutput>> transform,
    ExecutionDataflowBlockOptions dataflowBlockOptions)
{
    return CreateOutputBoundedTransformManyBlock<TInput, TOutput>(
        item => Task.FromResult(transform(item)), dataflowBlockOptions);
}

Пример использования:

var firstBlock = CreateOutputBoundedTransformManyBlock<char, string>(
    c => GetSequence(c), options);

Если коэффициент вывода конвейера ниже коэффициента отправки, сообщения будут накапливаться в конвейере до тех пор, пока не закончится память или не будет достигнут некоторый предел очереди. Если сообщения имеют значительный размер, процесс скоро будет нуждаться в памяти.

Установка BoundedCapacity в 1 приведет к отклонению сообщений очередью, если в очереди уже есть одно сообщение. Это нежелательное поведение, например, в таких случаях, как пакетная обработка. Прочтите этот сообщение, чтобы узнать больше.

Этот рабочий тест иллюстрирует мою точку зрения:

//Change BoundedCapacity to +1 to see it fail
[TestMethod]
public void stackOverflow()
{      
    var total = 1000;
    var processed = 0;
    var block = new ActionBlock<int>(
       (messageUnit) =>
       {
           Thread.Sleep(10);
           Trace.WriteLine($"{messageUnit}");
           processed++;
       },
        new ExecutionDataflowBlockOptions() { BoundedCapacity = -1 } 
   );

    for (int i = 0; i < total; i++)
    {
        var result = block.SendAsync(i);
        Assert.IsTrue(result.IsCompleted, $"failed for {i}");
    }

    block.Complete();
    block.Completion.Wait();

    Assert.AreEqual(total, processed);
}

Итак, мой подход заключается в том, чтобы ограничить публикацию, чтобы конвейер не накапливал много сообщений в очередях.

Ниже простой способ сделать это. Таким образом, поток данных продолжает обрабатывать сообщения на полной скорости, но сообщения не накапливаются, что позволяет избежать чрезмерного потребления памяти.

//Should be adjusted for specific use.
public void postAssync(Message message)
{

    while (totalPending = block1.InputCount + ... + blockn.InputCount> 100)
    {
        Thread.Sleep(200);
        //Note: if allocating huge quantities for of memory for each message the Garbage collector may keep up with the pace. 
        //This is the perfect place to force garbage collector to release memory.

    }
    block1.SendAssync(message)
}