Concorrência em Go

Go é fundamentada no modelo CSP (Communicating sequential processes) proposto por Tony Hoare. Neste modelo, os dados são compartilhados enviando mensagens através de canais.

Vale uma observação histórica. No CSP original de Hoare, um processo envia mensagens diretamente para outro processo, identificado pelo nome. Erlang seguiu esse caminho. Go vem de outro ramo da família, o das linguagens Newsqueak, Alef e Limbo. Nelas o canal é um valor de primeira classe, que pode ser guardado em variáveis, passado como parâmetro e até enviado por outro canal. Os dois modelos são equivalentes, só se expressam de forma diferente. Rob Pike compara com arquivos: em Erlang é como escrever em um arquivo pelo nome, em Go é como escrever por meio de um descritor de arquivo.

Outra ideia que vai aparecer várias vezes é que concorrência não é paralelismo. Concorrência é compor computações que executam de forma independente. É uma maneira de estruturar o programa. Paralelismo é executar várias computações ao mesmo tempo. Um programa concorrente pode rodar em um único processador. E um programa bem estruturado para concorrência costuma paralelizar bem quando há mais processadores.

As explicações e exemplos são altamente inspirados na apresentação do @andrebq.

Outras influências:

• O artigo sobre _pipelines_ e cancelamento em Go.
• A palestra Go Concurrency Patterns de Rob Pike (Google I/O 2012), de onde vêm os geradores, o fan-in, os timeouts com select e o canal de parada.
• Os Go Proverbs, também de Rob Pike (Gopherfest 2015): "_Don't communicate by sharing memory, share memory by communicating_", "_Concurrency is not parallelism_", "_Channels orchestrate; mutexes serialize_" e "_Clear is better than clever_".

Os fmt.Print dentro das funções estão ali só para você enxergar o que acontece durante a execução. Eles não fazem parte dos padrões. Em código real seriam _logs_, ou nem existiriam.

Cada pasta é um programa independente, que você executa com go run ./pipeline/, por exemplo. Cada pasta tem também um exemplo_test.go com uma função Example, que confere a saída do exemplo. Quando a ordem das linhas é fixa, ela usa // Output:. Quando só o conjunto de linhas é previsível, usa // Unordered output:. Para rodar todos os testes com o detector de corrida, use go test -race ./....

Os mesmos padrões aparecem com nomes diferentes em livros, artigos e outras linguagens. Por isso cada seção traz uma linha "Também conhecido como". Dois desses nomes pedem cuidado. "Produtor" e "consumidor" são papéis, não padrões, e quase todo exemplo tem os dois. Eles aparecem como nomes alternativos de Geradores e Trabalhador porque nessas seções cada papel aparece sozinho.

📑 Sumário

• Parte 1 · Fundamentos

• Parte 2 · Padrões básicos

• Parte 3 · Encerrando goroutines

• Parte 4 · Controlando o ritmo

• Parte 5 · Padrões avançados

• Curiosidade

Parte 1 · Fundamentos

Antes dos padrões, as peças da linguagem que todos eles usam. Esta parte não apresenta nenhum padrão. Ela explica os canais, o select e a forma de esperar _goroutines_ terminarem. Se você já conhece essas peças, pode ir direto para a Parte 2.

🔗 Canais

Canais (channels) são uma estrutura primitiva na linguagem, e você pode utilizá-los para envio e recebimento de valores entre rotinas (_goroutines_). Os valores podem ser de qualquer tipo, inclusive do tipo canal.

Um canal é um ponto de sincronização entre _goroutines_. Uma _goroutine_ vai ficar bloqueada escrevendo em um canal até que aquele canal seja lido.

Ler de um canal é semelhante, uma _goroutine_ vai ficar bloqueada lendo até que um valor seja enviado para o canal ou o canal seja fechado (quando isso ocorre, o valor zero do tipo é retornado).

Um canal pode ser fechado. Isso é útil para indicar que nenhum outro valor será escrito no canal.

Ler um canal fechado retorna um valor zero do tipo do canal.

Escrever em um canal fechado causa um erro em tempo de execução (_panic_).

Fechar como sinal. Fechar um canal é a forma usual em Go de comunicar um evento que acontece uma única vez, como "terminei" ou "pode parar". Funciona para qualquer número de leitores, porque todos os que estiverem lendo são desbloqueados ao mesmo tempo. Para isso costuma-se usar um chan struct{}, que não carrega dado nenhum. O primeiro uso aparece no trabalhador.

Canais com buffer. make(chan int, 3) cria um canal que guarda até três valores. Enviar só bloqueia quando o buffer está cheio, e receber só bloqueia quando ele está vazio. O buffer tira a sincronização entre quem envia e quem recebe, e por isso pede mais cuidado. Os exemplos daqui usam canais sem buffer sempre que podem. O buffer só aparece quando ele é a própria ideia do padrão, como na contrapressão, no semáforo e no primeiro a responder.

Direção. Na assinatura de uma função, um canal pode ser declarado só para leitura (<-chan int) ou só para escrita (chan<- int). Com isso o compilador impede que a função leia do canal em que só deveria escrever, ou escreva naquele em que só deveria ler. Todos os exemplos usam tipos direcionais nas assinaturas. A única exceção é o canal quit da parada com confirmação, usado nos dois sentidos de propósito.

🗺️ Olá Mundo

O primeiro exemplo mostra como criar um canal, que será utilizado como ponte entre a aplicação principal e uma _goroutine_.

O programa principal fica bloqueado em <-canal até que a _goroutine_ envie a mensagem "Olá, mundo!".

Quando isto ocorre, o programa principal é desbloqueado, recebe a mensagem e a exibe. Como o canal não tem buffer, o envio e o recebimento acontecem juntos.

