Tutorial do Pool de Goroutines Tunny

Tunny é uma biblioteca Golang para criar e gerenciar pools de goroutines, permitindo limitar o trabalho a partir de qualquer número de goroutines usando APIs síncronas.

Quando seu trabalho vem de um número arbitrário de fontes assíncronas, mas sua capacidade de processamento paralelo é limitada, um pool fixo de goroutines é extremamente útil. Por exemplo, ao processar trabalhos de solicitações HTTP intensivas em CPU, você pode criar um pool do tamanho do número de CPUs.

Instalação

go get github.com/Jeffail/tunny

Alternativamente, usando dep:

dep ensure -add github.com/Jeffail/tunny

Uso

Para a maioria dos casos, seu trabalho pesado pode ser representado por um simples func(), caso em que você pode usar NewFunc. Vamos ver como usar nosso exemplo de contagem de CPU para solicitações HTTP:

package main

import (
	"io/ioutil"
	"net/http"
	"runtime"

	"github.com/Jeffail/tunny"
)

func main() {
	numCPUs := runtime.NumCPU()

	pool := tunny.NewFunc(numCPUs, func(payload interface{}) interface{} {
		var result []byte

		// TODO: Realizar algumas operações intensivas em CPU usando payload

		return result
	})
	defer pool.Close()

	http.HandleFunc("/work", func(w http.ResponseWriter, r *http.Request) {
		input, err := ioutil.ReadAll(r.Body)
		if err != nil {
			http.Error(w, "Erro interno", http.StatusInternalServerError)
		}
		defer r.Body.Close()

		// Importar este trabalho para o nosso pool. Esta chamada é síncrona e bloqueará até que o trabalho esteja completo.
		result := pool.Process(input)

		w.Write(result.([]byte))
	})

	http.ListenAndServe(":8080", nil)
}

Tunny também suporta timeouts. Você pode substituir a chamada Process acima pelo seguinte código:

result, err := pool.ProcessTimed(input, time.Second*5)
if err == tunny.ErrJobTimedOut {
	http.Error(w, "Requisição expirou", http.StatusRequestTimeout)
}

Você também pode usar o contexto da solicitação (ou qualquer outro contexto) para lidar com timeouts e prazos. Basta substituir a chamada Process pelo seguinte código:

result, err := pool.ProcessCtx(r.Context(), input)
if err == context.DeadlineExceeded {
	http.Error(w, "Requisição expirou", http.StatusRequestTimeout)
}

Modificando o Tamanho do Pool

Você pode usar SetSize(int) para mudar o tamanho do pool Tunny a qualquer momento.

pool.SetSize(10) // 10 goroutines
pool.SetSize(100) // 100 goroutines

Isso é seguro mesmo que outras goroutines ainda estejam processando.

Goroutine com Estado

Às vezes, cada goroutine no pool Tunny precisa de seu próprio estado de gerenciamento. Nesse caso, você deve implementar tunny.Worker, que inclui chamadas para término, interrupção (se um trabalho expira e não é mais necessário) e bloqueio da alocação do próximo trabalho até que uma certa condição seja atendida.

Ao criar um pool com o tipo Worker, você precisa fornecer um construtor para gerar sua implementação personalizada:

pool := tunny.New(poolSize, func() Worker {
	// TODO: Realizar alocação de estado para cada goroutine aqui.
	return newCustomWorker()
})

Dessa forma, Tunny pode limpar a criação e destruição do tipo Worker quando o tamanho do pool muda.

Ordenação

Trabalhos em backlog não têm garantia de serem processados na ordem. Devido à implementação atual de canais e blocos de seleção, as pilhas de trabalhos em backlog serão processadas como uma fila FIFO. No entanto, esse comportamento não faz parte da especificação e não deve ser confiado.