Pipe, Channel e paralelismo em C#

A motivação

Toda vez que abro um projeto novo em C# e preciso “processar um monte de coisa ao mesmo tempo”, a mesma dúvida aparece: Channel<T>? Parallel.ForEach? IAsyncEnumerable<T>? Só um Task.WhenAll já resolve? E System.IO.Pipelines, onde entra nisso tudo?

Essas ferramentas parecem intercambiáveis porque todas lidam com “fluxo de dados” ou “várias coisas acontecendo ao mesmo tempo”. Na prática, cada uma foi desenhada pra um formato de problema diferente — usar a errada geralmente ainda funciona (o código roda, os testes passam), mas o preço aparece depois: memória alocada à toa, fila crescendo sem controle, ou complexidade de coordenação que não trazia benefício nenhum.

Este case de estudo comparou as quatro ferramentas cara a cara, cada uma no cenário onde ela deveria brilhar, com números reais capturados (Stopwatch + GC.GetAllocatedBytesForCurrentThread) em vez de achismo.

As ferramentas comparadas

Ferramenta	Resolve o problema de…	Unidade de trabalho	Quando usar
System.IO.Pipelines (Pipe/PipeReader)	Ler/parsear bytes de um stream de I/O com backpressure e alocação mínima	ReadOnlySequence<byte>	Parsers de protocolo (HTTP, sockets TCP, arquivos grandes linha a linha), servidores de alto throughput
System.Threading.Channels (Channel<T>)	Coordenar produtores e consumidores in-process com tipos arbitrários	T (qualquer objeto)	Pipelines internos, múltiplos produtores/consumidores, backpressure explícito
Parallel.ForEach/Parallel.For (TPL)	Distribuir CPU de uma coleção conhecida entre núcleos	Item de uma coleção	Trabalho CPU-bound sobre dados já disponíveis em memória
IAsyncEnumerable<T> (await foreach)	Streaming assíncrono lazy de um único produtor pra um único consumidor	T produzido sob demanda	Consumir APIs paginadas, ler resultados grandes sem materializar tudo

Regra prática que resume o estudo inteiro: Pipelines é sobre bytes e I/O. Channels é sobre coordenação entre tasks com tipos arbitrários. Elas parecem primas porque as duas lidam com “fluxo de dados”, mas raramente se substituem — é comum, inclusive, usar as duas juntas (Pipelines lendo bytes crus do socket, Channels distribuindo as mensagens já desserializadas entre workers).

Pipe vs. Channel: dois tipos de backpressure

O ponto mais fácil de confundir é achar que Pipe e Channel<T> são a mesma coisa com nome diferente. Os dois têm backpressure, mas em camadas completamente distintas: um opera em bytes crus de um stream, o outro em objetos tipados entre tasks.

flowchart TB
    subgraph PIPE["System.IO.Pipelines — bytes, um produtor, um consumidor"]
        direction LR
        Sock["Socket/Stream\n(chunks fragmentados)"] --> PR["PipeReader.ReadAsync"]
        PR --> Seq{"ReadOnlySequence tem\num delimitador completo?"}
        Seq -- não --> Adv["AdvanceTo(Start, End)\npede mais bytes, zero cópia"]
        Adv --> PR
        Seq -- sim --> Slice["Slice da mensagem\nsem alocar string"]
        Slice --> Proc["Processa campos"]
    end

    subgraph CHAN["System.Threading.Channels — objetos T, N produtores, M consumidores"]
        direction LR
        P1["Producer 0..N"] -->|"WriteAsync"| CH["Channel<T>\nbounded, capacity=200"]
        CH -->|"ReadAllAsync"| C1["Consumer 0..M"]
        CH -.->|"fila cheia →\nWriteAsync espera"| P1
    end

No Pipe, o backpressure é implícito no protocolo AdvanceTo: o leitor só avança o buffer depois de processar, então nunca existe fila explícita — é sempre “leia o que já chegou, peça mais se precisar”. No Channel, o backpressure é uma escolha explícita de configuração (BoundedChannelFullMode.Wait): a fila existe de verdade, tem um tamanho, e o produtor é pausado quando ela enche.

O que os benchmarks mostraram

1 · Pipelines — parsing sem alocar string

Cenário: um “servidor” simulado entrega 500.000 mensagens de texto num stream fragmentado em pedaços de ~4 KB, como um NetworkStream real entregaria. StreamReader.ReadLineAsync (aloca uma string por linha) contra PipeReader + ReadOnlySequence<byte> (processa a linha sem materializar string).

Abordagem	Tempo (ms)	Alocado (MB)
StreamReader (naive)	31,6	97,57
PipeReader (Pipelines)	28,0	0,02

Pipelines alocou ~6.495x menos memória. O ganho de tempo é modesto — o gargalo real é achar o delimitador, não a alocação — mas o ganho de memória é o ponto: em produção, sob carga real, é a diferença entre um GC tranquilo e um GC brigando por CPU com o resto da aplicação.