Quando o programa principal termina, a _goroutine_ é também terminada.

package main
import "fmt"
func main() {
	canal := make(chan string)
	go func() {
		canal <- "Olá, mundo!"
	}()
fmt.Println(<-canal)
}

🎛️ Select e timeouts

O select é uma estrutura de controle feita para concorrência. Ele parece um switch, mas cada case é uma comunicação, ou seja, um envio ou um recebimento em um canal. O select bloqueia até que alguma das comunicações possa prosseguir. Se várias puderem ao mesmo tempo, ele escolhe uma de forma pseudoaleatória. Se houver um default, ele não bloqueia. Na palestra Go Concurrency Patterns, Rob Pike diz que o select é a razão de canais e _goroutines_ fazerem parte da linguagem, em vez de serem uma biblioteca.

Quase todos os padrões das próximas partes dependem dele. Aqui vemos o uso mais comum, que é dar um prazo a uma comunicação. Os dois exemplos usam o mesmo gerador, tagarela, que fala cada vez mais devagar:

• Timeout por mensagem. time.After devolve um canal que recebe um valor depois do tempo indicado. Colocado em um case dentro do laço, ele é recriado a cada volta. O prazo vale para cada mensagem e é renovado sempre que uma chega.
• Timeout para a conversa inteira. É o mesmo time.After, mas criado uma única vez, fora do laço. O prazo vale para a conversa toda, não importa quantas mensagens cheguem.

O tagarela recebe um canal quit, que a função fecha ao sair para que o gerador também termine. Esse é o canal de parada, assunto da Parte 3.

package main
import (
	"fmt"
	"time"
)
// tagarela é um gerador que fala cada vez mais devagar: a pausa entre as
// mensagens cresce 100ms a cada envio. Ele só para quando o canal quit é
// fechado (veja o exemplo canal_de_parada).
func tagarela(nome string, quit <-chan struct{}) <-chan string {
	saida := make(chan string)
	go func() {
		defer close(saida)
		for i := 0; ; i++ {
			// O envio disputa com o sinal de parada: o que puder
			// prosseguir primeiro, vence.
			select {
			case saida <- fmt.Sprintf("%s %d", nome, i):
				time.Sleep(time.Duration(i)  100  time.Millisecond)
			case <-quit:
				return
			}
		}
	}()
	return saida
}
// timeoutPorMensagem desiste quando UMA mensagem demora mais do que 350ms.
// O time.After é criado a cada volta do laço, então o prazo é renovado
// sempre que uma mensagem chega.
func timeoutPorMensagem() {
	quit := make(chan struct{})
	defer close(quit)
	c := tagarela("Ana", quit)
for {
		select {
		case s := <-c:
			fmt.Println(s)
		case <-time.After(350 * time.Millisecond):
			fmt.Println("Ana demorou demais para falar.")
			return
		}
	}
}
// timeoutDaConversa limita a duração da conversa INTEIRA a 500ms.
// O time.After é criado uma única vez, fora do laço: o prazo não é renovado.
func timeoutDaConversa() {
	quit := make(chan struct{})
	defer close(quit)
	c := tagarela("Beto", quit)
timeout := time.After(500 * time.Millisecond)
	for {
		select {
		case s := <-c:
			fmt.Println(s)
		case <-timeout:
			fmt.Println("A conversa com o Beto acabou.")
			return
		}
	}
}
func main() {
	timeoutPorMensagem()
	timeoutDaConversa()
}

⏳ Esperando goroutines (WaitGroup)

No Olá Mundo, o programa principal esperou a _goroutine_ lendo de um canal. Quando o que se quer é só esperar várias _goroutines_ terminarem, sem receber nenhum dado delas, a ferramenta certa é o sync.WaitGroup.

São duas chamadas. O wg.Go dispara a função em uma nova _goroutine_ e registra no WaitGroup que ela precisa terminar. O wg.Wait() bloqueia até que todas as _goroutines_ disparadas assim terminem.

No exemplo, três tarefas com durações diferentes são disparadas de uma vez. A função principal só imprime a última linha depois que as três terminaram. Experimente comentar o wg.Wait() e veja o programa acabar antes de qualquer tarefa imprimir.

O wg.Go existe desde o Go 1.25. Em código mais antigo você vai encontrar a forma equivalente, com wg.Add(1) antes de cada go e defer wg.Done() dentro da _goroutine_. O wg.Go faz as duas coisas de uma vez e evita os erros clássicos dessa forma, como esquecer o Done ou chamar o Add dentro da _goroutine_.

Por que um WaitGroup e não um canal? Canais servem para orquestrar o fluxo de dados entre _goroutines_. Contar quantas _goroutines_ já terminaram é um problema menor, e para esses Rob Pike recomenda o pacote sync. Na palestra Go Concurrency Patterns ele avisa "_Don't overdo it_", porque às vezes só é preciso um contador, e pede "_Always use the right tool for the job_". Nos Go Proverbs a mesma ideia aparece como "_Channels orchestrate; mutexes serialize_". Por isso, nos padrões a seguir, os canais transportam os dados e o WaitGroup apenas conta quem terminou.

package main
import (
	"fmt"
	"sync"
	"time"
)
// tarefa simula um trabalho que leva algum tempo.
func tarefa(id int) {
	time.Sleep(time.Duration(id)  50  time.Millisecond)
	fmt.Printf("tarefa %d terminou\n", id)
}
func main() {
	var wg sync.WaitGroup
// wg.Go dispara a função em uma nova goroutine e registra no
	// WaitGroup que ela precisa terminar.
	for i := range 3 {
		wg.Go(func() {
			tarefa(i + 1)
		})
	}
// Bloqueia até que todas as goroutines disparadas com wg.Go terminem.
	// Sem esta linha o programa acabaria antes de as tarefas imprimirem.
	wg.Wait()
	fmt.Println("todas as tarefas terminaram")
}

