GoSudaŠtartovací balíček pre súbežnosť v Go
Prehľad
Krátky úvod
Jazyk Go poskytuje mnoho nástrojov na správu súbežnosti. V tomto článku predstavím niektoré z nich a niekoľko trikov.
Goroutines?
Goroutine je nová forma modelu súbežnosti podporovaného jazykom Go. Program zvyčajne získa od OS vlákna OS, aby mohol súčasne vykonávať viacero úloh, a paralelne ich vykonáva v počte zodpovedajúcom počtu jadier. Na dosiahnutie súbežnosti v menšom rozsahu sa v používateľskom priestore (userland) vytvárajú zelené vlákna (green threads), ktoré v rámci jedného vlákna OS striedavo vykonávajú úlohy. Goroutine však túto formu zelených vlákien navrhli ešte menšie a efektívnejšie. Goroutine využívajú menej pamäte ako vlákna a možno ich vytvárať a prepínať rýchlejšie než vlákna.
Na používanie goroutine stačí jednoducho použiť kľúčové slovo go. To umožňuje intuitívne spúšťať synchrónny kód ako asynchrónny počas písania programu.
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}
Tento kód jednoducho mení synchrónny kód, ktorý po sekunde pozastavenia vypíše Hello, World!, na asynchrónny tok. Hoci je tento príklad jednoduchý, pri premene zložitejšieho kódu zo synchrónneho na asynchrónny sa čitateľnosť, prehľadnosť a zrozumiteľnosť kódu oproti doterajším spôsobom, ako sú async await alebo promise, zlepšia.
Vo veľa prípadoch sa však bez pochopenia jednoduchého asynchrónneho volania takéhoto synchrónneho kódu a tokov typu fork & join (podobných rozdeľuj a panuj) môže vytvoriť zlý kód goroutine. Na riešenie takýchto prípadov predstavím v tomto článku niekoľko metód a techník.
Správa súbežnosti
context
Že ako prvú techniku správy uvádzam context, môže byť prekvapujúce. V jazyku Go však context presahuje jednoduchú funkciu zrušenia a výrazne sa uplatňuje pri správe celého stromu úloh. Pre tých, ktorí tento balík nepoznajú, ho stručne vysvetlím.
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}
Uvedený kód pomocou context vypíše Context is done! po jednej sekunde. Stav zrušenia možno zistiť metódou Done() a metódy ako WithCancel, WithTimeout, WithDeadline alebo WithValue poskytujú rôzne spôsoby zrušenia.
Vytvorme jednoduchý príklad. Predpokladajme, že na získanie dát používate vzor aggregator a píšete kód, ktorý získava user, post a comment. Ak musia všetky požiadavky prebehnúť do 2 sekúnd, môžete napísať nasledujúci kód.
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}
Uvedený kód vypíše Timeout!, ak sa všetky dáta nepodarí získať do 2 sekúnd, a ak sa všetky dáta získajú, vypíše All data is fetched!. Týmto spôsobom možno pomocou context jednoducho spravovať zrušenie a časové limity aj v kóde, ktorý spúšťa viacero goroutine.
Rôzne funkcie a metódy súvisiace s context možno nájsť na stránke godoc context. Dúfam, že si jednoduché veci osvojíte a budete ich môcť pohodlne používať.
channel
unbuffered channel
channel je nástroj na komunikáciu medzi goroutine. channel možno vytvoriť pomocou make(chan T). Pritom T je typ dát, ktoré daný channel prenáša. Dáta sa posielajú a prijímajú pomocou <- a channel možno zatvoriť pomocou 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}
Uvedený kód pomocou channel vypíše 1 a 2. Tento kód ukazuje iba jednoduché posielanie a prijímanie hodnôt cez channel, no channel poskytuje viac funkcií. Najprv si priblížim buffered channel a unbuffered channel. Predtým však treba uviesť, že uvedený príklad používa unbuffered channel, pri ktorom sa musí odoslanie dát do kanála a ich prijatie uskutočniť súčasne. Ak sa tieto akcie nevykonajú súčasne, môže dôjsť k deadlocku.
buffered channel
Čo ak uvedený kód nevykonáva jednoduchý výpis, ale dva procesy s náročnou úlohou? Ak druhý proces prijíma dáta a dlhodobo sa zdržiava spracovaním, zastaví sa aj prvý proces na tento čas. Aby sme tomu zabránili, môžeme použiť 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}
Uvedený kód pomocou buffered channel vypíše 1 a 2. V tomto kóde sa použitím buffered channel dosiahlo, že odoslanie dát do channel a ich prijatie nemusia prebiehať súčasne. Keď má kanál buffer, vzniká rezerva zodpovedajúca jeho dĺžke, vďaka čomu možno zabrániť oneskoreniu úloh spôsobenému vplyvom neskorších úloh.
select
Pri práci s viacerými kanálmi možno pomocou príkazu select jednoducho implementovať štruktúru 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}
Uvedený kód vytvára tri kanály, ktoré periodicky posielajú hodnoty 1, 2 a 3, a pomocou select prijíma hodnoty z kanálov a vypisuje ich. Týmto spôsobom možno pomocou select súčasne prijímať dáta z viacerých kanálov a spracovávať ich podľa toho, ako prichádzajú.
for range
channel možno jednoducho používať na prijímanie dát pomocou for range. Keď sa for range použije na kanál, vykoná sa vždy, keď sa do kanála pridajú dáta, a keď sa kanál zatvorí, slučka sa ukončí.
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}
Uvedený kód pomocou channel vypíše 1 a 2. V tomto kóde sa pomocou for range pri každom pridaní dát do kanála dáta prijmú a vypíšu. Keď sa kanál zatvorí, slučka sa ukončí.
Ako som uviedol vyššie, túto syntax možno použiť aj ako jednoduchý synchronizačný prostriedok.
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}
Uvedený kód po sekunde pozastavenia vypíše Hello, World!. V tomto kóde sa pomocou channel zmenil synchrónny kód na asynchrónny. Týmto spôsobom možno pomocou channel jednoducho previesť synchrónny kód na asynchrónny a nastaviť bod join.
etc
- Ak sa do nil kanála pošlú alebo z neho prijmú dáta, môže sa dostať do nekonečnej slučky a vzniknúť deadlock.
- Ak sa po zatvorení kanála pošlú dáta, vznikne panic.
- Ani keď kanál explicitne nezatvoríte, GC ho pri uvoľňovaní pamäte zatvorí.
mutex
spinlock
spinlock je synchronizačná metóda, ktorá v cykle opakovane pokúša získať zámok. V jazyku Go možno spinlock jednoducho implementovať pomocou ukazovateľov.
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}
Uvedený kód implementuje balík spinlock. V tomto kóde sa SpinLock implementuje pomocou balíka sync/atomic. Metóda Lock sa pokúša získať zámok pomocou atomic.CompareAndSwapUintptr a metóda Unlock zámok uvoľní pomocou atomic.StoreUintptr. Keďže sa tento spôsob bez prestávky opakovane pokúša získať zámok, až dokým ho nezíska, neustále spotrebúva CPU a môže uviaznuť v nekonečnej slučke. Preto sa spinlock odporúča používať iba na jednoduchú synchronizáciu alebo v prípadoch, keď sa čaká len krátko.
sync.Mutex
mutex je nástroj na synchronizáciu medzi goroutine. mutex poskytovaný balíkom sync ponúka metódy Lock, Unlock, RLock, RUnlock a ďalšie. mutex možno vytvoriť pomocou sync.Mutex a na zámok na čítanie/zápis možno použiť 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}
V uvedenom kóde sa dve goroutine takmer súčasne dostanú k rovnakej premennej count. Ak sa kód, ktorý pristupuje k premennej count, spraví kritickou sekciou pomocou mutex, zabráni sa súbežnému prístupu k premennej count. Potom tento kód vypíše 2 bez ohľadu na to, koľkokrát sa spustí.
sync.RWMutex
sync.RWMutex je mutex, ktorý umožňuje rozlišovať zámok na čítanie a zámok na zápis. Zámok na čítanie možno nastaviť a uvoľniť pomocou metód RLock a 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}
Uvedený kód implementuje ConcurrentMap pomocou sync.RWMutex. V tomto kóde metóda Get nastavuje zámok na čítanie a metóda Set nastavuje zámok na zápis, vďaka čomu možno bezpečne pristupovať k mape data a upravovať ju. Zámok na čítanie je potrebný preto, aby sa v prípadoch, keď je čítanie časté, nenastavoval zámok na zápis, ale iba zámok na čítanie, a aby mohlo viacero goroutine vykonávať čítanie súčasne. Vďaka tomu možno zvýšiť výkon v prípadoch, keď nedochádza k zmene stavu a zámok na zápis nie je potrebný.
fakelock
fakelock je jednoduchý trik, ktorý implementuje sync.Locker. Táto štruktúra poskytuje rovnaké metódy ako sync.Mutex, ale v skutočnosti nevykonáva žiadnu činnosť.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Uvedený kód implementuje balík fakelock. Tento balík implementuje sync.Locker a poskytuje metódy Lock a Unlock, no v skutočnosti nevykonáva žiadnu činnosť. Prečo je taký kód potrebný, vysvetlím pri vhodnej príležitosti.
waitgroup
sync.WaitGroup
sync.WaitGroup je nástroj, ktorý čaká, kým sa dokončia všetky úlohy goroutine. Poskytuje metódy Add, Done a Wait: metódou Add sa pridáva počet goroutine, metódou Done sa signalizuje, že úloha goroutine skončila, a metódou Wait sa čaká, kým sa dokončia úlohy všetkých goroutine.
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}
Uvedený kód pomocou sync.WaitGroup pripočítava hodnotu do premennej c súbežne zo 100 goroutine. V tomto kóde sa pomocou sync.WaitGroup počká, kým skončia všetky goroutine, a potom sa vypíše hodnota pripočítaná do premennej c. Pri jednoduchom fork & join niekoľkých úloh stačí použiť iba kanály, no pri fork & join veľkého množstva úloh je dobrou voľbou aj sync.WaitGroup.
with slice
Ak sa použije spolu so slice, waitgroup sa môže stať výborným nástrojom na správu súbežne vykonávaných úloh bez zámkov.
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}
Uvedený kód používa iba waitgroup, pričom každá goroutine súbežne vygeneruje 10 náhodných celých čísel a uloží ich na pridelený index. V tomto kóde sa pomocou waitgroup počká, kým skončia všetky goroutine, a potom sa vypíše Done. Týmto spôsobom možno pomocou waitgroup nechať viacero goroutine vykonávať úlohy súbežne, dáta ukladať bez zámkov, kým neskončia všetky goroutine, a po skončení úloh ich spracovať naraz.
golang.org/x/sync/errgroup.ErrGroup
errgroup je balík, ktorý rozširuje sync.WaitGroup. Na rozdiel od sync.WaitGroup errgroup pri vzniku chyby v čo i len jednej z úloh goroutine zruší všetky goroutine a vráti chybu.
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}
Uvedený kód pomocou errgroup vytvára 10 goroutine a v piatej goroutine spôsobuje chybu. Chybu v piatej goroutine som vyvolal úmyselne, aby som ukázal prípad, keď nastane chyba. Pri skutočnom používaní sa errgroup použije tak, že sa vytvoria goroutine a v každej goroutine sa pri vzniku chyby vykonajú rôzne následné spracovania.
once
Je to nástroj na jednorazové vykonanie kódu, ktorý sa má spustiť iba raz. Súvisiaci kód možno spustiť pomocou nasledujúcich konštruktorov.
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 jednoducho zabezpečí, že sa daná funkcia počas celého behu vykoná presne raz.
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}
Uvedený kód pomocou sync.OnceFunc vypíše Hello, World!. V tomto kóde sa pomocou sync.OnceFunc vytvorí funkcia once, a aj keď sa funkcia once zavolá viackrát, Hello, World! sa vypíše iba raz.
OnceValue
OnceValue nezabezpečuje iba to, že sa daná funkcia vykoná presne raz počas celého behu, ale uloží aj jej návratovú hodnotu a pri ďalšom volaní vráti uloženú hodnotu.
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}
Uvedený kód pomocou sync.OnceValue zvyšuje premennú c o 1. V tomto kóde sa pomocou sync.OnceValue vytvorí funkcia once, a aj keď sa funkcia once zavolá viackrát, premenná c sa zvýši iba raz a vráti sa hodnota 1.
OnceValues
OnceValues funguje rovnako ako OnceValue, ale môže vrátiť viacero hodnôt.
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}
Uvedený kód pomocou sync.OnceValues zvyšuje premennú c o 1. V tomto kóde sa pomocou sync.OnceValues vytvorí funkcia once, a aj keď sa funkcia once zavolá viackrát, premenná c sa zvýši iba raz a vráti sa hodnota 1.
atomic
Balík atomic poskytuje atomické operácie. Balík atomic ponúka metódy ako Add, CompareAndSwap, Load, Store a Swap, no v poslednej dobe sa odporúča používať typy ako Int64, Uint64 a 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}
Ide o príklad, ktorý sme použili skôr. Uvedený kód pomocou typu atomic.Int64 atomicky zvyšuje premennú c. Metódami Add a Load možno premennú atomicky zvýšiť a načítať. Ďalej možno metódou Store uložiť hodnotu, metódou Swap hodnotu vymeniť a metódou CompareAndSwap porovnať hodnotu a, ak vyhovuje, vymeniť ju.
cond
sync.Cond
Balík cond poskytuje podmienkové premenné. Balík cond možno vytvoriť pomocou sync.Cond a poskytuje metódy Wait, Signal a 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}
Uvedený kód pomocou sync.Cond čaká, kým premenná ready nadobudne hodnotu true. V tomto kóde sa pomocou sync.Cond počká, kým premenná ready nadobudne hodnotu true, a potom sa vypíše Ready!. Týmto spôsobom možno pomocou sync.Cond nechať viacero goroutine súbežne čakať, kým sa nesplní určitá podmienka.
Využitím tohto možno implementovať jednoduchú 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}
Ak sa takto použije sync.Cond, namiesto toho, aby sa spotrebúvalo veľa CPU pomocou spin-lock, možno efektívne čakať a po splnení podmienky znovu pokračovať v činnosti.
semaphore
golang.org/x/sync/semaphore.Semaphore
Balík semaphore poskytuje semafór. Balík semaphore možno vytvoriť pomocou golang.org/x/sync/semaphore.Semaphore a poskytuje metódy Acquire, Release a 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}
Uvedený kód vytvára semafór pomocou semaphore, získava ho metódou Acquire a uvoľňuje metódou Release. V tomto kóde sa ukázal spôsob získania a uvoľnenia semafóra pomocou semaphore.
Záver
Základný obsah by mal stačiť. Na základe tohto článku dúfam, že pochopíte, ako pomocou goroutine spravovať súbežnosť, a budete ju vedieť používať v praxi. Dúfam, že tento článok vám bude užitočný. Ďakujem.