Tutorial del Tunny Goroutine Pool

Tunny es una biblioteca de Golang para crear y gestionar pools de goroutines, lo que te permite limitar el trabajo de cualquier número de goroutines utilizando APIs síncronas.

Cuando tu trabajo proviene de un número arbitrario de fuentes asíncronas pero tu capacidad de procesamiento paralelo es limitada, un pool de goroutines fijo es extremadamente útil. Por ejemplo, al procesar trabajos de solicitudes HTTP intensivas en CPU, puedes crear un pool del tamaño del número de CPUs.

Instalación

go get github.com/Jeffail/tunny

O alternativamente, usando dep:

dep ensure -add github.com/Jeffail/tunny

Uso

En la mayoría de los casos, tu trabajo pesado puede estar representado por un simple func(), en cuyo caso puedes usar NewFunc. Veamos cómo usar nuestro ejemplo de solicitudes HTTP para el conteo de CPU:

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 algunas operaciones intensivas en CPU utilizando 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, "Error interno", http.StatusInternalServerError)
		}
		defer r.Body.Close()

		// Importar este trabajo a nuestro pool. Esta llamada es síncrona y bloqueará hasta que el trabajo esté completo.
		result := pool.Process(input)

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

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

Tunny también admite tiempos de espera. Puedes reemplazar la llamada Process anterior con el siguiente código:

result, err := pool.ProcessTimed(input, time.Second*5)
if err == tunny.ErrJobTimedOut {
	http.Error(w, "Tiempo de espera agotado", http.StatusRequestTimeout)
}

También puedes usar el contexto de la solicitud (u cualquier otro contexto) para manejar tiempos de espera y fechas límite. Simplemente reemplaza la llamada Process con el siguiente código:

result, err := pool.ProcessCtx(r.Context(), input)
if err == context.DeadlineExceeded {
	http.Error(w, "Tiempo de espera agotado", http.StatusRequestTimeout)
}

Modificando el tamaño del pool

Puedes usar SetSize(int) para cambiar el tamaño del pool Tunny en cualquier momento.

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

Esto es seguro incluso si otras goroutines todavía están procesando.

Goroutine con estado

A veces, cada goroutine en el pool Tunny necesita su propio estado de gestión. En ese caso, debes implementar tunny.Worker, que incluye llamadas para terminación, interrupción (si un trabajo tarda y ya no es necesario) y bloqueo de la asignación del próximo trabajo hasta que se cumpla cierta condición.

Al crear un pool con el tipo Worker, debes proporcionar un constructor para generar tu implementación personalizada:

pool := tunny.New(poolSize, func() Worker {
	// TODO: Realizar la asignación de estado para cada goroutine aquí.
	return newCustomWorker()
})

De esta manera, Tunny puede limpiar la creación y destrucción del tipo Worker cuando cambia el tamaño del pool.

Orden

Los trabajos en cola no están garantizados para ser procesados en orden. Debido a la implementación actual de canales y bloques de selección, las pilas de trabajos en cola serán procesadas como una cola FIFO. Sin embargo, este comportamiento no es parte de la especificación y no se debe confiar en él.