É possível fazer só com canais. Um canal com buffer de tamanho n e um laço que lê n vezes fazem o mesmo papel. Cada tarefa envia um sinal ao terminar, e a função principal espera receber todos. Funciona, mas é um WaitGroup refeito à mão.

>

> terminar := make(chan struct{}, 3)

Parte 2 · Padrões básicos

Os blocos de montar. Cada padrão desta parte faz uma coisa só e usa apenas o que foi visto nos fundamentos. Os padrões das partes seguintes são combinações e variações destes.

🆕 Geradores

Também conhecido como: produtor, _source_. É o mesmo papel do produtor da seção de contrapressão.

Geradores são funções que iniciam uma _goroutine_ para escrever uma lista de valores em um canal que é retornado para quem acionou a função.

No exemplo, uma sequência de números inteiros é gerada e enviada para um canal.

A função sequenciaNumeros reaparece em vários exemplos. Ela é copiada de propósito, para que cada arquivo possa ser lido e executado sozinho. Como diz um dos Go Proverbs, "_a little copying is better than a little dependency_".

A função principal (_main_) irá realizar a leitura do canal e imprimir os valores. Essa é uma característica interessante sobre canais, quando utilizados com o _range_, a iteração continuará até que o canal seja fechado.

Atenção: este gerador não é cancelável. Se o consumidor parar de ler antes do fim, a _goroutine_ vaza. Veja a Parte 3.

package main
import "fmt"
func sequenciaNumeros(inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		for i := inicial; i <= final; i++ {
			saida <- i
		}
		// após gerar todos os valores, fecha o canal
		close(saida)
	}()
	return saida
}
func main() {
	valores := sequenciaNumeros(1, 1000)
	for valor := range valores {
		fmt.Printf("valor: %v\n", valor)
	}
}

🚧 Trabalhador (worker)

Também conhecido como: consumidor, _sink_. O segundo só vale quando o trabalhador é o último estágio, isto é, quando não repassa nada adiante.

Um trabalhador é uma _goroutine_ que recebe valores de um canal e os processa.

No exemplo, valores inteiros são enviados pela função principal (main) através do canal de entrada e processados por um trabalhador.

É possível criar vários trabalhadores para processarem um mesmo canal.

Repare que o término é sinalizado com close(pronto), e não com o envio de um valor. É o fechamento usado como sinal, visto em canais.

package main
import "fmt"
func trabalhador(entrada <-chan int) {
	for valor := range entrada {
		fmt.Printf("valor: %v\n", valor)
	}
}
func main() {
	entrada := make(chan int)
	pronto := make(chan struct{})
	// Um trabalhador é iniciado e aguarda por valores no canal de entrada
	go func() {
		trabalhador(entrada)
		// Fechar o canal é o idioma para sinalizar um evento único:
		// comunica "terminou" a qualquer número de leitores.
		close(pronto)
	}()
	for i := range 10 {
		entrada <- i
	}
	// Após ter enviado todos os valores, fecha o canal de entrada
	// avisando ao trabalhador que o trabalho terminou
	close(entrada)
	// Aguarda o trabalhador terminar
	<-pronto
}

🧑‍🏭 Pipeline

Também conhecido como: cadeia de estágios. Cada função do _pipeline_ é um _estágio_ (_stage_).

Um _pipeline_ trabalha recebendo valores de um canal e escrevendo em outro canal, normalmente após realizar alguma transformação no valor.

No exemplo temos a função dobro atuando como um _pipeline_, que irá receber os valores enviados ao canal de entrada retornando os valores transformados.

Repare nas assinaturas. A função dobro recebe um <-chan int e devolve outro <-chan int. São os tipos direcionais vistos em canais, e aqui eles deixam claro quem lê e quem escreve em cada estágio.

Os valores gerados por sequenciaNumeros são enviados para o canal de entrada do _pipeline_. A função principal recebe os valores transformados pelo canal de saída e os imprime.

Vários pipelines poderiam ser encadeados para realizar múltiplas transformações.

Atenção: o gerador e as etapas deste _pipeline_ não são canceláveis. Se o consumidor parar de ler antes do fim, a _goroutine_ vaza. Veja a Parte 3.

package main
import "fmt"
func dobro(entrada <-chan int) <-chan int {
	saida := make(chan int)
	go func() {
		for valor := range entrada {
			saida <- valor * 2
		}
		// Após ter terminado de transformar os valores de entrada,
		//  fecha o canal de saida
		close(saida)
	}()
	return saida
}
func sequenciaNumeros(inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		for i := inicial; i <= final; i++ {
			saida <- i
		}
		// após gerar todos os valores, fecha o canal
		close(saida)
	}()
	return saida
}
func main() {
	for valor := range dobro(dobro(sequenciaNumeros(1, 10))) {
		fmt.Printf("valor: %v\n", valor)
	}
}

📣 Fan-out

Também conhecido como: distribuição, _work distribution_.

Um fan-out distribui os valores de um canal de entrada entre várias _goroutines_. O artigo sobre _pipelines_ define assim: múltiplas funções lendo do mesmo canal até que ele seja fechado. Cada valor é processado por exatamente uma delas, o que permite dividir um trabalho demorado entre vários trabalhadores.

Não é preciso nenhum código para decidir quem recebe o quê, porque o próprio canal faz a distribuição. Quando várias _goroutines_ estão bloqueadas lendo o mesmo canal, cada envio é entregue a apenas uma delas.

No exemplo, três trabalhadores dividem entre si os dez valores gerados por sequenciaNumeros. Repare na saída que nenhum valor aparece duas vezes. Um sync.WaitGroup aguarda o término de todos. O grupo de trabalhadores, mais adiante, é uma aplicação deste padrão.

