--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Startovní balíček pro souběžnost v Go description: Představujeme různé nástroje a techniky pro správu souběžnosti v Go, včetně goroutin, kanálů, mutexů a dalších. language: cs date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## 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í. ```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 {} } ``` 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. ```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) } ``` 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. ```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!") } } ``` 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](https://pkg.go.dev/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`. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` 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`. ```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) } } ``` 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`. ```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) } } } ``` 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. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` 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. ```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 {} } ``` 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 1. Odeslání nebo přijetí dat z nil channel může vést k nekonečné smyčce a deadlocku. 2. Odeslání dat do uzavřeného kanálu způsobí panic. 3. 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ů. ```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{} } ``` 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`. ```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) } ``` 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`. ```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 } ``` 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á. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (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. ```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()) } ``` 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ů. ```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) } } ``` 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í. ```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) } } ``` 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ů. ```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` zajistí, že daná funkce bude v celém průběhu programu spuštěna pouze jednou. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` 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. ```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()) } ``` 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. ```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) } ``` 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ě. ```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()) } ``` 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`. ```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!") } ``` 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`. ```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 } ``` 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`. ```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) } ``` 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.