Go Concurență Pachet de start
Sinteză
Introducere succintă
Limbajul Go dispune de numeroase instrumente pentru gestionarea concurenței. În acest articol, vom prezenta o parte dintre acestea, precum și diverse tehnici aferente.
Goroutine?
O goroutine reprezintă un model de concurență de tip nou, suportat în limbajul Go. În mod obișnuit, pentru a executa simultan multiple sarcini, un program primește thread-uri de la sistemul de operare (OS) și le execută în paralel, corespunzător numărului de nuclee disponibile. Pentru a realiza o concurență la o granularitate mai fină, se creează thread-uri gestionate de utilizator (green threads), permițând ca mai multe astfel de thread-uri să ruleze și să execute sarcini în cadrul unui singur thread OS. Cu toate acestea, în cazul goroutinelor, această formă de thread-uri gestionate de utilizator a fost rafinată pentru a fi mai mici și mai eficiente. Aceste goroutine utilizează mai puțină memorie decât thread-urile și pot fi create și comutate mai rapid decât acestea.
Pentru a utiliza o goroutine, este suficientă simpla utilizare a cuvântului cheie go. Acest fapt permite executarea intuitivă a codului sincron ca și cod asincron în procesul de redactare a programului.
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}
Acest cod transformă codul sincron, care așteaptă simplu timp de 1 secundă înainte de a afișa Hello, World!, într-un flux asincron. Deși exemplul actual este simplu, atunci când se transformă codul sincron în cod asincron pentru secvențe mai complexe, lizibilitatea, vizibilitatea și înțelegerea codului devin superioare față de metodele existente precum async await sau promise.
Totuși, în multe cazuri, atunci când nu se înțelege fluxul de apelare simplu asincron a codului sincron sau fluxul de tip fork & join (similar cu abordarea divide et impera), se pot genera implementări goroutine suboptime. În acest articol, vom prezenta câteva metode și tehnici pentru a ne pregăti în vederea acestor situații.
Gestionarea Concurenței
context
Poate fi neașteptat ca context să apară ca prima tehnică de gestionare. Însă, în limbajul Go, context joacă un rol excelent în managementul întregului arbore de sarcini, depășind funcționalitatea simplă de anulare. Pentru cei care nu sunt familiarizați, voi prezenta pe scurt pachetul aferent.
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}
Codul de mai sus este unul care utilizează context pentru a afișa Context is done! după o secundă. context permite verificarea stării de anulare prin metoda Done(), oferind totodată diverse modalități de anulare prin metode precum WithCancel, WithTimeout, WithDeadline și WithValue.
Vom crea un exemplu simplu. Să presupunem că scrieți un cod care utilizează modelul aggregator pentru a extrage anumite date, cum ar fi user, post și comment. Dacă toate solicitările trebuie finalizate în decurs de 2 secunde, codul poate fi structurat astfel:
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}
Codul de mai sus afișează Timeout! dacă toate datele nu sunt preluate în decurs de 2 secunde și All data is fetched! dacă toate datele sunt obținute. Prin utilizarea context în acest mod, se poate gestiona cu ușurință anularea și limitarea temporală (timeout) chiar și în codul unde operează multiple goroutine-uri.
Diferitele funcții și metode legate de context, înrudite cu aceasta, pot fi consultate la godoc context. Sperăm că veți învăța elementele simple și le veți putea utiliza cu ușurință.
channel
unbuffered channel
channel este un instrument pentru comunicarea între goroutine-uri. channel poate fi creat prin make(chan T). În acest caz, T reprezintă tipul datelor pe care le va transmite respectivul channel. channel permite trimiterea și primirea de date prin <-, iar închiderea channel-ului se realizează prin 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}
Codul de mai sus este unul care utilizează un channel pentru a afișa 1 și 2. Acest cod demonstrează doar trimiterea și primirea simplă de valori în channel. Totuși, channel-ul oferă funcționalități mult mai extinse. În primul rând, vom examina buffered channel și unbuffered channel. Înainte de a începe, exemplul redactat mai sus este un unbuffered channel, ceea ce impune ca acțiunea de trimitere a datelor în canal și acțiunea de primire a datelor să aibă loc simultan. Dacă aceste acțiuni nu se desfășoară concomitent, poate surveni un deadlock.
buffered channel
Ce s-ar întâmpla dacă, în loc de o simplă afișare, cele două procese ar executa sarcini intensive? În cazul în care procesul secundar se blochează pentru o perioadă îndelungată în timp ce efectuează procesarea prin citire, și primul proces va fi suspendat pe durata respectivă. Putem utiliza buffered channel pentru a preveni o astfel de situație.
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}
Codul de mai sus este unul care utilizează un buffered channel pentru a afișa 1 și 2. În acest cod, prin utilizarea buffered channel, s-a permis ca acțiunile de trimitere și primire a datelor în channel să nu mai fie necesar să se realizeze simultan. Prin alocarea unui buffer în canal, se creează o marjă de flexibilitate egală cu lungimea bufferului, prevenind astfel întârzierile de execuție cauzate de influența sarcinilor subsecvente.
select
Atunci când se gestionează multiple canale, sintaxa select permite implementarea facilă a structurii 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}
Codul de mai sus creează trei canale care transmit periodic valorile 1, 2 și 3 și utilizează select pentru a primi și afișa valorile din canale. Folosind select în acest mod, se pot primi date simultan de la multiple canale și se poate realiza procesarea imediat ce o valoare este recepționată din orice canal.
for range
channel permite primirea facilă de date utilizând for range. Atunci când for range este aplicat unui canal, acesta se execută de fiecare dată când o nouă dată este adăugată în canal, iar bucla se încheie la închiderea canalului.
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}
Codul de mai sus este unul care utilizează un 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 acestea sunt adăugate în canal. Buclea se termină atunci când canalul este închis.
Așa cum s-a menționat de mai multe ori mai sus, această sintagmă poate fi utilizată și ca simplu mijloc de sincronizare.
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}
Codul de mai sus este unul care așteaptă o secundă înainte de a afișa Hello, World!. În acest cod, s-a utilizat un channel pentru a transforma codul sincron în cod asincron. Utilizând channel în acest mod, se pot efectua conversii simple de la cod sincron la cod asincron și se poate stabili un punct de join.
diverse
- Expedierea sau primirea de date către sau de la un channel nil poate duce la intrarea într-o buclă infinită și la apariția unui deadlock.
- Expedierea de date după închiderea canalului va genera o eroare de tip panic.
- Chiar dacă canalul nu este închis explicit, Garbage Collector-ul (GC) îl va colecta și închide în procesul de curățare.
mutex
spinlock
spinlock este o metodă de sincronizare care încearcă continuu să obțină blocarea printr-o buclă iterativă. În limbajul Go, se poate implementa un spinlock cu ușurință utilizând pointeri.
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}
Codul de mai sus implementează pachetul spinlock. În acest cod, s-a utilizat pachetul sync/atomic pentru a implementa SpinLock. Metoda Lock încearcă obținerea blocării utilizând atomic.CompareAndSwapUintptr, iar metoda Unlock eliberează blocarea utilizând atomic.StoreUintptr. Deoarece această metodă încearcă blocarea neîntrerupt, poate consuma continuu resursele CPU până la obținerea blocării, riscând intrarea într-o buclă infinită. Prin urmare, se recomandă utilizarea spinlock doar pentru sincronizări simple sau pentru perioade de timp foarte scurte.
sync.Mutex
mutex este un instrument pentru sincronizarea între goroutine-uri. mutex oferit de pachetul sync furnizează metode precum Lock, Unlock, RLock și RUnlock. mutex poate fi creat ca sync.Mutex, iar pentru blocări de citire/scriere se poate utiliza 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}
În codul de mai sus, aproape simultan, două goroutine-uri accesează aceeași variabilă count. În acest moment, prin utilizarea mutex pentru a delimita secțiunea de cod care accesează variabila count ca regiune critică, se poate preveni accesul concurent la variabila count. Astfel, acest cod va afișa în mod consecvent valoarea 2, indiferent de numărul de execuții.
sync.RWMutex
sync.RWMutex este un mutex care permite utilizarea separată a blocării de citire și a blocării de scriere. Blocarea de citire se poate aplica și elibera utilizând metodele RLock și RUnlock.
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}
Codul de mai sus implementează ConcurrentMap utilizând sync.RWMutex. În acest cod, accesul și modificarea sigură a map-ului data sunt asigurate prin aplicarea unei blocări de citire în metoda Get și a unei blocări de scriere în metoda Set. Necesitatea blocării de citire apare în situațiile în care există multe operațiuni simple de citire, permițând ca mai multe goroutine-uri să execute concomitent operațiuni de citire fără a aplica blocarea de scriere. Prin aceasta, în cazurile în care nu există modificări de stare și nu este necesară aplicarea blocării de scriere, se poate îmbunătăți performanța aplicând doar blocarea de citire.
fakelock
fakelock este un truc simplu de implementare a interfeței sync.Locker. Această structură oferă aceleași metode ca sync.Mutex, dar nu execută nicio operațiune efectivă.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Codul de mai sus implementează pachetul fakelock. Acest pachet implementează sync.Locker, oferind metodele Lock și Unlock, dar fără a executa de fapt nicio acțiune. Motivul pentru care un astfel de cod este necesar va fi detaliat la o oportunitate viitoare.
waitgroup
sync.WaitGroup
sync.WaitGroup este un instrument utilizat pentru a aștepta finalizarea tuturor operațiunilor goroutine-urilor. Acesta oferă metodele Add, Done și Wait; metoda Add este folosită pentru a incrementa numărul de goroutine-uri, iar metoda Done semnalează finalizarea lucrului unei goroutine. Metoda Wait suspendă execuția până când toate goroutine-urile și-au încheiat sarcinile.
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}
Acesta este exemplul utilizat anterior. Acest cod incrementează atomic variabila c utilizând tipul atomic.Int64. Se poate incrementa variabila atomic și se poate citi valoarea acesteia prin metodele Add și Load. De asemenea, se poate stoca o valoare cu metoda Store, se poate înlocui valoarea cu metoda Swap, și se poate înlocui valoarea condiționat după o comparație folosind metoda CompareAndSwap.
cu slice
Dacă este utilizat împreună cu un slice, waitgroup poate deveni un instrument excelent pentru gestionarea operațiunilor concurente fără a necesita blocări (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}
Codul de mai sus folosește exclusiv waitgroup pentru a permite fiecărei goroutine să genereze concomitent 10 numere întregi aleatorii și să le stocheze la indexul alocat. În acest cod, waitgroup este folosit pentru a aștepta finalizarea tuturor goroutine-urilor, după care se afișează Done. Utilizând waitgroup în acest mod, se poate gestiona simultan lucrul efectuat de multiple goroutine-uri, stocarea datelor fără blocări și efectuarea procesării ulterioare în bloc după terminarea tuturor sarcinilor.
golang.org/x/sync/errgroup.ErrGroup
errgroup este o extensie a pachetului sync.WaitGroup. Spre deosebire de sync.WaitGroup, errgroup anulează toate goroutine-urile și returnează o eroare dacă una dintre sarcinile goroutine-urilor eșuează.
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}
Codul de mai sus utilizează errgroup pentru a crea 10 goroutine-uri, provocând o eroare în a cincea goroutine. Intenționat s-a generat o eroare în a cincea goroutine pentru a ilustra cazul de eșec. Totuși, în utilizarea practică, se generează goroutine-uri cu errgroup și se realizează diverse procesări ulterioare pentru cazurile în care apar erori în cadrul fiecărei goroutine.
once
Acesta este un instrument utilizat pentru a executa cod care trebuie rulat o singură dată. Codul aferent poate fi executat prin constructorii următori.
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 asigură că funcția respectivă poate fi executată o singură dată pe parcursul întregii execuții.
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}
Codul de mai sus utilizează sync.OnceFunc pentru a afișa Hello, World!. În acest cod, funcția once este creată prin sync.OnceFunc, iar chiar dacă este apelată de mai multe ori, Hello, World! va fi afișat o singură dată.
OnceValue
OnceValue nu numai că asigură execuția funcției o singură dată pe parcursul întregii execuții, dar stochează și valoarea de return a acelei funcții pentru a o returna la apelări ulterioare.
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}
Codul de mai sus incrementează variabila c cu 1 utilizând sync.OnceValue. În acest cod, funcția once este creată prin sync.OnceValue, iar chiar dacă este apelată de mai multe ori, variabila c va fi incrementată o singură dată, iar valoarea returnată va fi 1.
OnceValues
OnceValues funcționează identic cu OnceValue, dar permite returnarea mai multor 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}
Codul de mai sus incrementează variabila c cu 1 utilizând sync.OnceValues. În acest cod, funcția once este creată prin sync.OnceValues, iar chiar dacă este apelată de mai multe ori, variabila c va fi incrementată o singură dată, iar valoarea returnată va fi 1 pentru ambele componente.
atomic
Pachetul atomic furnizează operațiuni atomice. Pachetul atomic oferă metode precum Add, CompareAndSwap, Load, Store și Swap, dar recent se recomandă utilizarea tipurilor precum Int64, Uint64 sau 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}
Acesta este exemplul utilizat anterior. Acest cod incrementează atomic variabila c utilizând tipul atomic.Int64. Se poate incrementa variabila atomic și se poate citi valoarea acesteia prin metodele Add și Load. De asemenea, se poate stoca o valoare cu metoda Store, se poate înlocui valoarea cu metoda Swap, și se poate înlocui valoarea condiționat după o comparație folosind metoda CompareAndSwap.
cond
sync.Cond
Pachetul cond furnizează variabile de condiție. Pachetul cond poate fi creat ca sync.Cond și oferă metodele Wait, Signal și 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}
Codul de mai sus este unul care utilizează sync.Cond pentru a aștepta până când variabila ready devine true. În acest cod, se așteaptă până când variabila ready devine true prin utilizarea sync.Cond, după care se afișează Ready!. Prin utilizarea sync.Cond în acest mod, se poate determina ca multiple goroutine-uri să aștepte simultan până la îndeplinirea unei anumite condiții.
Această funcționalitate poate fi utilizată pentru a implementa o coadă simplă (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}
Prin utilizarea sync.Cond în acest mod, în loc să se consume intensiv resursele CPU cu spin-lock, se poate aștepta eficient și se poate relua execuția atunci când condiția este îndeplinită.
semaphore
golang.org/x/sync/semaphore.Semaphore
Pachetul semaphore furnizează semafoare. Pachetul semaphore poate fi creat ca golang.org/x/sync/semaphore.Semaphore și oferă metodele Acquire, Release și 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}
Codul de mai sus creează un semafor utilizând pachetul semaphore și demonstrează obținerea semaforului prin metoda Acquire și eliberarea acestuia prin metoda Release. În acest cod, s-a ilustrat modul de achiziționare și eliberare a unui semafor.
Concluzie
Cred că informațiile de bază prezentate sunt suficiente. Sperăm că, pe baza conținutului acestui articol, veți reuși să înțelegeți modul de gestionare a concurenței utilizând goroutine-uri și să le aplicați în practică. Sperăm că acest articol v-a fost de folos. Vă mulțumim.