Execute o exemplo mais de uma vez e veja que a ordem da saída muda. Os trabalhadores concorrem pelos valores da entrada, e quem decide qual deles roda a cada momento é o escalonador. Esta é a primeira vez que o não determinismo aparece por aqui. Ele tem a ver com a ideia de que concorrência não é paralelismo: o programa descreve computações independentes, mas não diz em que ordem elas executam. Por isso um programa concorrente correto não pode depender dessa ordem.

Não confunda com o tee, em que cada valor é copiado para todos os consumidores.

package main
import (
	"fmt"
	"sync"
	"time"
)
// trabalhador lê do canal de entrada, que é compartilhado com os demais
// trabalhadores. Cada valor é entregue a exatamente um deles: quem estiver
// livre primeiro, recebe.
func trabalhador(id int, entrada <-chan int) {
	for valor := range entrada {
		fmt.Printf("id: %d processando valor: %v\n", id, valor)
		// Simula um processamento demorado
		time.Sleep(100 * time.Millisecond)
	}
}
// fanout distribui os valores de um único canal de entrada entre n
// trabalhadores e só retorna quando todos terminarem.
func fanout(entrada <-chan int, n int) {
	var wg sync.WaitGroup
for i := range n {
		wg.Go(func() {
			trabalhador(i+1, entrada)
		})
	}
	wg.Wait()
}
func sequenciaNumeros(inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		for i := inicial; i <= final; i++ {
			saida <- i
		}
		// após gerar todos os valores, fecha o canal
		close(saida)
	}()
	return saida
}
func main() {
	// Três trabalhadores dividem entre si os dez valores da sequência
	fanout(sequenciaNumeros(1, 10), 3)
}

🔀 Tee (broadcast)

Também conhecido como: _broadcast_, _publish/subscribe_ em memória. O segundo é aproximado: em um _pub/sub_ os assinantes costumam entrar e sair dinamicamente, enquanto o tee tem um conjunto fixo de saídas.

Um tee copia cada valor de um canal de entrada para todos os canais de saída, de modo que todos os consumidores veem todos os valores. O nome vem do comando tee do Unix, que duplica o que recebe. É o oposto do fan-out, em que cada valor vai para um único consumidor.

No exemplo, uma sequência de números é gerada e copiada para múltiplos canais de saída. Estes canais possuem seus respectivos trabalhadores que irão fazer o processamento do valor.

O tee lê cada valor da entrada e o envia, em sequência, para cada uma das saídas. Quando a entrada é fechada, ele fecha todas as saídas. Para aguardar o término dos trabalhadores, a função principal usa um sync.WaitGroup.

Como os canais não têm buffer, o tee só passa para o próximo valor depois que todas as saídas receberam o atual. A consequência é que um consumidor lento atrasa todos os outros, e também o produtor. É a contrapressão aplicada ao broadcast. Ninguém perde mensagem, mas todos andam no ritmo do mais lento.

package main
import (
	"fmt"
	"sync"
	"time"
)
// tee copia cada valor da entrada para todas as saídas: todos os consumidores
// veem todos os valores. O envio é sequencial e sem buffer, então o tee só
// avança quando todas as saídas receberam o valor: um consumidor lento
// atrasa todos os outros.
func tee(entrada <-chan int, saidas ...chan<- int) {
	for valor := range entrada {
		for _, saida := range saidas {
			saida <- valor
		}
	}
	// Como a entrada foi consumida, fecha os canais de saída
	for _, saida := range saidas {
		close(saida)
	}
}
func sequenciaNumeros(inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		for i := inicial; i <= final; i++ {
			saida <- i
		}
		// após gerar todos os valores, fecha o canal
		close(saida)
	}()
	return saida
}
// trabalhador consome os valores de uma das saídas do tee. O parâmetro
// demora simula o tempo de processamento de cada valor.
func trabalhador(id int, entrada <-chan int, demora time.Duration) {
	for valor := range entrada {
		fmt.Println("id: ", id, " valor: ", valor)
		time.Sleep(demora)
	}
}
func main() {
	saida1 := make(chan int)
	saida2 := make(chan int)
// Aguarda o término dos trabalhadores
	var wg sync.WaitGroup
	wg.Go(func() { trabalhador(1, saida1, 0) })
	wg.Go(func() { trabalhador(2, saida2, 0) })
// Copia a sequência de números para todos os canais de saída
	tee(sequenciaNumeros(1, 10), saida1, saida2)
	wg.Wait()
// Tee com timeout (veja tee_timeout.go): agora o trabalhador 2 é mais lento
	// do que o timeout, então parte dos valores destinados a ele é descartada.
	saida1 = make(chan int)
	saida2 = make(chan int)
wg.Go(func() { trabalhador(1, saida1, 0) })
	wg.Go(func() { trabalhador(2, saida2, 250*time.Millisecond) })
teeComTimeout(sequenciaNumeros(1, 5), 100*time.Millisecond, saida1, saida2)
	wg.Wait()
}

Tee com timeout

Se um consumidor lento não pode segurar os demais, uma alternativa é desistir do envio depois de um tempo. Nesta variante, cada envio é feito dentro de um select que disputa com time.After, e vence o que acontecer primeiro. Se o tempo esgotar, o valor é descartado apenas para aquela saída e o tee segue em frente. Um select por saída dentro do laço é suficiente, não é preciso criar uma _goroutine_ para cada envio.

Descartar mensagens é uma decisão de projeto, e não parte do padrão. Com o descarte, o consumidor lento deixa de ver todos os valores, que era justamente a garantia do tee. Por isso o exemplo avisa na saída cada vez que descarta, em vez de descartar em silêncio. Repare também que o timeout limita o atraso, mas não acaba com ele, pois cada valor ainda pode esperar até timeout em cada saída lenta. Outras formas de lidar com um consumidor lento aparecem na janela deslizante e na contrapressão.