2 · Channels — o backpressure muda a fila, não o tempo

Cenário: 4 produtores geram 20.000 “pedidos” no total, 8 consumidores processam cada um simulando I/O (Task.Delay de 1-3ms). Canal unbounded vs. bounded (capacidade 200) vs. o mesmo trabalho via Task.WhenAll + SemaphoreSlim.

Abordagem	Tempo (ms)	Fila máx. observada
Unbounded (sem backpressure)	38.801,0	19.986
Bounded (capacity=200, Wait)	38.718,2	200
Task.WhenAll (scatter-gather)	38.754,5	n/a

O tempo total é praticamente igual — o gargalo é a latência simulada de I/O, não a coordenação. O que muda de verdade é a fila interna: unbounded deixa ela crescer quase até o volume total (19.986 itens represados em memória); bounded trava em 200, porque WriteAsync bloqueia o produtor assim que enche. Isso é backpressure de verdade — controle de memória sob pico de carga, sem lógica manual de throttling.

3 · Parallel.ForEach — CPU pura não precisa de Channel

Cenário: contar números primos em 32 sub-intervalos de 400.000 números cada (12,8M números, CPU pura, zero I/O). Sequencial vs. Parallel.ForEach vs. a mesma carga distribuída manualmente via Channel<T> como fila de trabalho.

Abordagem	Tempo (ms)
Sequencial (for)	1.888,4
Parallel.ForEach	309,5
Via Channel<T> (fila manual)	335,5

Parallel.ForEach foi ~6,1x mais rápido que o sequencial (16 núcleos lógicos). A versão via Channel<T> chegou perto (~8% mais lenta) — mas com mais código, mais complexidade, e nenhum benefício: não havia backpressure relevante pra gerenciar nem produtores/consumidores desacoplados no tempo. É overhead de coordenação pago à toa.

4 · IAsyncEnumerable — o que importa é o primeiro item, não o total

Cenário: 200 “páginas” simuladas de 2.000 itens cada (como uma API paginada, 1ms de latência por página), consumidas com await foreach. Streaming (processa cada página ao chegar) vs. materializado (só processa depois de buscar todas as páginas).

Abordagem	Tempo até 1º item (ms)	Tempo total (ms)
Streaming (await foreach)	17,8	3.092,5
Materializado (lista completa)	3.051,1	3.054,1

O tempo total é equivalente — ambos esperam a mesma latência simulada. A diferença que importa é o tempo até o primeiro item: 17,8ms contra ~3,05 segundos. Numa UI, numa API com resposta incremental, ou num pipeline alimentando outro processo, essa diferença é “responsivo” contra “trava até terminar tudo”.

O limite do IAsyncEnumerable<T> é modelar bem só um produtor → um consumidor. Assim que o cenário vira “vários workers buscando em paralelo, alimentando um único consumidor”, yield return sequencial não dá conta — é exatamente o caso de uso de Channel<T>, com vários Writers concorrentes.

Quando usar cada uma

• Bytes chegando de um stream de I/O, com backpressure e mínima alocação → System.IO.Pipelines.
• Objetos/mensagens tipados entre tasks produtoras e consumidoras, possivelmente múltiplas de cada lado, com backpressure configurável → System.Threading.Channels.
• CPU pura sobre uma coleção conhecida em memória → Parallel.ForEach/Parallel.For.
• Streaming assíncrono simples, um produtor pra um consumidor → IAsyncEnumerable<T>.
• Disparar N operações independentes e esperar o resultado de todas, volume conhecido e finito → Task.WhenAll (com SemaphoreSlim se precisar limitar concorrência).

Escolher a ferramenta errada normalmente ainda funciona — o código roda, os testes passam — mas o preço aparece em memória (Pipelines vs. StreamReader), em controle de carga (Channel bounded vs. unbounded), ou em complexidade desnecessária (Channel vs. Parallel.ForEach pra CPU pura).

Stack

• ◆ RUNTIME: .NET 10 (build -c Release), C#.
• ⛭ FERRAMENTAS COMPARADAS: System.IO.Pipelines, System.Threading.Channels, TPL (Parallel.ForEach), IAsyncEnumerable<T>, Task.WhenAll + SemaphoreSlim.
• ▤ MEDIÇÃO: Stopwatch (tempo) + GC.GetAllocatedBytesForCurrentThread (alocação) — números reais de cada execução, não estimativa.

Como rodar os benchmarks

Cada demo é um console app independente em demos/, rodável isoladamente:

cd demos/PipeDemo;               dotnet run -c Release
cd demos/ChannelDemo;             dotnet run -c Release
cd demos/ParallelForEachDemo;     dotnet run -c Release
cd demos/AsyncEnumerableDemo;     dotnet run -c Release

Ou compilar tudo de uma vez, a partir da raiz:

dotnet build CSharpParallelismStudy.slnx -c Release

O PLAN.md do repositório documenta o racional de cada cenário escolhido — por que aquele volume de dados, por que aquela forma de medir, o que cada comparação estava tentando isolar.