GoSudaKit inicial de concurrencia en Go
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.
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch := make(chan struct{})
10 go func() {
11 defer close(ch)
12 time.Sleep(1 * time.Second)
13 fmt.Println("Hello, World!")
14 }()
15
16 fmt.Println("Waiting for goroutine...")
17 for range ch {}
18}
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.
1package main
2
3func main() {
4 ctx, cancel := context.WithCancel(context.Background())
5 defer cancel()
6
7 go func() {
8 <-ctx.Done()
9 fmt.Println("Context is done!")
10 }()
11
12 time.Sleep(1 * time.Second)
13
14 cancel()
15
16 time.Sleep(1 * time.Second)
17}
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.
1package main
2
3func main() {
4 ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
5 defer cancel()
6
7 ch := make(chan struct{})
8 go func() {
9 defer close(ch)
10 user := getUser(ctx)
11 post := getPost(ctx)
12 comment := getComment(ctx)
13
14 fmt.Println(user, post, comment)
15 }()
16
17 select {
18 case <-ctx.Done():
19 fmt.Println("Timeout!")
20 case <-ch:
21 fmt.Println("All data is fetched!")
22 }
23}
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. 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.
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
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.
1package main
2
3func main() {
4 ch := make(chan int, 2)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
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.
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch1 := make(chan int, 10)
10 ch2 := make(chan int, 10)
11 ch3 := make(chan int, 10)
12
13 go func() {
14 for {
15 ch1 <- 1
16 time.Sleep(1 * time.Second)
17 }
18 }()
19 go func() {
20 for {
21 ch2 <- 2
22 time.Sleep(2 * time.Second)
23 }
24 }()
25 go func() {
26 for {
27 ch3 <- 3
28 time.Sleep(3 * time.Second)
29 }
30 }()
31
32 for i := 0; i < 3; i++ {
33 select {
34 case v := <-ch1:
35 fmt.Println(v)
36 case v := <-ch2:
37 fmt.Println(v)
38 case v := <-ch3:
39 fmt.Println(v)
40 }
41 }
42}
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.
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
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.
1package main
2
3func main() {
4 ch := make(chan struct{})
5 go func() {
6 defer close(ch)
7 time.Sleep(1 * time.Second)
8 fmt.Println("Hello, World!")
9 }()
10
11 fmt.Println("Waiting for goroutine...")
12 for range ch {}
13}
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
- Si se envían o reciben datos en un nil channel, puede caerse en un bucle infinito y producirse un deadlock.
- Si se envían datos después de cerrar un canal, se produce un panic.
- 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.
1package spinlock
2
3import (
4 "runtime"
5 "sync/atomic"
6)
7
8type SpinLock struct {
9 lock uintptr
10}
11
12func (s *SpinLock) Lock() {
13 for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
14 runtime.Gosched()
15 }
16}
17
18func (s *SpinLock) Unlock() {
19 atomic.StoreUintptr(&s.lock, 0)
20}
21
22func NewSpinLock() *SpinLock {
23 return &SpinLock{}
24}
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.
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 var mu sync.Mutex
9 var count int
10
11 go func() {
12 mu.Lock()
13 count++
14 mu.Unlock()
15 }()
16
17 mu.Lock()
18 count++
19 mu.Unlock()
20
21 println(count)
22}
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.
1package cmap
2
3import (
4 "sync"
5)
6
7type ConcurrentMap[K comparable, V any] struct {
8 sync.RWMutex
9 data map[K]V
10}
11
12func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
13 m.RLock()
14 defer m.RUnlock()
15
16 value, ok := m.data[key]
17 return value, ok
18}
19
20func (m *ConcurrentMap[K, V]) Set(key K, value V) {
21 m.Lock()
22 defer m.Unlock()
23
24 m.data[key] = value
25}
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.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (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.
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
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.
1package main
2
3import (
4 "fmt"
5 "sync"
6 "rand"
7)
8
9func main() {
10 var wg sync.WaitGroup
11 arr := [10]int{}
12
13 for i := 0; i < 10; i++ {
14 wg.Add(1)
15 go func(id int) {
16 defer wg.Done()
17
18 arr[id] = rand.Intn(100)
19 }(i)
20 }
21
22 wg.Wait()
23 fmt.Println("Done")
24
25 for i, v := range arr {
26 fmt.Printf("arr[%d] = %d\n", i, v)
27 }
28}
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.
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/errgroup"
7)
8
9func main() {
10 g, ctx := errgroup.WithContext(context.Background())
11 _ = ctx
12
13 for i := 0; i < 10; i++ {
14 i := i
15 g.Go(func() error {
16 if i == 5 {
17 return fmt.Errorf("error")
18 }
19 return nil
20 })
21 }
22
23 if err := g.Wait(); err != nil {
24 fmt.Println(err)
25 }
26}
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.
1func OnceFunc(f func()) func()
2func OnceValue[T any](f func() T) func() T
3func 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.
1package main
2
3import "sync"
4
5func main() {
6 once := sync.OnceFunc(func() {
7 println("Hello, World!")
8 })
9
10 once()
11 once()
12 once()
13 once()
14 once()
15}
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.
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValue(func() int {
8 c += 1
9 return c
10 })
11
12 println(once())
13 println(once())
14 println(once())
15 println(once())
16 println(once())
17}
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.
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValues(func() (int, int) {
8 c += 1
9 return c, c
10 })
11
12 a, b := once()
13 println(a, b)
14 a, b = once()
15 println(a, b)
16 a, b = once()
17 println(a, b)
18 a, b = once()
19 println(a, b)
20 a, b = once()
21 println(a, b)
22}
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.
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
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.
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 c := sync.NewCond(&sync.Mutex{})
9 ready := false
10
11 go func() {
12 c.L.Lock()
13 ready = true
14 c.Signal()
15 c.L.Unlock()
16 }()
17
18 c.L.Lock()
19 for !ready {
20 c.Wait()
21 }
22 c.L.Unlock()
23
24 println("Ready!")
25}
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.
1package queue
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8type Node[T any] struct {
9 Value T
10 Next *Node[T]
11}
12
13type Queue[T any] struct {
14 sync.Mutex
15 Cond *sync.Cond
16 Head *Node[T]
17 Tail *Node[T]
18 Len int
19}
20
21func New[T any]() *Queue[T] {
22 q := &Queue[T]{}
23 q.Cond = sync.NewCond(&q.Mutex)
24 return q
25}
26
27func (q *Queue[T]) Push(value T) {
28 q.Lock()
29 defer q.Unlock()
30
31 node := &Node[T]{Value: value}
32 if q.Len == 0 {
33 q.Head = node
34 q.Tail = node
35 } else {
36 q.Tail.Next = node
37 q.Tail = node
38 }
39 q.Len++
40 q.Cond.Signal()
41}
42
43func (q *Queue[T]) Pop() T {
44 q.Lock()
45 defer q.Unlock()
46
47 for q.Len == 0 {
48 q.Cond.Wait()
49 }
50
51 node := q.Head
52 q.Head = q.Head.Next
53 q.Len--
54 return node.Value
55}
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.
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/semaphore"
7)
8
9func main() {
10 s := semaphore.NewWeighted(1)
11
12 if s.TryAcquire(1) {
13 fmt.Println("Acquired!")
14 } else {
15 fmt.Println("Not Acquired!")
16 }
17
18 s.Release(1)
19}
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.