No exemplo, a função principal executa as duas versões. Na segunda, o trabalhador 2 leva 250ms por valor e o timeout é de 100ms, então parte dos valores destinados a ele é descartada.

package main
import (
	"fmt"
	"time"
)
// teeComTimeout é um tee que não espera indefinidamente por um consumidor
// lento: se uma saída não receber o valor dentro de timeout, o valor é
// descartado para aquela saída e o tee segue em frente.
// Descartar mensagens é uma decisão de projeto, não parte do padrão.
func teeComTimeout(entrada <-chan int, timeout time.Duration, saidas ...chan<- int) {
	for valor := range entrada {
		for i, saida := range saidas {
			// Um select por saída: o que acontecer primeiro, o envio ou o timeout
			select {
			case saida <- valor:
			case <-time.After(timeout):
				// Sinalizamos o descarte explicitamente para não perder
				// a informação silenciosamente.
				fmt.Printf("tee: descarte por timeout, saida=%d valor=%d\n", i+1, valor)
			}
		}
	}
	// Como a entrada foi consumida, fecha os canais de saída
	for _, saida := range saidas {
		close(saida)
	}
}

⚗️ Fan-in

Também conhecido como: _merge_, multiplexação (o termo que Rob Pike usa na palestra de 2012).

Um fan-in copia dados de múltiplos canais de entrada e escreve em um único canal de saída. Normalmente um fan-in só termina quando todos os canais de entrada são fechados.

A função fan-in pode receber vários canais de entrada através de parâmetros múltiplos.

No exemplo abaixo, enviamos vários geradores como entrada para a função fan-in e nos é retornado um único canal de saída. Internamente, uma _goroutine_ é criada para ler os valores de cada canal de entrada, porém todas escrevem no mesmo canal de saída.

Envio de mensagem em um canal fechado causa um erro (_panic_), por isso é importante garantir que todos os canais de entrada estejam fechados antes de fechar o canal de saída. Utilizamos um sync.WaitGroup para saber quando todos os canais de entrada foram processados.

Repare que temos uma _goroutine_ que aguarda em wg.Wait() até que todas as entradas sejam consumidas, finalizando assim o canal de saída.

Atenção: estes geradores não são canceláveis. Se o consumidor parar de ler antes do fim, a _goroutine_ vaza. Veja a Parte 3.

package main
import (
	"fmt"
	"sync"
)
// fanin combina vários canais de entrada em um único canal de saída.
// Utiliza um WaitGroup para saber quando todos os canais de entrada foram processados.
func fanin(entradas ...<-chan int) <-chan int {
	saida := make(chan int)
	var wg sync.WaitGroup
for _, entrada := range entradas {
		// Uma goroutine por entrada; o WaitGroup é avisado quando ela termina
		wg.Go(func() {
			for valor := range entrada {
				saida <- valor
			}
		})
	}
// Quando todos os canais de entrada terminarem, fecha o canal de saída
	go func() {
		wg.Wait()
		close(saida)
	}()
return saida
}
// sequenciaNumeros cria um canal que envia uma sequência de números de inicial a final.
func sequenciaNumeros(inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		for i := inicial; i <= final; i++ {
			saida <- i
		}
		close(saida)
	}()
	return saida
}
func main() {
	// Combina três canais de sequência em um único canal
	canal := fanin(
		sequenciaNumeros(1, 10),
		sequenciaNumeros(11, 20),
		sequenciaNumeros(21, 30),
	)
// Lê e imprime os valores do canal combinado
	for valor := range canal {
		fmt.Printf("valor: %v\n", valor)
	}
// Com um número fixo de entradas, uma única goroutine com select basta
	// (veja fan_in_select.go)
	canal = faninSelect(
		sequenciaNumeros(31, 40),
		sequenciaNumeros(41, 50),
	)
	for valor := range canal {
		fmt.Printf("valor (select): %v\n", valor)
	}
}

Fan-in com uma _goroutine_ e select

Quando o número de entradas é fixo e conhecido, Rob Pike mostra na palestra Go Concurrency Patterns uma variante mais enxuta. Uma única _goroutine_ com um select repassa para a saída o valor da entrada que estiver pronta primeiro.

A versão da palestra roda para sempre. Aqui ela também trata o fechamento das entradas, com o truque do canal nil visto em select. Quando uma entrada é fechada, a variável vira nil e aquele case deixa de ser escolhido. Quando todas viram nil, o laço termina e a saída é fechada. Como só uma _goroutine_ escreve na saída, ela mesma fecha o canal, sem WaitGroup.

Quando usar cada uma? Se o número de canais é variável, como em um _slice_ ou em parâmetros múltiplos, use uma _goroutine_ por entrada, porque um select tem um número fixo de cases escrito no código. Se o número é fixo e pequeno, o select é mais direto. Basta uma _goroutine_, e não é preciso contar quem terminou.

package main
// faninSelect combina um número fixo de canais de entrada (aqui, dois) usando
// uma única goroutine e um select, em vez de uma goroutine por entrada.
// Como só uma goroutine escreve na saída, ela mesma fecha o canal ao terminar:
// não é preciso contar ninguém.
func faninSelect(entrada1, entrada2 <-chan int) <-chan int {
	saida := make(chan int)
	go func() {
		defer close(saida)
		for entrada1 != nil || entrada2 != nil {
			select {
			case valor, ok := <-entrada1:
				if !ok {
					// Entrada fechada: um canal nil nunca é selecionado,
					// o que desabilita este case.
					entrada1 = nil
					continue
				}
				saida <- valor
			case valor, ok := <-entrada2:
				if !ok {
					entrada2 = nil
					continue
				}
				saida <- valor
			}
		}
	}()
	return saida
}

