Concorrência Go - Pipeline

No gerador uma gorrotina escrevia no canal. No trabalhador ela lia.

E se ela fizer os dois? Lê de um canal, transforma o valor e escreve em outro.

O que é um pipeline Link para o cabeçalho

Um pipeline é uma cadeia de estágios ligados por canais. Cada estágio (stage) é uma função que recebe um canal, dispara uma gorrotina para ler dele e devolve o canal onde escreve o resultado.

Ou seja, cada estágio é trabalhador de quem vem antes e gerador para quem vem depois.

Vamos começar com um estágio que dobra o que recebe:

func dobro(entrada <-chan int) <-chan int {
	saida := make(chan int)
	go func() {
		for valor := range entrada {
			saida <- valor * 2
		}
		// quando a entrada fecha, a saída fecha também
		close(saida)
	}()
	return saida
}

Repare na assinatura. A dobro recebe um <-chan int e devolve outro <-chan int. Como a saída tem o mesmo tipo da entrada, você pode encaixar um dobro dentro do outro:

func main() {
	for valor := range dobro(dobro(sequenciaNumeros(1, 10))) {
		fmt.Printf("valor: %v\n", valor)
	}
}

A sequenciaNumeros é a mesma do post dos geradores, por enquanto sem contexto. Ela gera de 1 a 10, o primeiro dobro multiplica por dois e o segundo multiplica de novo:

$ go run .
valor: 4
valor: 8
...
valor: 40

São três gorrotinas trabalhando ao mesmo tempo, e cada canal carrega um valor por vez.

Mas quem fechou o canal da main? Link para o cabeçalho

Repare que a main não fechou canal nenhum, e mesmo assim o range dela terminou.

Quem fechou foi o gerador, depois de mandar o 10. O range do primeiro dobro terminou, ele fechou a própria saída, o range do segundo terminou, fechou a dele, e o range da main acabou.

Logo, um close na ponta de cima encerra a cadeia inteira, e cada estágio só precisa fechar o canal que ele mesmo criou.

E se eu parar de ler no meio? Link para o cabeçalho

E se, como no gerador, eu der um break no meio?

Parei logo depois do primeiro valor e contei quem ficou vivo:

for valor := range dobro(dobro(sequenciaNumeros(1, 10))) {
	if valor == 8 {
		break
	}
	fmt.Printf("valor: %v\n", valor)
}
fmt.Println("gorrotinas vivas:", runtime.NumGoroutine())
// valor: 4
// gorrotinas vivas: 4

Opa! A main é uma. As outras três são o gerador e os dois estágios, cada um parado em um envio que ninguém vai ler.

No gerador ficava uma gorrotina para trás. Aqui fica uma por estágio, e quanto mais comprido o pipeline, mais gorrotinas sobram.

A solução Link para o cabeçalho

A solução é a mesma do gerador, um context.Context. A diferença é que ele precisa atravessar a cadeia inteira:

-func dobro(entrada <-chan int) <-chan int {
+func dobro(ctx context.Context, entrada <-chan int) <-chan int {
 	saida := make(chan int)
 	go func() {
+		defer close(saida)
 		for valor := range entrada {
-			saida <- valor * 2
+			select {
+			case saida <- valor * 2:
+			case <-ctx.Done():
+				return
+			}
 		}
-		// quando a entrada fecha, a saída fecha também
-		close(saida)
 	}()
 	return saida
 }

A sequenciaNumeros volta a ser a versão com contexto do post dos geradores, e a main passa o mesmo ctx para os três:

ctx, cancelar := context.WithCancel(context.Background())
valores := dobro(ctx, dobro(ctx, sequenciaNumeros(ctx, 1, 10)))
for valor := range valores {
	if valor == 8 {
		break
	}
	fmt.Printf("valor: %v\n", valor)
}
cancelar()
// drena até o último estágio fechar
for range valores {
}
fmt.Println("pipeline encerrado")
// valor: 4
// pipeline encerrado

Um único cancelar() e as três gorrotinas vão embora. Cada uma sai pelo ctx.Done(), sem depender de o estágio seguinte voltar a ler, e fecha o canal que criou no defer.

Na verdade, como a main continua drenando, às vezes o select escolhe o envio, e o estágio acaba saindo porque a entrada dele fechou. O fim é o mesmo.

É o que o artigo Go Concurrency Patterns: Pipelines and cancellation, do Sameer Ajmani, chama de cancelamento explícito.

E quando um estágio dá erro? Link para o cabeçalho

O exemplo não trata erro, e não existe um jeito só de fazer isso. O estágio pode devolver um segundo canal só de erros e parar no primeiro, ou mandar o erro junto com o valor em uma struct e deixar quem lê decidir. Também dá para o estágio receber um canal de erros e seguir em frente.

Acho que depende de um erro dever ou não parar os outros estágios, e isto volta mais para frente na série, quando chegarmos ao semáforo.

De onde veio o exemplo Link para o cabeçalho

O código está na pasta pipeline do concorrencia-go, com o teste que confere os dez valores multiplicados por quatro.

No próximo vem o fan-out. Quando um estágio só não dá conta, vários trabalhadores leem do mesmo canal.

Lembrando que se o ctx entra na cadeia, ele entra em todos os estágios. Basta um ficar de fora para aquele estágio ficar preso no envio.

Então é isso, pessoal!

Até a próxima!

{}’s