--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Kit introduttivo alla concorrenza in Go description: Vengono presentati vari strumenti e tecniche per la gestione della concorrenza in Go, tra cui goroutine, canali e mutex. language: it date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## 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. ```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 {} } ``` 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. ```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) } ``` 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. ```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!") } } ``` 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](https://pkg.go.dev/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`. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` 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`. ```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) } } ``` 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`. ```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) } } } ``` 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. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` 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. ```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 {} } ``` 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 1. Inviare o ricevere dati su un nil channel può portare a un ciclo infinito e causare un deadlock. 2. Se si invia un dato dopo aver chiuso un canale, si verifica un panic. 3. 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. ```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{} } ``` 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`. ```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) } ``` 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. ```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 } ``` 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. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (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. ```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()) } ``` 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. ```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) } } ``` 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. ```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) } } ``` 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. ```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` fa semplicemente in modo che la funzione in questione venga eseguita una sola volta in tutto il programma. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` 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. ```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()) } ``` 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. ```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) } ``` 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`. ```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()) } ``` È 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`. ```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!") } ``` 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`. ```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 } ``` 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`. ```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) } ``` 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.