👷‍♂️👷‍♀️ Grupo de Trabalhadores (pool of workers)

Também conhecido como: _worker pool_, _pool_ de _goroutines_.

A piscina de marmotinhas (carinhosamente chamada pela minha esposa) é uma coleção de _goroutines_ que ficam esperando tarefas serem atribuídas a elas. Quando a _goroutine_ finaliza a tarefa que foi atribuída, se torna disponível novamente para execução de uma nova tarefa.

No exemplo, um grupo de n trabalhadores aguarda a chegada de valores pelo canal de entrada. Cada trabalhador executa seu processamento e envia o resultado por um canal.

O grupo de trabalhadores é uma aplicação de fan-out. Várias _goroutines_ leem do mesmo canal de entrada, e cada valor é processado por uma só delas. Além de distribuir o trabalho, o grupo cuida dos trabalhadores, que voltam a ficar disponíveis ao terminar uma tarefa, e junta os resultados em um canal de saída.

O grupo fixa quantas _goroutines_ existem. Se a ideia for ter uma _goroutine_ por tarefa e limitar apenas quantas executam ao mesmo tempo, veja o semáforo.

Como no fan-out, a ordem da saída muda a cada execução.

Os trabalhadores são iniciados com wg.Go, e uma outra _goroutine_ aguarda em wg.Wait() para então fechar o canal de saída, como visto em esperando goroutines. Repare que o trabalhador nem sabe que o WaitGroup existe. Ele só processa valores, e quem o dispara é que cuida de esperar.

package main
import (
	"fmt"
	"sync"
)
// trabalhador processa valores recebidos do canal de entrada e envia resultados para o canal de saída.
func trabalhador(id int, entrada <-chan int, saida chan<- int) {
	for valor := range entrada {
		fmt.Printf("id: %d processou valor: %v\n", id, valor)
		saida <- valor * 2
	}
fmt.Printf("id: %d terminou\n", id)
}
func grupoDeTrabalhadores(entrada <-chan int, nTrabalhadores int) <-chan int {
	saida := make(chan int)
	// Os canais transportam os dados; o WaitGroup apenas conta
	// quantos trabalhadores ainda não terminaram.
	var wg sync.WaitGroup
// Cria e inicia os trabalhadores. wg.Go dispara a função em uma nova
	// goroutine e registra no WaitGroup que ela precisa terminar.
	for i := range nTrabalhadores {
		wg.Go(func() {
			trabalhador(i+1, entrada, saida)
		})
	}
// Goroutine para fechar o canal de saída quando todos os trabalhadores terminarem
	go func() {
		wg.Wait()
		close(saida)
	}()
return saida
}
func sequenciaNumeros(inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		for i := inicial; i <= final; i++ {
			saida <- i
		}
		// Após gerar todos os valores, fecha o canal
		close(saida)
	}()
	return saida
}
func main() {
	// Produz uma sequência de 10 valores
	entrada := sequenciaNumeros(1, 10)
	// Um grupo de trabalhadores irá processar esses números
	saida := grupoDeTrabalhadores(entrada, 2)
// Somente termina quando todo o trabalho for processado
	for s := range saida {
		fmt.Println(s)
	}
}

📨 Requisição e resposta

Também conhecido como: canal de resposta, _RPC_ interno, _restoring sequencing_ (o nome que Rob Pike dá a um uso específico da ideia, comentado abaixo).

Canais são valores como qualquer outro, então uma mensagem pode carregar um canal. Quem envia uma requisição inclui nela o canal pelo qual quer receber a resposta e fica bloqueado lendo desse canal. Quem atende processa e responde no canal que veio na mensagem. Nenhum estado é compartilhado, pois pedido e resposta viajam por canais. É assim que se faz uma _goroutine_ funcionar como um serviço, e a ideia reaparece na goroutine dona do estado.

No exemplo, a função principal envia cinco requisições ao servico e espera cada resposta antes de enviar a próxima. O campo resposta é declarado como chan<- int, então o serviço só pode escrever nele.

Na palestra Go Concurrency Patterns, Pike usa a mesma ideia para "restaurar a sequência" de um fan-in. Cada mensagem carrega um canal wait, e quem produziu só envia a próxima depois que o leitor avisa, por esse canal, que terminou de processar a anterior.

package main
import "fmt"
// requisicao carrega, além do valor, o canal pelo qual quem pediu quer
// receber a resposta. O serviço só precisa escrever nele, por isso chan<-.
type requisicao struct {
	valor    int
	resposta chan<- int
}
// servico atende uma requisição por vez e responde no canal que veio
// dentro da própria mensagem.
func servico(entrada <-chan requisicao) {
	for req := range entrada {
		req.resposta <- req.valor * 2
	}
}
func main() {
	entrada := make(chan requisicao)
	pronto := make(chan struct{})
	go func() {
		servico(entrada)
		close(pronto)
	}()
for i := range 5 {
		resposta := make(chan int)
		entrada <- requisicao{valor: i, resposta: resposta}
		// Fica bloqueado até o serviço responder
		fmt.Println("resposta:", <-resposta)
	}
// Sem mais requisições: o serviço termina
	close(entrada)
	<-pronto
}

Parte 3 · Encerrando goroutines

Os geradores da Parte 2 têm um defeito em comum: só terminam se alguém ler todos os valores. Esta parte trata de como mandar uma _goroutine_ parar, como saber que ela parou e o que acontece quando ninguém faz isso.

🚏 Canal de parada (quit channel)

Também conhecido como: _quit channel_, canal done.

