GoSudaKit introduttivo alla concorrenza in Go
Panoramica
Breve introduzione
Il linguaggio Go dispone di molti strumenti per la gestione della concorrenza. In questo articolo presenterò alcuni di essi insieme ad alcuni accorgimenti.
Goroutine?
Le goroutine sono un nuovo modello di concorrenza supportato dal linguaggio Go. In generale, per eseguire più attività contemporaneamente, un programma riceve dal sistema operativo dei thread OS e svolge il lavoro in parallelo, in numero pari ai core disponibili. Per realizzare una concorrenza a granularità più fine, si creano invece green thread a livello userland, in modo che più green thread vengano eseguiti a turno all'interno di un singolo thread OS. Le goroutine, tuttavia, rendono questo tipo di green thread più piccolo ed efficiente. Esse utilizzano meno memoria rispetto ai thread e possono essere create e sostituite più rapidamente di questi.
Per utilizzare le goroutine è sufficiente usare la parola chiave go. Ciò consente, in modo intuitivo durante la scrittura del programma, di eseguire come codice asincrono un codice che è sincrono.
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}
Questo codice trasforma in un flusso asincrono un codice sincrono che attende brevemente un secondo e poi stampa Hello, World!. L'esempio è semplice, ma quando si converte da sincrono ad asincrono un codice un po' più complesso, la leggibilità, la visibilità e la comprensibilità del codice migliorano rispetto a metodi come async await o promise.
Tuttavia, in molti casi, senza comprendere il semplice flusso di chiamata asincrona di questo codice sincrono e flussi come fork & join (un flusso simile alla divide et impera), si finisce per produrre codice con goroutine di scarsa qualità. Per affrontare queste situazioni, in questo articolo presenterò alcuni metodi e tecniche.
Gestione della concorrenza
context
Potrebbe sembrare sorprendente che il primo strumento di gestione presentato sia context. Tuttavia, in Go, context va oltre la semplice funzionalità di cancellazione e svolge un ruolo eccellente nella gestione dell'intero albero dei lavori. Per chi non lo conoscesse, descrivo brevemente questo package.
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}
Il codice sopra utilizza context per stampare Context is done! dopo un secondo. context permette di verificare se è avvenuta la cancellazione tramite il metodo Done(), e offre vari modi di cancellazione attraverso metodi come WithCancel, WithTimeout, WithDeadline e WithValue.
Creiamo un semplice esempio. Supponiamo di scrivere codice che, utilizzando il pattern aggregator, recupera user, post e comment, e che tutte le richieste debbano completarsi entro due secondi. Si può scrivere come segue.
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}
Il codice sopra stampa Timeout! se non tutti i dati vengono recuperati entro due secondi, mentre stampa All data is fetched! se tutti i dati vengono recuperati. In questo modo, utilizzando context, è possibile gestire facilmente cancellazioni e timeout anche in codice con più goroutine in esecuzione.
Le varie funzioni e metodi relativi a context sono disponibili in godoc context. Spero che, dopo averne studiato le parti semplici, possiate utilizzarli con facilità.
channel
unbuffered channel
channel è uno strumento per la comunicazione tra goroutine. Un channel può essere creato con make(chan T), dove T è il tipo dei dati che il channel trasmetterà. I dati si scambiano tramite <-, e il channel può essere chiuso 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}
Il codice sopra utilizza un channel per stampare 1 e 2. In questo esempio si mostra soltanto l'invio e la ricezione di valori, ma il channel offre molte altre funzionalità. Vediamo innanzitutto buffered channel e unbuffered channel. Prima di iniziare, l'esempio scritto sopra è un unbuffered channel: l'azione di inviare dati nel canale e quella di riceverli devono avvenire contemporaneamente. Se tali azioni non avvengono nello stesso momento, può verificarsi un deadlock.
buffered channel
Che cosa accadrebbe se il codice precedente, anziché limitarsi a stampare, eseguisse due processi che svolgono lavori pesanti? Se il secondo processo, mentre legge ed elabora, si bloccasse per un lungo periodo, anche il primo processo si fermerebbe per tutto quel tempo. Per evitare una simile situazione, si può utilizzare un 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}
Il codice sopra utilizza un buffered channel per stampare 1 e 2. In questo esempio, grazie al buffered channel, l'invio e la ricezione dei dati non devono più avvenire contemporaneamente. Dotando il canale di un buffer, si ottiene uno spazio pari alla sua lunghezza, che permette di prevenire i ritardi di lavoro causati dall'influenza di operazioni con priorità inferiore.
select
Quando si gestiscono più canali, l'istruzione select permette di implementare facilmente una struttura 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}
Il codice sopra crea tre canali che trasmettono periodicamente 1, 2 e 3, e utilizza select per ricevere e stampare i valori dai canali. Utilizzando select in questo modo, è possibile ricevere dati contemporaneamente da più canali ed elaborarli nell'ordine in cui arrivano.
for range
channel può ricevere dati facilmente tramite for range. Usando for range su un canale, il ciclo viene eseguito ogni volta che nel canale viene aggiunto un dato, e termina quando il canale viene chiuso.
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}
Il codice sopra utilizza un channel per stampare 1 e 2. In questo esempio, tramite for range, ogni dato aggiunto al canale viene ricevuto e stampato. Quando il canale viene chiuso, il ciclo termina.
Come mostrato più volte sopra, questa sintassi può anche essere usata come semplice strumento di sincronizzazione.
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}
Il codice sopra attende un secondo e poi stampa Hello, World!. In questo esempio, tramite un channel, il codice sincrono è stato trasformato in codice asincrono. In questo modo, utilizzando channel, è facile trasformare codice sincrono in asincrono e definire un punto di join.
etc
- Inviare o ricevere dati su un nil channel può portare a un ciclo infinito e causare un deadlock.
- Se si invia un dato dopo aver chiuso un canale, si verifica un panic.
- Anche senza chiudere esplicitamente un canale, il GC lo raccoglie e lo chiude.
mutex
spinlock
spinlock è un metodo di sincronizzazione che tenta ripetutamente di acquisire il lock in un ciclo. In Go, uno spinlock può essere implementato facilmente utilizzando i puntatori.
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}
Il codice sopra implementa il package spinlock. Qui SpinLock è realizzato utilizzando il package sync/atomic. Il metodo Lock tenta di acquisire il lock tramite atomic.CompareAndSwapUintptr, mentre il metodo Unlock rilascia il lock tramite atomic.StoreUintptr. Poiché questo metodo tenta di acquisire il lock senza sosta, la CPU viene utilizzata continuamente finché il lock non viene ottenuto, e può quindi cadere in un ciclo infinito. Pertanto, lo spinlock è preferibile per una sincronizzazione semplice o per l'uso limitato a brevi intervalli di tempo.
sync.Mutex
mutex è uno strumento per la sincronizzazione tra goroutine. Il mutex fornito dal package sync offre metodi come Lock, Unlock, RLock e RUnlock. Un mutex può essere creato come sync.Mutex, mentre per un lock di lettura/scrittura si può usare 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}
Nel codice sopra, due goroutine accedono quasi simultaneamente alla stessa variabile count. In questo caso, racchiudendo in una sezione critica il codice che accede a count tramite mutex, si impedisce l'accesso concorrente alla variabile count. In questo modo il codice stamperà sempre 2, indipendentemente da quante volte venga eseguito.
sync.RWMutex
sync.RWMutex è un mutex che permette di distinguere e utilizzare separatamente il lock di lettura e quello di scrittura. I metodi RLock e RUnlock consentono di acquisire e rilasciare il lock di lettura.
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}
Il codice sopra implementa ConcurrentMap utilizzando sync.RWMutex. In questo codice, il metodo Get acquisisce il lock di lettura, mentre il metodo Set acquisisce il lock di scrittura, permettendo di accedere e modificare la mappa data in modo sicuro. Il lock di lettura è necessario perché, nei casi in cui le operazioni di sola lettura sono numerose, si può acquisire solo il lock di lettura, senza il lock di scrittura, in modo che più goroutine possano eseguire letture contemporaneamente. In questo modo, nei casi in cui non è necessario il lock di scrittura perché lo stato non viene modificato, si può migliorare le prestazioni acquisendo solo il lock di lettura.
fakelock
fakelock è un semplice accorgimento che implementa sync.Locker. Questa struttura fornisce gli stessi metodi di sync.Mutex, ma in realtà non esegue alcuna operazione.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Il codice sopra implementa il package fakelock. Questo package implementa sync.Locker e fornisce i metodi Lock e Unlock, ma in realtà non esegue alcuna operazione. Spiegherò in seguito, se se ne presenterà l'occasione, perché un codice simile sia necessario.
waitgroup
sync.WaitGroup
sync.WaitGroup è uno strumento che attende il completamento di tutti i lavori delle goroutine. Fornisce i metodi Add, Done e Wait: Add aggiunge il numero di goroutine, Done segnala che il lavoro di una goroutine è terminato, e Wait attende che tutte le goroutine completino il lavoro.
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}
Il codice sopra utilizza sync.WaitGroup affinché 100 goroutine sommino concorrentemente un valore alla variabile c. In questo codice, si attende con sync.WaitGroup che tutte le goroutine terminino, dopodiché si stampa il valore accumulato in c. Quando si esegue un fork & join di pochi lavori semplici, è sufficiente usare solo i canali, mentre per un fork & join di molti lavori, l'uso di sync.WaitGroup è una buona scelta.
with slice
Se utilizzato insieme a uno slice, waitgroup può diventare uno strumento eccellente per gestire lavori concorrenti senza lock.
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}
Il codice sopra utilizza solo waitgroup affinché ciascuna goroutine generi concorrentemente 10 numeri interi casuali e li salvi nell'indice assegnato. In questo codice, si attende con waitgroup che tutte le goroutine terminino, dopodiché si stampa Done. In questo modo, utilizzando waitgroup, più goroutine possono eseguire lavori contemporaneamente, salvare i dati senza lock finché tutte non terminano, e poi eseguire in blocco la post-elaborazione al termine dei lavori.
golang.org/x/sync/errgroup.ErrGroup
errgroup è un package che estende sync.WaitGroup. A differenza di sync.WaitGroup, errgroup annulla tutte le goroutine e restituisce l'errore se anche solo una delle goroutine genera un errore.
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}
Il codice sopra utilizza errgroup per creare 10 goroutine, e fa sì che la quinta goroutine generi un errore. Ho generato intenzionalmente un errore nella quinta goroutine per mostrare il caso in cui si verifica un errore. Nell'uso reale, si crea le goroutine con errgroup e si eseguono varie operazioni di post-elaborazione nel caso in cui si verifichi un errore in ciascuna goroutine.
once
È uno strumento che esegue codice che deve essere eseguito una sola volta. Il codice correlato può essere eseguito tramite i costruttori seguenti.
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 fa semplicemente in modo che la funzione in questione venga eseguita una sola volta in tutto il programma.
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}
Il codice sopra utilizza sync.OnceFunc per stampare Hello, World!. In questo codice, si crea la funzione once con sync.OnceFunc, e anche chiamando più volte la funzione once, Hello, World! viene stampato una sola volta.
OnceValue
OnceValue non si limita a eseguire la funzione una sola volta in tutto il programma, ma memorizza il valore restituito dalla funzione e, alle chiamate successive, restituisce il valore memorizzato.
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}
Il codice sopra utilizza sync.OnceValue per incrementare di 1 la variabile c. In questo codice, si crea la funzione once con sync.OnceValue, e anche chiamando più volte la funzione once, la variabile c viene incrementata una sola volta e restituisce 1.
OnceValues
OnceValues funziona allo stesso modo di OnceValue, ma può restituire più valori.
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}
Il codice sopra utilizza sync.OnceValues per incrementare di 1 la variabile c. In questo codice, si crea la funzione once con sync.OnceValues, e anche chiamando più volte la funzione once, la variabile c viene incrementata una sola volta e restituisce 1.
atomic
Il package atomic fornisce operazioni atomiche. Offre metodi come Add, CompareAndSwap, Load, Store e Swap, ma di recente si consiglia l'uso di tipi come Int64, Uint64 e 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}
È l'esempio già visto in precedenza. Il codice utilizza il tipo atomic.Int64 per incrementare in modo atomico la variabile c. Con i metodi Add e Load si può incrementare atomicamente la variabile e leggerne il valore. Inoltre, il metodo Store permette di memorizzare un valore, il metodo Swap di sostituirlo, e il metodo CompareAndSwap di confrontare il valore e, se è appropriato, sostituirlo.
cond
sync.Cond
Il package cond fornisce variabili di condizione. Può essere creato come sync.Cond e offre i metodi Wait, Signal e 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}
Il codice sopra utilizza sync.Cond per attendere che la variabile ready diventi true. In questo codice, si attende con sync.Cond che ready diventi true, dopodiché si stampa Ready!. In questo modo, utilizzando sync.Cond, più goroutine possono attendere che una specifica condizione sia soddisfatta.
Sfruttando questo, si può implementare una semplice queue.
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}
Utilizzando così sync.Cond, invece di consumare molta CPU con uno spin-lock, si può attendere in modo efficiente e riprendere l'esecuzione quando la condizione viene soddisfatta.
semaphore
golang.org/x/sync/semaphore.Semaphore
Il package semaphore fornisce semafori. Può essere creato come golang.org/x/sync/semaphore.Semaphore e offre i metodi Acquire, Release e 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}
Il codice sopra crea un semaforo con semaphore e lo utilizza per acquisire il semaforo tramite il metodo Acquire e rilasciarlo tramite il metodo Release. In questo codice si è mostrato come acquisire e rilasciare un semaforo utilizzando semaphore.
Conclusioni
Penso che fin qui le nozioni di base siano sufficienti. Spero che, sulla base di questo articolo, possiate comprendere come gestire la concorrenza utilizzando le goroutine e che possiate effettivamente applicarla. Spero che questo articolo vi sia stato utile. Grazie.