--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Pachetul de pornire pentru concurență în Go description: Prezentăm diverse instrumente și tehnici pentru gestionarea concurenței în Go, precum goroutine, canale și mutex-uri. language: ro date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Prezentare generală ### Scurtă introducere Limbajul Go dispune de multe instrumente pentru gestionarea concurenței. În acest articol vom prezenta o parte dintre acestea și câteva trucuri. ### Goroutine? goroutine este un nou model de concurență suportat de limbajul Go. În mod obișnuit, un program primește de la sistemul de operare fire de execuție OS pentru a efectua mai multe sarcini simultan, executând sarcinile în paralel, în număr egal cu numărul de nuclee. Pentru a realiza concurența la o scară mai mică, se creează în spațiul utilizatorului (userland) fire verzi (green threads), astfel încât mai multe fire verzi rulează pe rând în cadrul unui singur fir OS. Goroutine-urile reprezintă o formă de fire verzi mai mică și mai eficientă. Ele utilizează mai puțină memorie decât firele de execuție și pot fi create și schimbate mai rapid decât acestea. Pentru a utiliza goroutine, este suficient să folosiți cuvântul cheie `go`. Acest lucru permite ca, în procesul scrierii programului, codul sincron să fie executat intuitiv ca cod asincron. ```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 {} } ``` Acest cod transformă un cod sincron, care așteaptă 1 secundă și apoi afișează `Hello, World!`, într-un flux asincron. Exemplul este simplu, dar atunci când un cod mai complex este transformat din sincron în asincron, lizibilitatea, vizibilitatea și înțelegerea codului devin mai bune decât cu metode precum async await sau promise. Totuși, în multe cazuri, dacă nu se înțeleg fluxul de apelare asincronă a unui cod sincron și fluxuri precum `fork & join` (un flux asemănător cu divide et impera), pot rezulta goroutine de calitate slabă. Pentru a face față unor astfel de cazuri, în acest articol vom prezenta câteva metode și tehnici. ## Gestionarea concurenței ### context Este posibil să fie surprinzător că prima tehnică prezentată este `context`. Totuși, în limbajul Go, `context` depășește simpla funcționalitate de anulare și joacă un rol excelent în gestionarea întregului arbore de sarcini. Pentru cei care nu cunosc acest pachet, îl voi descrie pe scurt. ```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) } ``` Codul de mai sus utilizează `context` pentru a afișa `Context is done!` după 1 secundă. `context` permite verificarea anulării prin metoda `Done()` și oferă diverse metode de anulare, precum `WithCancel`, `WithTimeout`, `WithDeadline` și `WithValue`. Să construim un exemplu simplu. Să presupunem că scrieți un cod care utilizează modelul `aggregator` pentru a obține `user`, `post` și `comment`, iar toate cererile trebuie să fie efectuate în decurs de 2 secunde. Atunci puteți scrie astfel: ```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!") } } ``` Codul de mai sus afișează `Timeout!` dacă nu reușește să obțină toate datele în decurs de 2 secunde și afișează `All data is fetched!` dacă le obține pe toate. În acest mod, folosind `context`, anularea și expirarea timpului pot fi gestionate cu ușurință chiar și în cadrul unui cod în care rulează mai multe goroutine. Diversele funcții și metode legate de context pot fi consultate pe [godoc context](https://pkg.go.dev/context). Sper că, după ce veți studia cele simple, le veți putea utiliza cu ușurință. ### channel #### unbuffered channel `channel` este un instrument pentru comunicarea între goroutine. `channel` poate fi creat cu `make(chan T)`, unde `T` este tipul de date pe care îl va transmite acel `channel`. Datele se trimit și se primesc cu `<-`, iar `channel` poate fi închis cu `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) } } ``` Codul de mai sus utilizează `channel` pentru a afișa 1 și 2. Aici se demonstrează doar trimiterea și primirea simplă a valorilor prin `channel`. Totuși, `channel` oferă mai multe funcționalități decât atât. Mai întâi, să analizăm `buffered channel` și `unbuffered channel`. Exemplul de mai sus este un `unbuffered channel`, în care acțiunea de trimitere a datelor în canal și acțiunea de primire a acestora trebuie să aibă loc simultan. Dacă aceste acțiuni nu au loc simultan, poate apărea un deadlock. #### buffered channel Ce se întâmplă dacă, în loc de o simplă afișare, codul de mai sus ar reprezenta două procese care efectuează sarcini grele? Dacă al doilea proces, care citește și procesează datele, rămâne blocat o perioadă lungă, atunci și primul proces se va opri pe durata respectivă. Pentru a preveni această situație, putem utiliza `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) } } ``` Codul de mai sus utilizează `buffered channel` pentru a afișa 1 și 2. În acest cod, `buffered channel` face ca acțiunea de trimitere a datelor în canal și acțiunea de primire a acestora să nu fie obligatoriu simultane. Atunci când canalul are un buffer, există un spațiu liber egal cu lungimea acestuia, ceea ce permite prevenirea întârzierii sarcinilor cauzate de influența sarcinilor ulterioare. #### select Atunci când lucrați cu mai multe canale, puteți implementa cu ușurință o structură `fan-in` folosind `select`. ```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) } } } ``` Codul de mai sus creează 3 canale care transmit periodic valorile 1, 2 și 3, și utilizează `select` pentru a primi valori din canale și a le afișa. În acest mod, folosind `select`, se pot primi date simultan din mai multe canale și le puteți procesa pe măsură ce sosesc. #### for range `channel` poate fi utilizat cu `for range` pentru a primi datele cu ușurință. Atunci când `for range` este folosit pe un canal, bucla se execută de fiecare dată când se adaugă date în canal și se încheie când canalul este închis. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Codul de mai sus utilizează `channel` pentru a afișa 1 și 2. În acest cod, `for range` este utilizat pentru a primi și afișa datele de fiecare dată când sunt adăugate în canal. Când canalul este închis, bucla se încheie. După cum s-a arătat de câteva ori mai sus, această sintaxă poate fi utilizată și ca simplu mecanism de sincronizare. ```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 {} } ``` Codul de mai sus așteaptă 1 secundă și apoi afișează `Hello, World!`. În acest cod, `channel` este utilizat pentru a transforma codul sincron în cod asincron. În acest mod, folosind `channel`, codul sincron poate fi transformat ușor în cod asincron, iar un punct de `join` poate fi stabilit. #### etc 1. Dacă se trimit sau se primesc date pe un nil channel, se poate intra într-o buclă infinită, ceea ce poate duce la deadlock. 2. Dacă se trimit date după ce canalul a fost închis, apare un panic. 3. Chiar dacă nu închideți canalul în mod explicit, GC îl va colecta și îl va închide. ### mutex #### spinlock `spinlock` este o metodă de sincronizare care încearcă în mod repetat să obțină blocarea, parcurgând o buclă. În limbajul Go, un spinlock poate fi implementat cu ușurință folosind pointeri. ```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{} } ``` Codul de mai sus este implementarea pachetului `spinlock`. În acest cod, `SpinLock` este implementat folosind pachetul `sync/atomic`. Metoda `Lock` încearcă să obțină blocarea folosind `atomic.CompareAndSwapUintptr`, iar metoda `Unlock` eliberează blocarea folosind `atomic.StoreUintptr`. Deoarece această metodă încearcă neîntrerupt să obțină blocarea, procesorul este utilizat continuu până când blocarea este obținută, ceea ce poate duce la o buclă infinită. Prin urmare, `spinlock` este recomandat pentru sincronizări simple sau pentru utilizare doar pe perioade scurte de timp. #### sync.Mutex `mutex` este un instrument pentru sincronizarea între goroutine. `mutex`-ul furnizat de pachetul `sync` oferă metode precum `Lock`, `Unlock`, `RLock` și `RUnlock`. `mutex` poate fi creat cu `sync.Mutex`, iar pentru blocarea de citire/scriere se poate utiliza `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) } ``` În codul de mai sus, două goroutine accesează aproape simultan aceeași variabilă `count`. Dacă folosim `mutex` pentru a transforma codul care accesează variabila `count` într-o secțiune critică, putem preveni accesul concurent la aceasta. Astfel, acest cod va afișa întotdeauna `2`, indiferent de numărul de rulări. #### sync.RWMutex `sync.RWMutex` este un `mutex` care permite utilizarea separată a blocărilor de citire și de scriere. Cu ajutorul metodelor `RLock` și `RUnlock` se pot seta și elibera blocările de citire. ```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 } ``` Codul de mai sus implementează `ConcurrentMap` folosind `sync.RWMutex`. În acest cod, metoda `Get` setează o blocare de citire, iar metoda `Set` setează o blocare de scriere, astfel încât harta `data` poate fi accesată și modificată în siguranță. Motivul pentru care se utilizează blocarea de citire este că, în cazul în care există multe operații simple de citire, se permite ca mai multe goroutine să efectueze simultan operații de citire, fără a bloca și scrierea. Astfel, în situațiile în care nu există modificări de stare și nu este necesară blocarea de scriere, performanța poate fi îmbunătățită utilizând doar blocarea de citire. #### fakelock `fakelock` este un truc simplu de implementare a `sync.Locker`. Această structură oferă aceleași metode ca `sync.Mutex`, dar nu efectuează nicio acțiune reală. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` Codul de mai sus este implementarea pachetului `fakelock`. Acest pachet implementează `sync.Locker` și oferă metodele `Lock` și `Unlock`, dar în realitate nu efectuează nicio operație. Voi explica mai târziu, dacă va fi ocazia, de ce este necesar un astfel de cod. ### waitgroup #### sync.WaitGroup `sync.WaitGroup` este un instrument care așteaptă până când toate sarcinile goroutine-urilor sunt finalizate. Oferă metodele `Add`, `Done` și `Wait`: prin `Add` se adaugă numărul de goroutine, prin `Done` se anunță că o goroutine și-a încheiat sarcina, iar prin `Wait` se așteaptă până când toate goroutine-urile își termină sarcinile. ```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()) } ``` Codul de mai sus utilizează `sync.WaitGroup` pentru ca 100 de goroutine să adauge simultan valori variabilei `c`. În acest cod, `sync.WaitGroup` este utilizat pentru a aștepta finalizarea tuturor goroutine-urilor, după care se afișează valoarea acumulată în `c`. Pentru un `fork & join` al unui număr mic de sarcini, este suficient să utilizați doar canale, dar pentru un `fork & join` al unui număr mare de sarcini, utilizarea `sync.WaitGroup` este o alegere bună. #### with slice Atunci când este utilizat împreună cu un slice, `waitgroup` poate deveni un instrument excelent pentru gestionarea sarcinilor executate concurent, fără blocări. ```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) } } ``` Codul de mai sus utilizează doar `waitgroup` pentru ca fiecare goroutine să genereze simultan câte un număr întreg aleatoriu și să îl stocheze la indexul atribuit. În acest cod, `waitgroup` este utilizat pentru a aștepta finalizarea tuturor goroutine-urilor, după care se afișează `Done`. În acest mod, folosind `waitgroup`, mai multe goroutine pot efectua sarcini simultan și pot stoca datele fără blocări, iar după finalizarea tuturor goroutine-urilor se poate efectua centralizat o prelucrare finală. #### golang.org/x/sync/errgroup.ErrGroup `errgroup` este un pachet care extinde `sync.WaitGroup`. Spre deosebire de `sync.WaitGroup`, `errgroup` anulează toate goroutine-urile și returnează eroarea dacă apare o eroare în una dintre sarcini. ```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) } } ``` Codul de mai sus utilizează `errgroup` pentru a crea 10 goroutine, iar în a cincea goroutine se generează o eroare. Am generat intenționat o eroare în a cincea goroutine pentru a arăta cazul în care apare o eroare. În utilizarea reală, `errgroup` se folosește pentru a crea goroutine și pentru a efectua diverse prelucrări ulterioare în cazul în care apare o eroare în oricare dintre ele. ### once Este un instrument pentru executarea codului care trebuie să ruleze o singură dată. Codul respectiv poate fi executat prin constructorii de mai jos. ```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` permite pur și simplu ca funcția respectivă să fie executată exact o singură dată, pe întreaga durată. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` Codul de mai sus utilizează `sync.OnceFunc` pentru a afișa `Hello, World!`. În acest cod, `sync.OnceFunc` este utilizat pentru a crea funcția `once`, iar chiar dacă aceasta este apelată de mai multe ori, `Hello, World!` va fi afișat o singură dată. #### OnceValue `OnceValue` nu înseamnă doar că funcția respectivă este executată exact o singură dată pe întreaga durată, ci și că valoarea returnată de funcție este stocată, iar la apelurile ulterioare se returnează valoarea salvată. ```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()) } ``` Codul de mai sus utilizează `sync.OnceValue` pentru a incrementa variabila `c` cu 1. În acest cod, `sync.OnceValue` este utilizat pentru a crea funcția `once`, iar chiar dacă aceasta este apelată de mai multe ori, variabila `c` este incrementată o singură dată și funcția returnează valoarea 1. #### OnceValues `OnceValues` funcționează în același mod ca `OnceValue`, dar poate returna mai multe 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) } ``` Codul de mai sus utilizează `sync.OnceValues` pentru a incrementa variabila `c` cu 1. În acest cod, `sync.OnceValues` este utilizat pentru a crea funcția `once`, iar chiar dacă aceasta este apelată de mai multe ori, variabila `c` este incrementată o singură dată și funcția returnează valorile corespunzătoare lui 1. ### atomic Pachetul `atomic` oferă operații atomice. Acesta oferă metode precum `Add`, `CompareAndSwap`, `Load`, `Store` și `Swap`, dar în ultima vreme se recomandă utilizarea tipurilor precum `Int64`, `Uint64` și `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()) } ``` Este exemplul utilizat anterior. Codul folosește tipul `atomic.Int64` pentru a incrementa atomic variabila `c`. Metodele `Add` și `Load` permit incrementarea atomică a variabilei și citirea acesteia. De asemenea, metoda `Store` permite stocarea unei valori, metoda `Swap` permite înlocuirea valorii, iar metoda `CompareAndSwap` permite compararea valorii și, dacă aceasta corespunde, înlocuirea ei. ### cond #### sync.Cond Pachetul `cond` oferă variabile de condiție. Acesta poate fi creat cu `sync.Cond` și oferă metodele `Wait`, `Signal` și `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!") } ``` Codul de mai sus utilizează `sync.Cond` pentru a aștepta până când variabila `ready` devine `true`. În acest cod, `sync.Cond` este utilizat pentru a aștepta până când variabila `ready` devine `true`, după care se afișează `Ready!`. În acest mod, folosind `sync.Cond`, mai multe goroutine pot aștepta până când este îndeplinită o anumită condiție. Folosind acest mecanism, putem implementa o `queue` simplă. ```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 } ``` În acest mod, folosind `sync.Cond`, în loc să consumăm mult CPU cu `spin-lock`, putem aștepta eficient și putem relua activitatea atunci când condiția este îndeplinită. ### semaphore #### golang.org/x/sync/semaphore.Semaphore Pachetul `semaphore` oferă semafoare. Acesta poate fi creat cu `golang.org/x/sync/semaphore.Semaphore` și oferă metodele `Acquire`, `Release` și `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) } ``` Codul de mai sus creează un semafor folosind `semaphore`, apoi utilizează metoda `TryAcquire` pentru a încerca obținerea acestuia și metoda `Release` pentru a-l elibera. În acest cod, este prezentat modul în care semaforul poate fi obținut și eliberat. ## Încheiere Conținutul de bază ar trebui să fie suficient până aici. Pe baza acestui articol, sper că veți înțelege cum să gestionați concurența folosind goroutine și că veți putea aplica aceste cunoștințe în practică. Sper că acest articol v-a fost util. Mulțumesc.