Um gerador sem fim, ou um consumidor que desiste no meio do caminho, deixa uma _goroutine_ bloqueada para sempre em um envio que ninguém vai receber. O canal de parada resolve isso. O gerador faz cada envio disputar, em um select, com um canal quit. Quando quem consome não quer mais valores, fecha o quit, e o gerador termina em vez de ficar bloqueado. O padrão vem da palestra Go Concurrency Patterns, de Rob Pike.

No exemplo, o contador geraria números para sempre. A função principal lê os três primeiros e fecha o quit.

Repare que o gerador fecha a saída ao sair. A função principal lê a saída até ela ser fechada, e assim tem certeza de que o gerador terminou. A variante em que o gerador confirma a parada pelo próprio quit está em parada com confirmação.

package main
import "fmt"
// contador é um gerador sem fim: envia 0, 1, 2... até que o canal quit seja
// fechado. Ao sair, fecha o canal de saída.
func contador(quit <-chan struct{}) <-chan int {
	saida := make(chan int)
	go func() {
		defer close(saida)
		for i := 0; ; i++ {
			// O envio disputa com o sinal de parada: o que puder
			// prosseguir primeiro, vence.
			select {
			case saida <- i:
			case <-quit:
				return
			}
		}
	}()
	return saida
}
func main() {
	quit := make(chan struct{})
	valores := contador(quit)
// Só queremos os três primeiros valores
	for range 3 {
		fmt.Println(<-valores)
	}
// Manda o gerador parar. Sem isto ele ficaria bloqueado no próximo
	// envio para sempre.
	close(quit)
// O gerador fecha a saída ao terminar: ler até o fechamento garante que
	// ele parou de fato.
	for range valores {
	}
	fmt.Println("o gerador parou")
}

🛑 Vazamento de goroutines e context

Uma _goroutine_ bloqueada em um canal que ninguém mais vai ler (ou escrever) nunca termina. Dizemos que ela vaza. O coletor de lixo não recolhe _goroutines_, então a memória e os recursos que ela segura ficam presos até o fim do programa. Em um programa curto isso passa despercebido. Em um servidor que roda por meses, é um vazamento de memória.

O gerador sequenciaNumeros, usado em vários exemplos, tem esse problema: ele só termina se alguém ler todos os valores. No exemplo, a função principal lê apenas os três primeiros e para. A _goroutine_ fica presa no envio do quarto valor. O programa mostra isso comparando runtime.NumGoroutine() antes e depois.

A solução é a mesma do canal de parada. Cada envio disputa, em um select, com um sinal de cancelamento. Só que, em vez de um canal quit próprio, o costume em Go é receber um context.Context e observar ctx.Done(), um canal que é fechado quando o contexto é cancelado. A vantagem é que o mesmo contexto atravessa várias funções e etapas de um _pipeline_, carrega prazos (context.WithTimeout) e cancela todo mundo de uma vez. Para se aprofundar, veja o repositório sobre context e a segunda metade do artigo sobre _pipelines_.

Na versão cancelável, a função principal lê três valores e chama cancel(). Sem essa chamada, a _goroutine_ ficaria presa exatamente como a primeira.

package main
import (
	"context"
	"fmt"
	"runtime"
)
// sequenciaNumeros é o gerador usado nos outros exemplos. Ele não é
// cancelável: se o consumidor parar de ler antes do fim, o envio bloqueia
// para sempre e a goroutine vaza.
func sequenciaNumeros(inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		for i := inicial; i <= final; i++ {
			saida <- i
		}
		close(saida)
	}()
	return saida
}
// sequenciaNumerosCancelavel faz cada envio disputar com ctx.Done():
// se o contexto for cancelado, a goroutine desiste do envio e termina.
func sequenciaNumerosCancelavel(ctx context.Context, inicial, final int) <-chan int {
	saida := make(chan int)
	go func() {
		defer close(saida)
		for i := inicial; i <= final; i++ {
			select {
			case saida <- i:
			case <-ctx.Done():
				fmt.Println("gerador: cancelado, encerrando")
				return
			}
		}
	}()
	return saida
}
func main() {
	antes := runtime.NumGoroutine()
// Sem cancelamento: lemos só os 3 primeiros valores e paramos.
	valores := sequenciaNumeros(1, 1000)
	for range 3 {
		fmt.Printf("valor: %v\n", <-valores)
	}
	// Ninguém mais vai ler de valores: a goroutine do gerador está presa
	// em saida <- 4 e continuará assim até o programa terminar.
	fmt.Printf("goroutines presas: %d\n", runtime.NumGoroutine()-antes)
// Com cancelamento: lemos os 3 primeiros valores e cancelamos.
	ctx, cancel := context.WithCancel(context.Background())
	cancelaveis := sequenciaNumerosCancelavel(ctx, 1, 1000)
	for range 3 {
		fmt.Printf("valor: %v\n", <-cancelaveis)
	}
	// Sem esta chamada a goroutine ficaria presa, como a anterior.
	cancel()
	// O gerador fecha o canal ao sair: drenar até o fechamento garante que
	// ele terminou de fato.
	for range cancelaveis {
	}
	fmt.Println("gerador cancelável encerrado")
// Parada com confirmação (veja quit_confirmacao.go)
	quitComConfirmacao()
	// Vários sinais de parada combinados em um só (veja qualquer.go)
	combinarSinais()
}

🤝 Parada com confirmação

Também conhecido como: _shutdown_ com _ack_, _graceful stop_.

Mandar "pare" não garante que a _goroutine_ já parou. Se ela precisa liberar recursos antes de sair (fechar arquivos, encerrar conexões), quem pediu a parada deve esperar a confirmação. Na palestra Go Concurrency Patterns, Pike faz isso reaproveitando o próprio canal quit. Quem quer parar envia "pare", a _goroutine_ faz a limpeza e responde no mesmo canal. Por isso, neste exemplo, o quit é um canal bidirecional, um dos raros casos em que isso é intencional.

