GoSudaStartovní balíček pro souběžnost v Go
Přehled
Krátký úvod
Jazyk Go nabízí mnoho nástrojů pro správu souběžnosti. V tomto článku bychom vám rádi představili některé z nich a několik triků.
Goroutine?
goroutine je nový model souběžnosti podporovaný jazykem Go. Obecně program získává od OS vlákna OS, aby mohl souběžně provádět více úloh, a vykonává je paralelně až do počtu jader. Pro souběžnost menšího rozsahu se v uživatelském prostoru vytvářejí zelená vlákna (green threads), která v rámci jednoho vlákna OS střídavě provádějí práci. Goroutiny však tento typ zelených vláken ještě více zmenšily a zefektivnily. Goroutiny využívají méně paměti než vlákna a lze je vytvářet a přepínat rychleji než vlákna.
Pro použití goroutiny stačí jednoduše použít klíčové slovo go. To umožňuje při psaní programu intuitivně spouštět synchronní kód jako asynchronní.
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 představuje jednoduchý příklad, který mění synchronní kód, jenž počká 1 sekundu a poté vypíše Hello, World!, na asynchronní tok. Současný příklad je jednoduchý, ale při převodu složitějšího kódu ze synchronního na asynchronní se čitelnost, přehlednost a srozumitelnost kódu oproti dřívějším přístupům, jako jsou async await nebo promise, výrazně zlepšují.
Avšak v mnoha případech, pokud nerozumíme toku, kdy se takový synchronní kód jednoduše volá asynchronně, a toku podobnému fork & join (který se podobá rozděl a panuj), vzniká špatný kód s goroutinami. V tomto článku proto představíme několik metod a technik, jak se na tyto situace připravit.
Správa souběžnosti
context
Že jako první správní techniku uvádíme context, může být překvapivé. Jazyk Go však context nepoužívá pouze pro jednoduché rušení, ale výborně slouží ke správě celého stromu úloh. Pro ty, kdo tento balíček neznají, jej nyní stručně popíšeme.
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}
Výše uvedený kód pomocí context vypíše Context is done! po 1 sekundě. context umožňuje zjistit, zda byla operace zrušena, pomocí metody Done() a poskytuje různé způsoby rušení prostřednictvím metod WithCancel, WithTimeout, WithDeadline, WithValue a dalších.
Vytvoříme jednoduchý příklad. Předpokládejme, že pro získání dat použijete vzor aggregator a napíšete kód, který načítá user, post a comment. Pokud se všechny požadavky musí uskutečnit do 2 sekund, můžete napsat následující 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}
Výše uvedený kód vypíše Timeout!, pokud se nepodaří načíst všechna data do 2 sekund, a All data is fetched!, pokud se všechna data načtou. Použitím context tímto způsobem lze snadno spravovat rušení a časové limity i v kódu, kde běží více goroutin.
Různé funkce a metody související s context lze nalézt v godoc context. Doufáme, že si jednoduché věci osvojíte a budete je moci pohodlně používat.
channel
unbuffered channel
channel je nástroj pro komunikaci mezi goroutinami. channel lze vytvořit pomocí make(chan T). Zde T označuje typ dat, která daný channel přenáší. Data se posílají a přijímají pomocí <- a channel lze uzavřít pomocí 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}
Výše uvedený kód pomocí channel vypíše 1 a 2. Tento kód ukazuje pouze jednoduché odesílání a přijímání hodnot přes channel. channel však nabízí více funkcí. Nejprve se seznámíme s buffered channel a unbuffered channel. Před začátkem poznamenejme, že příklad uvedený výše používá unbuffered channel, u něhož musí akce odeslání dat do kanálu a akce jejich přijetí proběhnout současně. Pokud k těmto akcím nedojde současně, může nastat deadlock.
buffered channel
Co kdyby výše uvedený kód nebyl pouze pro výpis, ale představoval dva procesy provádějící náročné úlohy? Pokud druhý proces čte a zpracovává data a zdrží se na dlouhou dobu, bude po tu dobu zastaven i první proces. Abychom tomuto stavu zabránili, můžeme použít 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}
Výše uvedený kód pomocí buffered channel vypíše 1 a 2. V tomto kódu je použit buffered channel, takže akce odeslání dat do channel a akce jejich přijetí nemusí probíhat současně. Když kanál získá buffer, vznikne prostor o jeho délce, a tím lze předejít zpoždění úloh způsobenému vlivem následných úloh.
select
Při práci s více kanály lze pomocí konstrukce select snadno implementovat strukturu 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}
Výše uvedený kód vytváří tři kanály, které periodicky posílají hodnoty 1, 2 a 3, a pomocí select přijímá hodnoty z kanálů a vypisuje je. Pomocí select lze přijímat data z více kanálů současně a zpracovávat je v okamžiku, kdy kterýkoli kanál hodnotu dodá.
for range
channel lze snadno číst pomocí for range. Použijeme-li for range nad kanálem, provede se vždy, když je do kanálu přidán nový údaj, a cyklus skončí, jakmile je kanál uzavřen.
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}
Výše uvedený kód pomocí channel vypíše 1 a 2. V tomto kódu se pomocí for range přijímají data z kanálu a vypisují se vždy, když jsou přidána. Po uzavření kanálu cyklus skončí.
Jak jsme již uvedli, tuto konstrukci lze použít i jako jednoduchý synchronizační prostředek.
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}
Výše uvedený kód počká 1 sekundu a poté vypíše Hello, World!. V tomto kódu jsme pomocí channel převedli synchronní kód na asynchronní. Pokud takto použijeme channel, lze snadno převádět synchronní kód na asynchronní a nastavit bod join.
etc
- Odeslání nebo přijetí dat z nil channel může vést k nekonečné smyčce a deadlocku.
- Odeslání dat do uzavřeného kanálu způsobí panic.
- I když kanál explicitně neuzavřete, GC jej při úklidu uzavře.
mutex
spinlock
spinlock je synchronizační metoda, která v cyklu opakovaně zkouší získat zámek. V jazyce Go lze spinlock snadno implementovat pomocí ukazatelů.
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}
Výše uvedený kód implementuje balíček spinlock. V tomto kódu je SpinLock implementován pomocí balíčku sync/atomic. Metoda Lock se pokouší získat zámek pomocí atomic.CompareAndSwapUintptr a metoda Unlock zámek uvolňuje pomocí atomic.StoreUintptr. Tento přístup se nezastaví a zámek neustále zkouší získat, takže dokud zámek nezíská, neustále spotřebovává CPU, a může uváznout v nekonečné smyčce. Proto je spinlock vhodné používat pouze pro jednoduchou synchronizaci nebo pro velmi krátké časové úseky.
sync.Mutex
mutex je nástroj pro synchronizaci mezi goroutinami. mutex poskytovaný balíčkem sync nabízí metody Lock, Unlock, RLock, RUnlock a další. mutex lze vytvořit jako sync.Mutex a pro zámek pro čtení/zápis lze použít 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 uvedeném kódu se téměř současně k téže proměnné count dostanou dvě goroutiny. Pokud přístup k proměnné count obalíme mutexem jako kritickou sekci, lze zabránit souběžnému přístupu k proměnné count. Potom tento kód vždy vypíše 2, bez ohledu na to, kolikrát jej spustíte.
sync.RWMutex
sync.RWMutex je mutex, který umožňuje rozlišovat zámek pro čtení a zámek pro zápis. Zámek pro čtení lze získat a uvolnit pomocí metod 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}
Výše uvedený kód implementuje ConcurrentMap pomocí sync.RWMutex. V tomto kódu metoda Get získává zámek pro čtení a metoda Set zámek pro zápis, takže mapa data může být bezpečně čtena a upravována. Zámek pro čtení je potřeba proto, aby v případech, kdy převažují jednoduché čtecí operace, nebylo nutné zamykat pro zápis, ale pouze pro čtení, a aby více goroutin mohlo současně číst. Tím lze zvýšit výkon v případech, kdy nedochází ke změně stavu a zámek pro zápis není potřeba.
fakelock
fakelock je jednoduchý trik, jak implementovat sync.Locker. Tato struktura poskytuje stejné metody jako sync.Mutex, ale ve skutečnosti nic nedělá.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Výše uvedený kód implementuje balíček fakelock. Tento balíček implementuje sync.Locker a poskytuje metody Lock a Unlock, ve skutečnosti však neprovádí žádnou operaci. Proč takový kód potřebujeme, vysvětlíme při nejbližší příležitosti.
waitgroup
sync.WaitGroup
sync.WaitGroup je nástroj, který čeká, dokud neskončí všechny úlohy goroutin. Poskytuje metody Add, Done a Wait: metodou Add přidáváme počet goroutin, metodou Done signalizujeme, že úloha goroutiny skončila, a metodou Wait čekáme, dokud neskončí všechny úlohy goroutin.
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}
Výše uvedený kód pomocí sync.WaitGroup nechá 100 goroutin souběžně přičítat hodnoty k proměnné c. V tomto kódu se pomocí sync.WaitGroup počká, až skončí všechny goroutiny, a poté se vypíše hodnota přičtená k proměnné c. Pro jednoduché fork & join několika úloh stačí použít pouze kanály, ale pro fork & join velkého množství úloh je dobrou volbou také sync.WaitGroup.
with slice
Použijeme-li waitgroup společně se slicem, může se stát vynikajícím nástrojem pro správu souběžně prováděných úloh bez zámků.
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}
Výše uvedený kód pomocí samotného waitgroup nechá každou goroutinu současně vygenerovat 10 náhodných celých čísel a uložit je na přidělený index. V tomto kódu se pomocí waitgroup počká, až skončí všechny goroutiny, a poté se vypíše Done. Pomocí waitgroup tedy mohou více goroutin pracovat současně, data se ukládají bez zámků, dokud neskončí všechny goroutiny, a po dokončení práce lze provést hromadné dodatečné zpracování.
golang.org/x/sync/errgroup.ErrGroup
errgroup je balíček rozšiřující sync.WaitGroup. Na rozdíl od sync.WaitGroup errgroup v okamžiku, kdy některá z goroutin vrátí chybu, zruší všechny goroutiny a chybu vrátí.
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}
Výše uvedený kód pomocí errgroup vytváří 10 goroutin a v páté goroutině vyvolá chybu. Záměrně způsobíme chybu v páté goroutině, abychom ukázali situaci, kdy chyba nastane. V praxi se však errgroup použije tak, že se vytvoří goroutiny a pro každou z nich se provede různé dodatečné zpracování, pokud nastane chyba.
once
Jde o nástroj pro spuštění kódu, který má být proveden pouze jednou. Související kód lze spustit pomocí následujících konstruktorů.
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 zajistí, že daná funkce bude v celém průběhu programu spuštěna pouze jednou.
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}
Výše uvedený kód pomocí sync.OnceFunc vypisuje Hello, World!. V tomto kódu se pomocí sync.OnceFunc vytvoří funkce once, a i když je funkce once volána vícekrát, Hello, World! se vypíše jen jednou.
OnceValue
OnceValue nezajišťuje pouze to, že daná funkce bude spuštěna jen jednou, ale také ukládá návratovou hodnotu funkce a při opakovaném volání vrací uloženou 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}
Výše uvedený kód pomocí sync.OnceValue zvyšuje proměnnou c o 1. V tomto kódu se pomocí sync.OnceValue vytvoří funkce once, a i když je once volána vícekrát, proměnná c se zvýší jen jednou a vrací hodnotu 1.
OnceValues
OnceValues funguje stejně jako OnceValue, ale může vracet více hodnot.
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}
Výše uvedený kód pomocí sync.OnceValues zvyšuje proměnnou c o 1. V tomto kódu se pomocí sync.OnceValues vytvoří funkce once, a i když je once volána vícekrát, proměnná c se zvýší jen jednou a vrací hodnotu 1.
atomic
Balíček atomic poskytuje atomické operace. Nabízí metody Add, CompareAndSwap, Load, Store, Swap a další, v poslední době se však doporučuje používat typy Int64, Uint64, Pointer a podobně.
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}
Jde o příklad, který jsme použili dříve. Typ atomic.Int64 se používá k atomickému zvýšení proměnné c. Metody Add a Load slouží k atomickému zvyšování proměnné a jejímu čtení. Dále lze metodou Store uložit hodnotu, metodou Swap hodnotu vyměnit a metodou CompareAndSwap porovnat hodnotu a v případě shody ji nahradit.
cond
sync.Cond
Balíček cond poskytuje podmínkové proměnné. Lze je vytvořit pomocí sync.Cond a nabízí metody 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}
Výše uvedený kód pomocí sync.Cond čeká, dokud proměnná ready nenabyde hodnoty true. V tomto kódu se pomocí sync.Cond počká, dokud ready nenabyde hodnoty true, a poté se vypíše Ready!. Pomocí sync.Cond lze nechat více goroutin čekat, dokud není splněna určitá podmínka.
Využitím tohoto přístupu lze implementovat jednoduchou 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}
Pokud použijeme sync.Cond, místo toho, abychom spotřebovávali mnoho CPU pomocí spin-lock, můžeme efektivně čekat a po splnění podmínky pokračovat v činnosti.
semaphore
golang.org/x/sync/semaphore.Semaphore
Balíček semaphore poskytuje semafor. Lze jej vytvořit jako golang.org/x/sync/semaphore.Semaphore a nabízí metody 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}
Výše uvedený kód vytváří semafor pomocí semaphore, získává jej metodou Acquire a uvolňuje metodou Release. V tomto kódu jsme ukázali, jak pomocí semaphore získat a uvolnit semafor.
Závěr
Základní obsah by zde měl stačit. Na základě tohoto článku doufáme, že pochopíte, jak správu souběžnosti zajišťovat pomocí goroutin, a budete ji moci skutečně používat. Doufáme, že vám tento článek byl užitečný. Děkujeme.