--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Go-concurrentiestarterpack description: Deze tekst introduceert diverse hulpmiddelen en technieken voor concurrency-beheer in Go, waaronder goroutines, kanalen en mutexen. language: nl date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Overzicht ### Korte inleiding De Go-taal beschikt over veel hulpmiddelen voor het beheer van concurrency. In dit artikel willen we enkele daarvan en een aantal trucs introduceren. ### Goroutines? Een goroutine is een nieuw type concurrency-model dat wordt ondersteund door de Go-taal. Gewoonlijk verkrijgt een programma van het besturingssysteem OS-threads om meerdere taken tegelijkertijd uit te voeren, en voert het taken parallel uit, zoveel als er cores zijn. Om concurrency op een kleinere schaal uit te voeren, worden in de userland groene threads aangemaakt, zodat meerdere groene threads binnen één OS-thread om de beurt taken uitvoeren. Goroutines zijn echter een verdere verkleining en verbetering van dit soort groene threads. Goroutines gebruiken minder geheugen dan threads en kunnen sneller worden aangemaakt en gewisseld dan threads. Om goroutines te gebruiken, hoeft men alleen het `go`-sleutelwoord te gebruiken. Hierdoor kan synchrone code tijdens het schrijven van een programma op intuïtieve wijze als asynchrone code worden uitgevoerd. ```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 {} } ``` Deze code verandert eenvoudig een synchrone code, die één seconde wacht en vervolgens `Hello, World!` afdrukt, in een asynchrone stroom. Hoewel het huidige voorbeeld eenvoudig is, worden de leesbaarheid, zichtbaarheid en begrijpelijkheid van de code beter wanneer een iets complexere code van synchroon naar asynchroon wordt omgezet, vergeleken met bestaande methoden zoals async await of promise. In veel gevallen kunnen echter slechte goroutine-codes ontstaan wanneer men de stroom van het eenvoudig asynchroon aanroepen van synchrone code, en stromen zoals `fork & join` (die enigszins lijken op divide and conquer), niet begrijpt. Om op deze situaties voorbereid te zijn, zullen we in dit artikel enkele methoden en technieken introduceren. ## Concurrency-beheer ### context Het is misschien verrassend dat `context` als eerste beheertechniek naar voren komt. In de Go-taal speelt `context` echter, naast een eenvoudige annuleringsfunctie, een uitstekende rol bij het beheer van de volledige taakboom. Voor wie het pakket niet kent, zullen we het kort toelichten. ```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) } ``` De bovenstaande code gebruikt `context` om na één seconde `Context is done!` af te drukken. Met de methode `Done()` kan men controleren of een context is geannuleerd, en via methoden zoals `WithCancel`, `WithTimeout`, `WithDeadline` en `WithValue` worden diverse annuleringsmethoden geboden. Laten we een eenvoudig voorbeeld maken. Stel dat u een code schrijft die, met behulp van het `aggregator`-patroon, `user`, `post` en `comment` ophaalt. Stel verder dat alle verzoeken binnen twee seconden moeten zijn voltooid. Dan kunt u dit als volgt schrijven. ```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!") } } ``` De bovenstaande code drukt `Timeout!` af als niet alle gegevens binnen twee seconden zijn opgehaald, en `All data is fetched!` als alle gegevens zijn opgehaald. Op deze manier kan men met `context` annulering en time-outs eenvoudig beheren, zelfs in code waarin meerdere goroutines draaien. Diverse context-gerelateerde functies en methoden zijn te vinden op [godoc context](https://pkg.go.dev/context). We hopen dat u na het bestuderen van de eenvoudige onderdelen deze comfortabel kunt gebruiken. ### channel #### unbuffered channel Een `channel` is een hulpmiddel voor communicatie tussen goroutines. Een `channel` kan worden aangemaakt met `make(chan T)`. Hierbij is `T` het type gegevens dat het betreffende `channel` zal doorgeven. Met `<-` kunnen gegevens worden verzonden en ontvangen, en met `close` kan een `channel` worden gesloten. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` De bovenstaande code gebruikt een `channel` om 1 en 2 af te drukken. Deze code toont alleen het eenvoudig verzenden en ontvangen van waarden via een `channel`. Een `channel` biedt echter meer mogelijkheden. Laten we eerst `buffered channel` en `unbuffered channel` bekijken. Vooraf moet worden opgemerkt dat het hierboven getoonde voorbeeld een `unbuffered channel` is, waarbij het verzenden van gegevens naar het kanaal en het ontvangen van gegevens tegelijkertijd moeten plaatsvinden. Als deze handelingen niet gelijktijdig plaatsvinden, kan een deadlock ontstaan. #### buffered channel Hoe zit het als de bovenstaande code geen eenvoudige afdruk is, maar twee zware processen? Als het tweede proces, terwijl het gegevens leest en verwerkt, lange tijd vastloopt, dan zal het eerste proces gedurende die tijd eveneens stilvallen. Om deze situatie te voorkomen, kunnen we een `buffered channel` gebruiken. ```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) } } ``` De bovenstaande code gebruikt een `buffered channel` om 1 en 2 af te drukken. In deze code is een `buffered channel` gebruikt, zodat het verzenden van gegevens naar het `channel` en het ontvangen van gegevens niet tegelijkertijd hoeven plaats te vinden. Door een buffer in het kanaal aan te brengen, ontstaat ruimte ter grootte van die buffer, waardoor vertragingen als gevolg van de invloed van latere taken kunnen worden voorkomen. #### select Wanneer men met meerdere kanalen werkt, kan men met de `select`-syntaxis eenvoudig een `fan-in`-structuur implementeren. ```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) } } } ``` De bovenstaande code maakt drie kanalen die periodiek 1, 2 en 3 doorgeven, en gebruikt `select` om waarden uit de kanalen te ontvangen en af te drukken. Op deze manier kan men met `select` tegelijkertijd gegevens van meerdere kanalen ontvangen en elke waarde verwerken zodra die binnenkomt. #### for range Met `for range` kunnen gegevens van een `channel` eenvoudig worden ontvangen. Wanneer `for range` op een kanaal wordt gebruikt, wordt het uitgevoerd telkens wanneer er gegevens aan het kanaal worden toegevoegd, en wordt de lus beëindigd zodra het kanaal is gesloten. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` De bovenstaande code gebruikt een `channel` om 1 en 2 af te drukken. In deze code gebruikt `for range` om telkens wanneer gegevens aan het kanaal worden toegevoegd, de gegevens te ontvangen en af te drukken. Wanneer het kanaal wordt gesloten, wordt de lus beëindigd. Zoals eerder meermaals is getoond, kan deze syntaxis ook als eenvoudig synchronisatiemiddel worden gebruikt. ```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 {} } ``` De bovenstaande code wacht één seconde en drukt vervolgens `Hello, World!` af. In deze code is een `channel` gebruikt om synchrone code om te zetten in asynchrone code. Op deze manier kan men met een `channel` synchrone code eenvoudig in asynchrone code omzetten en een `join`-punt instellen. #### etc 1. Wanneer men gegevens naar of van een nil channel verzendt of ontvangt, kan men in een oneindige lus terechtkomen, wat tot een deadlock kan leiden. 2. Wanneer men gegevens verzendt nadat een kanaal is gesloten, treedt een panic op. 3. Het is niet noodzakelijk om een kanaal expliciet te sluiten, omdat de GC het kanaal zal opruimen en sluiten. ### mutex #### spinlock Een `spinlock` is een synchronisatiemethode die in een lus herhaaldelijk probeert het slot te verkrijgen. In de Go-taal kan men met behulp van pointers eenvoudig een spinlock implementeren. ```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{} } ``` De bovenstaande code implementeert het pakket `spinlock`. In deze code is `SpinLock` geïmplementeerd met het pakket `sync/atomic`. In de methode `Lock` wordt geprobeerd het slot te verkrijgen met `atomic.CompareAndSwapUintptr`, en in de methode `Unlock` wordt het slot vrijgegeven met `atomic.StoreUintptr`. Omdat deze methode zonder pauze het slot blijft proberen te verkrijgen, blijft de CPU gedurende het hele wachten op het slot in gebruik, wat tot een oneindige lus kan leiden. Daarom is het verstandig om `spinlock` alleen te gebruiken voor eenvoudige synchronisatie of voor zeer korte periodes. #### sync.Mutex Een `mutex` is een hulpmiddel voor synchronisatie tussen goroutines. De `mutex` die het pakket `sync` biedt, heeft methoden zoals `Lock`, `Unlock`, `RLock` en `RUnlock`. Een `mutex` kan worden aangemaakt met `sync.Mutex`, en met `sync.RWMutex` kan men ook een lees-/schrijfslot gebruiken. ```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) } ``` In de bovenstaande code hebben twee goroutines vrijwel tegelijkertijd toegang tot dezelfde variabele `count`. Door de code die toegang tot de variabele `count` heeft in een kritieke sectie te plaatsen met een `mutex`, kan gelijktijdige toegang tot `count` worden voorkomen. Hierdoor zal deze code, ongeacht hoe vaak deze wordt uitgevoerd, altijd `2` afdrukken. #### sync.RWMutex `sync.RWMutex` is een `mutex` waarmee men onderscheid kan maken tussen een leeslot en een schrijflot. Met de methoden `RLock` en `RUnlock` kan men een leeslot zetten en opheffen. ```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 } ``` De bovenstaande code implementeert een `ConcurrentMap` met `sync.RWMutex`. In deze code zet de methode `Get` een leeslot, en de methode `Set` een schrijflot, zodat de map `data` veilig kan worden benaderd en gewijzigd. Het leeslot is nodig omdat er, wanneer er veel eenvoudige leesbewerkingen zijn, alleen een leeslot wordt gezet zonder schrijflot, zodat meerdere goroutines tegelijkertijd leesbewerkingen kunnen uitvoeren. Hierdoor kan de prestatie worden verbeterd door alleen een leeslot te zetten in gevallen waarin de staat niet wordt gewijzigd en een schrijflot niet nodig is. #### fakelock `fakelock` is een eenvoudige truc die `sync.Locker` implementeert. Deze struct biedt dezelfde methoden als `sync.Mutex`, maar voert in werkelijkheid geen enkele handeling uit. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` De bovenstaande code implementeert het pakket `fakelock`. Dit pakket implementeert `sync.Locker` en biedt de methoden `Lock` en `Unlock`, maar voert in werkelijkheid geen enkele handeling uit. Waarom deze code nodig is, zal ik te zijner tijd toelichten. ### waitgroup #### sync.WaitGroup `sync.WaitGroup` is een hulpmiddel om te wachten tot alle taken van goroutines zijn voltooid. Het biedt de methoden `Add`, `Done` en `Wait`. Met de methode `Add` voegt men het aantal goroutines toe, met de methode `Done` meldt men dat de taak van een goroutine is voltooid, en met de methode `Wait` wacht men tot de taken van alle goroutines zijn voltooid. ```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()) } ``` De bovenstaande code gebruikt `sync.WaitGroup` zodat 100 goroutines tegelijkertijd een waarde optellen bij de variabele `c`. In deze code wacht `sync.WaitGroup` tot alle goroutines zijn voltooid, waarna de bij `c` opgetelde waarde wordt afgedrukt. Wanneer men slechts enkele taken met `fork & join` verwerkt, is het gebruik van alleen kanalen voldoende, maar bij het `fork & join`-verwerken van een grote hoeveelheid taken is `sync.WaitGroup` ook een goede keuze. #### with slice Wanneer het samen met een slice wordt gebruikt, kan `waitgroup` een uitstekend hulpmiddel worden voor het beheer van gelijktijdig uitgevoerde taken zonder locks. ```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) } } ``` De bovenstaande code gebruikt alleen `waitgroup` zodat elke goroutine tegelijkertijd een willekeurig geheel getal van 10 cijfers genereert en dit opslaat op de toegewezen index. In deze code wacht `waitgroup` tot alle goroutines zijn voltooid, waarna `Done` wordt afgedrukt. Op deze manier kunnen met `waitgroup` meerdere goroutines tegelijkertijd taken uitvoeren, gegevens zonder locks opslaan totdat alle goroutines zijn afgerond, en daarna gezamenlijk nabewerking uitvoeren. #### golang.org/x/sync/errgroup.ErrGroup `errgroup` is een pakket dat `sync.WaitGroup` uitbreidt. In tegenstelling tot `sync.WaitGroup` annuleert `errgroup`, wanneer bij een van de taken van de goroutines een fout optreedt, alle goroutines en geeft het de fout terug. ```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) } } ``` De bovenstaande code gebruikt `errgroup` om 10 goroutines aan te maken, waarbij de vijfde goroutine een fout veroorzaakt. Om te laten zien hoe een fout kan optreden, is opzettelijk in de vijfde goroutine een fout geïntroduceerd. In de praktijk gebruikt men `errgroup` om goroutines aan te maken en voert men, wanneer er in een goroutine een fout optreedt, diverse nabewerkingen uit. ### once Dit is een hulpmiddel om code uit te voeren die slechts één keer mag worden uitgevoerd. Via de onderstaande constructors kan de betreffende code worden uitgevoerd. ```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` zorgt er eenvoudigweg voor dat de betreffende functie over de gehele looptijd precies één keer kan worden uitgevoerd. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` De bovenstaande code gebruikt `sync.OnceFunc` om `Hello, World!` af te drukken. In deze code wordt met `sync.OnceFunc` de functie `once` aangemaakt, en ook wanneer `once` meerdere keren wordt aangeroepen, wordt `Hello, World!` slechts één keer afgedrukt. #### OnceValue `OnceValue` zorgt er niet alleen voor dat de betreffende functie over de gehele looptijd precies één keer wordt uitgevoerd, maar slaat ook de retourwaarde van de functie op en geeft bij een volgende aanroep de opgeslagen waarde terug. ```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()) } ``` De bovenstaande code gebruikt `sync.OnceValue` om de variabele `c` met 1 te verhogen. In deze code wordt met `sync.OnceValue` de functie `once` aangemaakt, en ook wanneer `once` meerdere keren wordt aangeroepen, wordt de variabele `c` slechts één keer verhoogd en wordt de waarde 1 teruggegeven. #### OnceValues `OnceValues` werkt op dezelfde manier als `OnceValue`, maar kan meerdere waarden teruggeven. ```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) } ``` De bovenstaande code gebruikt `sync.OnceValues` om de variabele `c` met 1 te verhogen. In deze code wordt met `sync.OnceValues` de functie `once` aangemaakt, en ook wanneer `once` meerdere keren wordt aangeroepen, wordt de variabele `c` slechts één keer verhoogd en wordt de waarde 1 teruggegeven. ### atomic Het pakket `atomic` is een pakket dat atomaire bewerkingen biedt. Het pakket `atomic` biedt methoden zoals `Add`, `CompareAndSwap`, `Load`, `Store` en `Swap`, maar tegenwoordig wordt het gebruik van typen zoals `Int64`, `Uint64` en `Pointer` aanbevolen. ```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()) } ``` Dit is het voorbeeld dat eerder werd gebruikt. Het gebruikt het type `atomic.Int64` om de variabele `c` atomair te verhogen. Met de methoden `Add` en `Load` kan de variabele atomair worden verhoogd en uitgelezen. Daarnaast kan met de methode `Store` een waarde worden opgeslagen, met de methode `Swap` een waarde worden vervangen, en met de methode `CompareAndSwap` een waarde worden vergeleken en, indien passend, worden vervangen. ### cond #### sync.Cond Het pakket `cond` is een pakket dat condition variables biedt. Het pakket `cond` kan worden aangemaakt met `sync.Cond` en biedt de methoden `Wait`, `Signal` en `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!") } ``` De bovenstaande code gebruikt `sync.Cond` om te wachten tot de variabele `ready` de waarde `true` krijgt. In deze code wacht `sync.Cond` tot de variabele `ready` de waarde `true` krijgt, waarna `Ready!` wordt afgedrukt. Op deze manier kan men met `sync.Cond` meerdere goroutines laten wachten totdat aan een bepaalde voorwaarde is voldaan. Hiermee kan een eenvoudige `queue` worden geïmplementeerd. ```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 } ``` Door `sync.Cond` te gebruiken, kan men in plaats van veel CPU-capaciteit te verbruiken met een `spin-lock`, efficiënt wachten en weer doorgaan zodra aan de voorwaarde is voldaan. ### semaphore #### golang.org/x/sync/semaphore.Semaphore Het pakket `semaphore` is een pakket dat semaforen biedt. Het pakket `semaphore` kan worden aangemaakt met `golang.org/x/sync/semaphore.Semaphore` en biedt de methoden `Acquire`, `Release` en `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) } ``` De bovenstaande code maakt met `semaphore` een semafoor aan, verkrijgt de semafoor met de methode `Acquire` en geeft deze weer vrij met de methode `Release`. In deze code is getoond hoe een semafoor wordt verkregen en vrijgegeven met `semaphore`. ## Tot slot Het lijkt erop dat de basisinhoud hiermee voldoende is. We hopen dat u op basis van de inhoud van dit artikel begrijpt hoe u goroutines gebruikt om concurrency te beheren, en dat u deze in de praktijk kunt toepassen. We hopen dat dit artikel nuttig voor u is geweest. Dank u wel.