A palestra Advanced Go Concurrency Patterns, de Sameer Ajmani, chega ao mesmo resultado com requisição e resposta. O método Close envia um canal de resposta por um chan chan error e espera nele. A _goroutine_ faz a limpeza e responde com o erro, se houver. Prefira essa forma quando a confirmação precisa carregar alguma informação.

Com context, o equivalente é chamar cancel() e depois esperar um canal pronto, que a _goroutine_ fecha ao terminar a limpeza. Isso é necessário porque o ctx só leva o sinal em um sentido. Foi o que o exemplo de context fez ao ler o canal do gerador até ele ser fechado.

package main
import (
	"fmt"
	"time"
)
// tagarelaComConfirmacao envia mensagens até receber algo no canal quit.
// Antes de sair faz a limpeza e confirma no MESMO canal que terminou,
// por isso o canal é bidirecional.
func tagarelaComConfirmacao(nome string, quit chan string) <-chan string {
	saida := make(chan string)
	go func() {
		for i := 0; ; i++ {
			select {
			case saida <- fmt.Sprintf("%s %d", nome, i):
			case <-quit:
				limpeza()
				quit <- "parei"
				return
			}
		}
	}()
	return saida
}
// limpeza simula a liberação de recursos: fechar arquivos, conexões etc.
func limpeza() {
	fmt.Println("gerador: liberando recursos...")
	time.Sleep(100 * time.Millisecond)
}
func quitComConfirmacao() {
	quit := make(chan string)
	c := tagarelaComConfirmacao("Duda", quit)
	for range 3 {
		fmt.Println(<-c)
	}
	quit <- "pare"
	// Só seguimos em frente depois que o gerador confirmar que terminou
	fmt.Println("gerador:", <-quit)
}

🧩 Combinar sinais de parada (or-channel)

Também conhecido como: _or-channel_. Não confunda com o _or-done-channel_, do mesmo livro citado abaixo, que é outra técnica. Ele embrulha a leitura de um canal para que ela também respeite um sinal de parada.

Às vezes uma _goroutine_ deve parar quando _qualquer um_ de vários sinais chegar: o contexto da requisição, um sinal do sistema operacional, um prazo global. Em vez de um select com um case por origem em cada _goroutine_, a função qualquer combina os canais em um só, que é fechado quando o primeiro deles fechar.

A implementação usa uma _goroutine_ por canal de entrada, e a primeira a ser acordada fecha a saída. Ela toma dois cuidados. O primeiro é o sync.Once, que garante um único close mesmo que dois sinais cheguem juntos, já que fechar um canal duas vezes causa _panic_. O segundo é que cada _goroutine_ também observa a própria saída. Assim, quando um sinal vence, as demais terminam em vez de vazarem esperando canais que talvez nunca fechem.

Existem alternativas. Para duas ou três origens, um select explícito é o mais claro. Para o caso geral há a versão recursiva, que divide a lista ao meio, e o reflect.Select. As duas funcionam, mas são mais engenhosas do que claras, e os Go Proverbs lembram que "_Clear is better than clever_" e "_Reflection is never clear_".

Se todos os sinais são contextos, prefira derivar um do outro, como em context.WithTimeout(ctxRequisicao, ...). O contexto filho já é cancelado quando o pai é. Combinar canais vale a pena quando as origens são independentes.

De onde vem isso? Sinalizar a parada fechando um canal aparece no artigo sobre _pipelines_, com o canal done, e na palestra Advanced Go Concurrency Patterns, de Sameer Ajmani (2013), que fecha um canal quit para encerrar as _goroutines_ do seu Merge. Nenhum dos dois combina vários sinais em um só. O _or-channel_, com esse nome, é do livro _Concurrency in Go_, de Katherine Cox-Buday (O'Reilly, 2017, capítulo 4), que usa a versão recursiva.

package main
import (
	"context"
	"fmt"
	"sync"
	"time"
)
// qualquer combina vários sinais de parada em um só: o canal devolvido é
// fechado assim que o primeiro dos canais recebidos for fechado.
func qualquer(canais ...<-chan struct{}) <-chan struct{} {
	saida := make(chan struct{})
	// Mais de um sinal pode chegar ao mesmo tempo, e fechar um canal duas
	// vezes causa panic: o sync.Once garante um único close.
	var once sync.Once
	for _, c := range canais {
		go func() {
			select {
			case <-c:
				once.Do(func() { close(saida) })
			case <-saida:
				// Outro sinal chegou primeiro: esta goroutine termina
				// em vez de ficar presa esperando c para sempre.
			}
		}()
	}
	return saida
}
func combinarSinais() {
	// Três origens independentes para o sinal de parada
	ctxRequisicao, cancelarRequisicao := context.WithCancel(context.Background())
	defer cancelarRequisicao()
	ctxPrazo, cancelarPrazo := context.WithTimeout(context.Background(), 250*time.Millisecond)
	defer cancelarPrazo()
	desligar := make(chan struct{}) // seria fechado ao receber um sinal do sistema operacional
parar := qualquer(ctxRequisicao.Done(), ctxPrazo.Done(), desligar)
ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	for {
		// Um único case de parada, não importa quantas origens existam
		select {
		case <-ticker.C:
			fmt.Println("trabalhando...")
		case <-parar:
			fmt.Println("um dos sinais de parada chegou (aqui, o prazo de 250ms)")
			return
		}
	}
}

Parte 4 · Controlando o ritmo

O que fazer quando quem produz e quem consome andam em velocidades diferentes. Cada padrão desta parte dá uma resposta: fazer o produtor esperar, limitar quantos executam ao mesmo tempo, limitar a taxa, agrupar o trab

... (README truncated for length)