--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Kit inicial de concurrencia en Go description: Se presentan diversas herramientas y técnicas para la gestión de la concurrencia en Go, incluyendo goroutines, canales y mutexes. language: es date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Descripción general ### Introducción breve El lenguaje Go cuenta con muchas herramientas para la gestión de la concurrencia. En este artículo les presentaré algunas de ellas y algunos trucos. ### ¿Goroutines? Una goroutine es un nuevo modelo de concurrencia compatible con el lenguaje Go. En general, un programa obtiene hilos del sistema operativo (OS threads) del SO para realizar varias tareas al mismo tiempo, ejecutándolas en paralelo según el número de núcleos. Y para realizar concurrencia a una escala más pequeña, se crean hilos verdes (green threads) en el espacio de usuario, de modo que varios hilos verdes se ejecuten por turnos dentro de un único hilo del sistema operativo. Sin embargo, las goroutines hicieron este tipo de hilos verdes más pequeños y eficientes. Estas goroutines utilizan menos memoria que los hilos y pueden crearse y reemplazarse más rápidamente que ellos. Para usar una goroutine basta con emplear la palabra clave `go`. Esto permite ejecutar de forma intuitiva, durante el proceso de escritura del programa, código síncrono como código asíncrono. ```go package main import ( "fmt" "time" ) func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` Este código convierte en un flujo asíncrono un código síncrono que espera un segundo y luego imprime `Hello, World!`. El ejemplo actual es sencillo, pero al convertir código algo más complejo de síncrono a asíncrono, la legibilidad, la visibilidad y la comprensión del código mejoran respecto a enfoques como async/await o promise. Sin embargo, en muchos casos, si no se comprende el flujo de simplemente invocar de forma asíncrona este tipo de código síncrono, ni flujos como `fork & join` (que se asemejan a la división y conquista), pueden producirse goroutines de mala calidad. Para afrontar estas situaciones, en este artículo presentaré algunos métodos y técnicas. ## Gestión de la concurrencia ### context Puede resultar sorprendente que `context` aparezca como primera técnica de gestión. Pero en Go, `context` va más allá de la simple cancelación y desempeña un papel excelente en la gestión de todo el árbol de tareas. Para quienes no lo conozcan, explicaré brevemente este paquete. ```go package main func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { <-ctx.Done() fmt.Println("Context is done!") }() time.Sleep(1 * time.Second) cancel() time.Sleep(1 * time.Second) } ``` El código anterior utiliza `context` para imprimir `Context is done!` después de un segundo. `context` permite comprobar si se ha cancelado mediante el método `Done()`, y ofrece diversos métodos de cancelación mediante `WithCancel`, `WithTimeout`, `WithDeadline`, `WithValue`, entre otros. Vamos a crear un ejemplo sencillo. Supongamos que escribe un código que utiliza el patrón `aggregator` para obtener `user`, `post` y `comment`, y que todas las solicitudes deben completarse en menos de 2 segundos. Entonces puede escribirlo de la siguiente manera. ```go package main func main() { ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second) defer cancel() ch := make(chan struct{}) go func() { defer close(ch) user := getUser(ctx) post := getPost(ctx) comment := getComment(ctx) fmt.Println(user, post, comment) }() select { case <-ctx.Done(): fmt.Println("Timeout!") case <-ch: fmt.Println("All data is fetched!") } } ``` El código anterior imprime `Timeout!` si no se obtienen todos los datos en 2 segundos, y `All data is fetched!` si se obtienen todos. De esta manera, al usar `context`, incluso en código con varias goroutines en ejecución, pueden gestionarse fácilmente la cancelación y los tiempos de espera. Puede consultar las diversas funciones y métodos relacionados con context en [godoc context](https://pkg.go.dev/context). Espero que, tras estudiar lo básico, pueda utilizarlo con comodidad. ### channel #### unbuffered channel `channel` es una herramienta para la comunicación entre goroutines. Un `channel` puede crearse con `make(chan T)`. Aquí, `T` es el tipo de datos que transmitirá dicho `channel`. Los datos se envían y reciben con `<-`, y el `channel` puede cerrarse con `close`. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` El código anterior utiliza `channel` para imprimir 1 y 2. En este código solo se muestra el envío y la recepción simple de valores en el canal, pero `channel` ofrece más funciones que esta. Primero veamos `buffered channel` y `unbuffered channel`. Antes de empezar, el ejemplo escrito arriba es un `unbuffered channel`, en el que el envío de datos al canal y la recepción de datos deben producirse al mismo tiempo. Si estas acciones no ocurren simultáneamente, puede producirse un deadlock. #### buffered channel ¿Y si el código anterior no fuera una simple impresión, sino dos procesos que realizan tareas pesadas? Si el segundo proceso lee y procesa datos y tarda mucho tiempo, el primer proceso también quedará detenido durante ese tiempo. Para evitar esta situación, podemos utilizar `buffered channel`. ```go package main func main() { ch := make(chan int, 2) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` El código anterior utiliza `buffered channel` para imprimir 1 y 2. En este código, mediante `buffered channel`, se logra que el envío de datos al canal y la recepción de datos no tengan que ocurrir simultáneamente. Al colocar un buffer en el canal, se dispone de espacio equivalente a su longitud, lo que permite evitar retrasos en el trabajo causados por la influencia de tareas posteriores. #### select Cuando se trabaja con varios canales, la sintaxis `select` permite implementar fácilmente una estructura `fan-in`. ```go package main import ( "fmt" "time" ) func main() { ch1 := make(chan int, 10) ch2 := make(chan int, 10) ch3 := make(chan int, 10) go func() { for { ch1 <- 1 time.Sleep(1 * time.Second) } }() go func() { for { ch2 <- 2 time.Sleep(2 * time.Second) } }() go func() { for { ch3 <- 3 time.Sleep(3 * time.Second) } }() for i := 0; i < 3; i++ { select { case v := <-ch1: fmt.Println(v) case v := <-ch2: fmt.Println(v) case v := <-ch3: fmt.Println(v) } } } ``` El código anterior crea tres canales que transmiten periódicamente los valores 1, 2 y 3, y utiliza `select` para recibir valores de los canales e imprimirlos. De esta manera, al usar `select`, se pueden recibir datos de varios canales simultáneamente y procesarlos en cuanto llega cada valor. #### for range `channel` puede recibir datos fácilmente mediante `for range`. Al usar `for range` con un canal, el bucle se ejecuta cada vez que se añade un dato al canal, y termina cuando el canal se cierra. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` El código anterior utiliza `channel` para imprimir 1 y 2. En este código, `for range` recibe e imprime los datos cada vez que se añade un dato al canal. Y cuando el canal se cierra, el bucle termina. Como se ha escrito varias veces anteriormente, esta sintaxis también puede usarse como un simple mecanismo de sincronización. ```go package main func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` El código anterior espera un segundo y luego imprime `Hello, World!`. En este código, se utiliza `channel` para convertir código síncrono en código asíncrono. De esta manera, al usar `channel`, se puede convertir fácilmente código síncrono en asíncrono y establecer un punto de `join`. #### etc 1. Si se envían o reciben datos en un nil channel, puede caerse en un bucle infinito y producirse un deadlock. 2. Si se envían datos después de cerrar un canal, se produce un panic. 3. Aunque no se cierre un canal explícitamente, el GC lo recogerá y lo cerrará. ### mutex #### spinlock `spinlock` es un método de sincronización que intenta adquirir el lock continuamente dentro de un bucle. En Go, un spinlock puede implementarse fácilmente utilizando punteros. ```go package spinlock import ( "runtime" "sync/atomic" ) type SpinLock struct { lock uintptr } func (s *SpinLock) Lock() { for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) { runtime.Gosched() } } func (s *SpinLock) Unlock() { atomic.StoreUintptr(&s.lock, 0) } func NewSpinLock() *SpinLock { return &SpinLock{} } ``` El código anterior implementa el paquete `spinlock`. En este código, `SpinLock` se implementa utilizando el paquete `sync/atomic`. El método `Lock` intenta obtener el lock con `atomic.CompareAndSwapUintptr`, y el método `Unlock` libera el lock con `atomic.StoreUintptr`. Como este método intenta adquirir el lock sin descanso, consume CPU continuamente hasta obtenerlo, y puede caer en un bucle infinito. Por ello, `spinlock` es recomendable para una sincronización simple o para su uso durante periodos de tiempo breves. #### sync.Mutex `mutex` es una herramienta para la sincronización entre goroutines. El `mutex` que proporciona el paquete `sync` ofrece métodos como `Lock`, `Unlock`, `RLock` y `RUnlock`. Un `mutex` puede crearse con `sync.Mutex`, y también puede usarse un lock de lectura/escritura con `sync.RWMutex`. ```go package main import ( "sync" ) func main() { var mu sync.Mutex var count int go func() { mu.Lock() count++ mu.Unlock() }() mu.Lock() count++ mu.Unlock() println(count) } ``` En el código anterior, dos goroutines acceden casi simultáneamente a la misma variable `count`. En ese caso, si se utiliza `mutex` para convertir el código que accede a la variable `count` en una sección crítica, se puede evitar el acceso concurrente a `count`. Entonces, este código imprimirá siempre `2`, sin importar cuántas veces se ejecute. #### sync.RWMutex `sync.RWMutex` es un `mutex` que permite distinguir entre el lock de lectura y el lock de escritura. Con los métodos `RLock` y `RUnlock` se puede adquirir y liberar el lock de lectura. ```go package cmap import ( "sync" ) type ConcurrentMap[K comparable, V any] struct { sync.RWMutex data map[K]V } func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) { m.RLock() defer m.RUnlock() value, ok := m.data[key] return value, ok } func (m *ConcurrentMap[K, V]) Set(key K, value V) { m.Lock() defer m.Unlock() m.data[key] = value } ``` El código anterior implementa `ConcurrentMap` utilizando `sync.RWMutex`. En este código, el método `Get` adquiere el lock de lectura y el método `Set` adquiere el lock de escritura, de modo que se puede acceder y modificar el mapa `data` de forma segura. El motivo de usar el lock de lectura es que, en casos con muchas operaciones de lectura simples, no se adquiere el lock de escritura, sino solo el lock de lectura, para que varias goroutines puedan realizar operaciones de lectura al mismo tiempo. De este modo, en los casos en que no hay cambios de estado y no es necesario el lock de escritura, adquirir solo el lock de lectura permite mejorar el rendimiento. #### fakelock `fakelock` es un truco sencillo que implementa `sync.Locker`. Esta estructura proporciona los mismos métodos que `sync.Mutex`, pero en realidad no realiza ninguna acción. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` El código anterior implementa el paquete `fakelock`. Este paquete implementa `sync.Locker` y proporciona los métodos `Lock` y `Unlock`, pero en realidad no hace nada. Explicaré más adelante, si se presenta la ocasión, por qué se necesita este código. ### waitgroup #### sync.WaitGroup `sync.WaitGroup` es una herramienta que espera a que terminen todas las tareas de las goroutines. Proporciona los métodos `Add`, `Done` y `Wait`: `Add` agrega el número de goroutines, `Done` notifica que la tarea de una goroutine ha terminado, y `Wait` espera a que terminen las tareas de todas las goroutines. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` El código anterior utiliza `sync.WaitGroup` para que 100 goroutines sumen valores simultáneamente a la variable `c`. En este código, con `sync.WaitGroup` se espera a que terminen todas las goroutines y después se imprime el valor acumulado en `c`. Cuando se realiza un `fork & join` de unas pocas tareas sencillas, basta con usar únicamente canales, pero cuando se realiza un `fork & join` de una gran cantidad de tareas, usar `sync.WaitGroup` también es una buena opción. #### with slice Si se utiliza junto con un slice, `waitgroup` puede convertirse en una excelente herramienta para gestionar tareas concurrentes sin necesidad de locks. ```go package main import ( "fmt" "sync" "rand" ) func main() { var wg sync.WaitGroup arr := [10]int{} for i := 0; i < 10; i++ { wg.Add(1) go func(id int) { defer wg.Done() arr[id] = rand.Intn(100) }(i) } wg.Wait() fmt.Println("Done") for i, v := range arr { fmt.Printf("arr[%d] = %d\n", i, v) } } ``` El código anterior utiliza únicamente `waitgroup` para que cada goroutine genere de forma concurrente 10 números enteros aleatorios y los guarde en el índice asignado. En este código, con `waitgroup` se espera a que terminen todas las goroutines y después se imprime `Done`. De esta manera, al usar `waitgroup`, varias goroutines pueden realizar tareas simultáneamente, guardar datos sin locks hasta que todas terminen, y realizar un procesamiento posterior de forma conjunta al finalizar las tareas. #### golang.org/x/sync/errgroup.ErrGroup `errgroup` es un paquete que extiende `sync.WaitGroup`. A diferencia de `sync.WaitGroup`, si se produce un error en cualquiera de las tareas de las goroutines, `errgroup` cancela todas las goroutines y devuelve el error. ```go package main import ( "context" "fmt" "golang.org/x/sync/errgroup" ) func main() { g, ctx := errgroup.WithContext(context.Background()) _ = ctx for i := 0; i < 10; i++ { i := i g.Go(func() error { if i == 5 { return fmt.Errorf("error") } return nil }) } if err := g.Wait(); err != nil { fmt.Println(err) } } ``` El código anterior utiliza `errgroup` para crear 10 goroutines, y provoca un error en la quinta goroutine. Se generó intencionadamente un error en la quinta goroutine para mostrar el caso en el que ocurre un error. Sin embargo, en un uso real, basta con crear goroutines con `errgroup` y aplicar distintos procesamientos posteriores en los casos en que alguna goroutine produzca un error. ### once Es una herramienta para ejecutar código que debe ejecutarse una sola vez. El código relacionado puede ejecutarse mediante los siguientes constructores. ```go func OnceFunc(f func()) func() func OnceValue[T any](f func() T) func() T func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2) ``` #### OnceFunc `OnceFunc` simplemente permite que la función correspondiente se ejecute una única vez en todo el programa. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` El código anterior utiliza `sync.OnceFunc` para imprimir `Hello, World!`. En este código, se crea la función `once` con `sync.OnceFunc`, y aunque la función `once` se invoque varias veces, `Hello, World!` solo se imprime una vez. #### OnceValue `OnceValue` no solo ejecuta la función correspondiente una única vez en todo el programa, sino que también almacena el valor devuelto por dicha función y, al volver a invocarla, devuelve el valor almacenado. ```go package main import "sync" func main() { c := 0 once := sync.OnceValue(func() int { c += 1 return c }) println(once()) println(once()) println(once()) println(once()) println(once()) } ``` El código anterior utiliza `sync.OnceValue` para incrementar la variable `c` en 1. En este código, se crea la función `once` con `sync.OnceValue`, y aunque la función `once` se invoque varias veces, la variable `c` solo se incrementa una vez y se devuelve 1. #### OnceValues `OnceValues` funciona igual que `OnceValue`, pero puede devolver varios valores. ```go package main import "sync" func main() { c := 0 once := sync.OnceValues(func() (int, int) { c += 1 return c, c }) a, b := once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) } ``` El código anterior utiliza `sync.OnceValues` para incrementar la variable `c` en 1. En este código, se crea la función `once` con `sync.OnceValues`, y aunque la función `once` se invoque varias veces, la variable `c` solo se incrementa una vez y se devuelve 1. ### atomic El paquete `atomic` es un paquete que proporciona operaciones atómicas. Ofrece métodos como `Add`, `CompareAndSwap`, `Load`, `Store` y `Swap`, pero recientemente se recomienda el uso de tipos como `Int64`, `Uint64` y `Pointer`. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` Es el ejemplo que se utilizó anteriormente. Es un código que incrementa de forma atómica la variable `c` usando el tipo `atomic.Int64`. Con los métodos `Add` y `Load` se puede incrementar la variable de forma atómica y leer su valor. Además, con el método `Store` se puede guardar un valor, con `Swap` se puede reemplazar un valor, y con `CompareAndSwap` se puede comparar un valor y reemplazarlo si cumple la condición. ### cond #### sync.Cond El paquete `cond` es un paquete que proporciona variables de condición. Puede crearse con `sync.Cond` y ofrece los métodos `Wait`, `Signal` y `Broadcast`. ```go package main import ( "sync" ) func main() { c := sync.NewCond(&sync.Mutex{}) ready := false go func() { c.L.Lock() ready = true c.Signal() c.L.Unlock() }() c.L.Lock() for !ready { c.Wait() } c.L.Unlock() println("Ready!") } ``` El código anterior utiliza `sync.Cond` para esperar hasta que la variable `ready` sea `true`. En este código, se espera con `sync.Cond` hasta que `ready` sea `true` y, después, se imprime `Ready!`. De esta manera, al usar `sync.Cond`, varias goroutines pueden esperar simultáneamente a que se cumpla una condición determinada. Aprovechando esto, puede implementarse una `queue` sencilla. ```go package queue import ( "sync" "sync/atomic" ) type Node[T any] struct { Value T Next *Node[T] } type Queue[T any] struct { sync.Mutex Cond *sync.Cond Head *Node[T] Tail *Node[T] Len int } func New[T any]() *Queue[T] { q := &Queue[T]{} q.Cond = sync.NewCond(&q.Mutex) return q } func (q *Queue[T]) Push(value T) { q.Lock() defer q.Unlock() node := &Node[T]{Value: value} if q.Len == 0 { q.Head = node q.Tail = node } else { q.Tail.Next = node q.Tail = node } q.Len++ q.Cond.Signal() } func (q *Queue[T]) Pop() T { q.Lock() defer q.Unlock() for q.Len == 0 { q.Cond.Wait() } node := q.Head q.Head = q.Head.Next q.Len-- return node.Value } ``` Si se utiliza `sync.Cond` de esta manera, en lugar de consumir mucha CPU con un `spin-lock`, se puede esperar de forma eficiente y reanudar la ejecución cuando se cumple la condición. ### semaphore #### golang.org/x/sync/semaphore.Semaphore El paquete `semaphore` es un paquete que proporciona semáforos. Puede crearse con `golang.org/x/sync/semaphore.Semaphore` y ofrece los métodos `Acquire`, `Release` y `TryAcquire`. ```go package main import ( "context" "fmt" "golang.org/x/sync/semaphore" ) func main() { s := semaphore.NewWeighted(1) if s.TryAcquire(1) { fmt.Println("Acquired!") } else { fmt.Println("Not Acquired!") } s.Release(1) } ``` El código anterior crea un semáforo con `semaphore`, lo adquiere con el método `Acquire` y lo libera con el método `Release`. En este código se mostró cómo adquirir y liberar un semáforo usando `semaphore`. ## Conclusión Creo que con lo básico expuesto hasta aquí es suficiente. Espero que, con base en el contenido de este artículo, puedan comprender cómo gestionar la concurrencia usando goroutines y llegar a utilizarla en la práctica. Espero que este artículo les haya sido útil. Gracias.