Pipeline Pattern e Fan-Out, Fan-In em Go: Concorrência em Estágios

Enquanto Worker Pools e Rate Limiting controlam a execução de um conjunto homogêneo de tarefas contra uma fila única, o Pipeline Pattern divide um fluxo de processamento complexo em estágios encadeados e desacoplados, conectados entre si através de canais Go (chan).

Em um pipeline idiomático:

• Cada estágio é uma função independente executada por uma ou mais goroutines;
• Cada estágio recebe dados upstream através de um canal de entrada (<-chan In);
• Executa sua transformação específica;
• Emite os dados modificados downstream através de um canal de saída (chan<- Out);
• Fecha o canal de saída quando todas as emissões forem concluídas.

Para otimizar o throughput quando um dos estágios do pipeline é computacionalmente pesado (como compressão de imagens ou hashing criptográfico), combinamos o pipeline com as técnicas de Fan-Out (paralelização) e Fan-In (multiplexação).

---

1. O Pipeline Pattern em Go

O pipeline modela o fluxo contínuo de dados em pequenos blocos isolados, permitindo que diferentes fases de uma mesma carga de trabalho rodem simultaneamente em núcleos de CPU distintos:

As 3 Responsabilidades de um Estágio:

1. Fechar o canal de saída quando todo o trabalho estiver concluído (defer close(out)).
2. Consumir o canal de entrada até que ele seja fechado pelo estágio anterior (for val := range in).
3. Respeitar cancelamentos e timeouts de context.Context em todas as tentativas de envio ou recebimento.

---

2. Fan-Out e Fan-In

Em um pipeline linear simples, se o estágio de transformação demorar 50ms por item, o estágio consumidor fica ocioso esperando. Para acelerar o processamento, multiplicamos o estágio gargalo:

• Fan-Out: Múltiplas goroutines leem concorrentemente do mesmo canal de entrada upstream para paralelizar o trabalho pesado.
• Fan-In: Uma função multiplexadora consolida as saídas de múltiplos canais independentes em um único canal downstream para o consumidor final.

---

3. O Perigo Mortal de Goroutine Leaks em Pipelines

Em Go, canais não bufferizados ou cheios bloqueiam o remetente até que outro leitor retire a mensagem.

---

4. Implementação Moderna com Go Generics

A partir do Go 1.18+, podemos construir estágios de pipeline e multiplexadores totalmente genéricos e fortemente tipados para qualquer tipo de dado ([T, R any]):

4.1 Estágio 1: O Gerador (Generate)

Converte uma coleção de dados em memória ou arquivo em um fluxo de canal:

package pipeline

import "context"

// Generate emite dados em streaming respeitando cancelamento de contexto
func Generate[T any](ctx context.Context, values ...T) <-chan T {
	out := make(chan T)

	go func() {
		defer close(out)
		for _, v := range values {
			select {
			case <-ctx.Done():
				return // Encerra sem vazar goroutine se o consumidor desistir
			case out <- v:
			}
		}
	}()

	return out
}

---

4.2 Estágio 2: O Worker de Transformação (Fan-Out)

Cada worker consome tarefas do canal de entrada e aplica uma função pura de transformação:

// Transform consome de 'in', aplica 'fn' e emite em 'out'
func Transform[T, R any](ctx context.Context, in <-chan T, fn func(T) R) <-chan R {
	out := make(chan R)

	go func() {
		defer close(out)
		for v := range in {
			result := fn(v)

			select {
			case <-ctx.Done():
				return
			case out <- result:
			}
		}
	}()

	return out
}

---

4.3 Estágio 3: O Multiplexador Genérico (Fan-In / Merge)

O Merge recebe múltiplos canais de entrada paralelos e os unifica em um canal de saída único usando sync.WaitGroup:

package pipeline

import (
	"context"
	"sync"
)

// Merge unifica múltiplos canais paralelos (Fan-In) em um canal único
func Merge[T any](ctx context.Context, channels ...<-chan T) <-chan T {
	out := make(chan T)
	var wg sync.WaitGroup

	// Multiplexador para cada canal individual
	multiplex := func(c <-chan T) {
		defer wg.Done()
		for val := range c {
			select {
			case <-ctx.Done():
				return
			case out <- val:
			}
		}
	}

	wg.Add(len(channels))
	for _, c := range channels {
		go multiplex(c)
	}

	// Fecha o canal de saída unificado somente após todos os canais de entrada terminarem
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

---

5. Exemplo Completo: Pipeline de Criptografia Paralela

Vamos processar um lote de senhas/documentos executando hashing paralelo e persistência consolidada:

package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"time"

	"meu-projeto/internal/pipeline"
)

type Document struct {
	ID      int
	Content string
}

type ProcessedDocument struct {
	ID   int
	Hash string
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// 1. Gera massa de dados de teste
	docs := make([]Document, 100)
	for i := 0; i < 100; i++ {
		docs[i] = Document{ID: i + 1, Content: fmt.Sprintf("conteudo-sensivel-%d", i+1)}
	}

	// 2. Estágio 1: Gerador de Entrada
	inStream := pipeline.Generate(ctx, docs...)

	// Função de transformação intensiva em CPU (SHA-256)
	hashDoc := func(d Document) ProcessedDocument {
		h := sha256.Sum256([]byte(d.Content))
		return ProcessedDocument{
			ID:   d.ID,
			Hash: hex.EncodeToString(h[:]),
		}
	}

	// 3. Estágio 2: Fan-Out em 4 Workers Paralelos
	workerA := pipeline.Transform(ctx, inStream, hashDoc)
	workerB := pipeline.Transform(ctx, inStream, hashDoc)
	workerC := pipeline.Transform(ctx, inStream, hashDoc)
	workerD := pipeline.Transform(ctx, inStream, hashDoc)

	// 4. Estágio 3: Fan-In unificando os 4 workers em uma única saída
	outStream := pipeline.Merge(ctx, workerA, workerB, workerC, workerD)

	// 5. Consumidor Final (Sink)
	totalProcessados := 0
	for res := range outStream {
		totalProcessados++
		fmt.Printf("Doc #%d processado com hash: %s\n", res.ID, res.Hash[:16])
	}

	fmt.Printf("Pipeline finalizado! Total de documentos: %d\n", totalProcessados)
}

---

6. Comparativo de Decisão: Pipeline vs Worker Pool

---

7. Resumo Conceitual

• O Pipeline Pattern conecta estágios independentes de processamento através de canais Go unidirecionais.
• Utilize Fan-Out quando um estágio intermediário for lento ou consumir muita CPU, dividindo o canal de entrada entre várias goroutines.
• Utilize Fan-In (Merge) para reunir as saídas de múltiplos canais concorrentes em um único canal consolidado usando sync.WaitGroup.
• Sempre utilize context.Context em todas as operações de escrita em canais para prevenir vazamentos de goroutines (Goroutine Leaks) se o consumidor interromper a leitura.
• Tire proveito dos Generics do Go ([T, R any]) para criar pipelines reutilizáveis sem sacrificar a segurança estrita de tipos em tempo de compilação.