Go Concurrency Gedistilleerd

Dit is een snelle opfrisser over Go concurrency, geen beginnersgids. Als je concurrency vanaf de basis wilt leren met praktische oefeningen, raad ik mijn andere boek aan: Gist of Go: Concurrency.

Dit boek is AI-vrij.

---

Goroutines

De basis van concurrency in Go zijn goroutines – functies die worden gestart met het go-trefwoord:

func main() {
    var wg sync.WaitGroup
    wg.Add(2)
    go func() {
        defer wg.Done()
        fmt.Println("worker 1")
    }()
    go func() {
        defer wg.Done()
        fmt.Println("worker 2")
    }()
    wg.Wait()
}
// Output:
// worker 2
// worker 1

De Go runtime beheert deze goroutines en verdeelt ze over operating system (OS) threads die draaien op CPU-cores. In vergelijking met OS-threads zijn goroutines lichtgewicht, waardoor je er honderden of duizenden kunt aanmaken.

Goroutines zijn volledig onafhankelijk. De main-functie is ook een goroutine, maar deze start impliciet wanneer het programma opstart. Wanneer main eindigt, worden ook de andere goroutines afgesloten.

In het bovenstaande voorbeeld gebruiken we een wait group (sync.WaitGroup) om te wachten tot de goroutines klaar zijn. Een wait group bevat een teller. Het aanroepen van Add(n) verhoogt deze met n, terwijl Done() deze met één verlaagt. Wait() blokkeert de aanroepende goroutine (in dit geval main) totdat de teller nul bereikt. Op deze manier wacht main tot beide workers klaar zijn voordat het programma afsluit.

WaitGroup.Go verhoogt automatisch de teller van de wait group, voert een functie uit in een goroutine en verlaagt de teller wanneer deze klaar is:

func main() {
    var wg sync.WaitGroup
    wg.Go(func() {
        fmt.Println("worker 1")
    })
    wg.Go(func() {
        fmt.Println("worker 2")
    })
    wg.Wait()
}
// Output:
// worker 2
// worker 1

Channels

Goroutines kunnen waarden naar elkaar doorgeven via channels. Een channel is als een raam waar de ene goroutine iets doorheen kan gooien en een andere het kan opvangen:

func main() {
    messages := make(chan string)
    go func() { messages <- "ping" }()
    msg := <-messages
    fmt.Println(msg)
}
// Output: ping

Het verzenden van een waarde via een channel is een synchrone operatie. Wanneer de verzendende goroutine een waarde naar het channel schrijft (ch <- val), blokkeert deze en wacht tot iemand die waarde ontvangt (<-ch). Pas dan gaat hij verder.

Output channel

Het retourneren van een output channel vanuit een functie en het vullen hiervan binnen een interne goroutine is een veelvoorkomend patroon in Go. Dit stelt de aanroeper in staat om waarden via het channel te ontvangen, terwijl de eigenlijke functie de controle behoudt:

func generate(start, stop int) chan int {
    out := make(chan int)
    go func() {
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

Het sluiten van een channel

Om lezers te signaleren dat alle data is verzonden, sluit de schrijvende goroutine het channel met close():

func generate(start, stop int) chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

De lezer controleert de status van het channel met een tweede waarde ("comma OK") bij het lezen:

func main() {
    in := generate(5, 10)
    for {
        num, ok := <-in
        if !ok {
            break
        }
        fmt.Print(num, " ")
    }
}
// Output: 5 6 7 8 9

Zolang het channel open is, ontvangt de lezer de volgende waarde en een true status. Als het channel gesloten is, ontvangt de lezer een nulwaarde (zero value) en een false status.

Een channel kan slechts één keer worden gesloten. Het opnieuw sluiten of schrijven naar een gesloten channel veroorzaakt een panic.

De enige reden om een channel te sluiten is om de lezers te laten weten dat alle data is verzonden. Als dit niet belangrijk is voor de lezers, hoef je het niet te sluiten. Wanneer een channel niet meer wordt gebruikt, zal de garbage collector van Go de resources vrijgeven, ongeacht of het gesloten is.

Channel-iteratie

range leest automatisch de volgende waarde uit het channel en controleert of het gesloten is. Als het channel gesloten is, verlaat range de loop:

func main() {
    nums := generate(5, 10)
    for n := range nums {
        fmt.Print(n, " ")
    }
}
// Output: 5 6 7 8 9

In tegenstelling tot range over een slice, retourneert range over een channel slechts één waarde, geen paar.

Directional channels

Je kunt jezelf beschermen tegen onbedoelde schrijf- of sluitfouten door de richting van het channel in te stellen. Channels kunnen zijn:

• chan (bidirectioneel): voor lezen en schrijven (standaard);
• chan<- (alleen verzenden): alleen voor schrijven;
• <-chan (alleen ontvangen): alleen voor lezen.

Je kunt niet lezen uit een send-only channel of schrijven naar een receive-only channel (en je kunt dit laatste ook niet sluiten).

Channels worden meestal geïnitialiseerd voor zowel lezen als schrijven, en worden als directioneel gespecificeerd in functieparameters. Go converteert een regulier channel automatisch naar een directioneel channel:

stream := make(chan int)
go func(in chan<- int) {
    in <- 42
}(stream)

func(out <-chan int) {
    fmt.Println(<-out)
}(stream)
// Output: 42

Buffered channels

Buffered channels werken als een FIFO-wachtrij met een buffer van vaste grootte voor het opslaan van waarden.

Zolang de buffer vrije ruimte heeft, blokkeert het schrijven naar het channel de goroutine niet. Evenzo blokkeert het lezen uit het channel de goroutine niet, zolang de buffer waarden bevat:

stream := make(chan int, 3)
stream <- 11
stream <- 12
stream <- 13
fmt.Println(<-stream)
fmt.Println(<-stream)
// Output: 11 12

Standaard is een channel unbuffered (buffergrootte is nul) als je geen grootte opgeeft. Buffered channels werken samen met de ingebouwde functies len() en cap():

stream := make(chan int, 3)
stream <- 11
fmt.Println(cap(stream), len(stream))
// Output: 3 1

Het lezen uit een gesloten buffered channel retourneert de waarden uit de buffer met een true status. Zodra alle waarden zijn opgehaald, retourneert het een nulwaarde en een false status, net als bij een regulier channel:

stream := make(chan int, 1)
stream <- 11
close(stream)
val, ok := <-stream
fmt.Println(val, ok) // 11 true
val, ok = <-stream
fmt.Println(val, ok) // 0 false

Nil channel

Net als elk type in Go hebben channels een nulwaarde, namelijk nil. Schrijven naar of lezen uit een nil channel blokkeert de goroutine onbepaald:

var stream chan int
go func() {
    // blokkeert voor altijd
    stream <- 1
}()
// blokkeert voor altijd
<-stream

Het sluiten van een nil channel veroorzaakt een panic:

var stream chan int
close(stream)
// panic: close of nil channel

Select

De select-instructie lijkt op een switch, maar is specifiek ontworpen voor channels. Dit is hoe het werkt:

1. Het controleert welke cases niet geblokkeerd zijn.
2. Als meerdere cases klaar zijn, wordt er willekeurig één gekozen voor uitvoering.
3. Als alle cases geblokkeerd zijn en er is een default case, dan wordt deze uitgevoerd.
4. Als alle cases geblokkeerd zijn en er is geen default case, wacht select tot er één klaar is.

select wordt gebruikt om de datastroom in pipelines te beheren:

// merge verzendt waarden van in1 en in2 naar het output channel.
func merge(in1, in2 <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for in1 != nil || in2 != nil {
            select {
            case val1, ok := <-in1:
                if ok { out <- val1 } else { in1 = nil }
            case val2, ok := <-in2:
                if ok { out <- val2 } else { in2 = nil }
            }
        }
    }()
    return out
}
// Bijvoorbeeld: we sturen 10..12 naar in1, 20..22 naar in2,
// en roepen merge(in1, in2) aan.
// Resultaat: 10 11 20 12 21 22 (volgorde kan variëren)

Om goroutines te annuleren:

// process modificeert waarden uit 'in' en stuurt ze naar 'out'
// totdat 'in' leeg is of 'cancel' gesloten wordt.
func process(cancel chan struct{}, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for val := range in {
            select {
            case out <- val*10:
            case <-cancel:
                fmt.Println("canceled")
                return
            }
        }
    }()
    return out
}
// Bijvoorbeeld: we sturen 11 en 12 naar 'in'
// en roepen vervolgens close(cancel) aan.
// Resultaat: 110 120 canceled

Voor niet-blokkerende operaties:

// multiplier retourneert een functie die de input met 10 vermenigvuldigt
// en naar het channel stuurt, of een fout retourneert als het channel druk is.
func multiplier(ch chan<- int) func(n int) error {
    return func(n int) error {
        select {
        case ch <- n*10:
            return nil
        default:
            return errors.New("busy")
        }
    }
}

func main() {
    nums := make(chan int, 1)
    multiply := multiplier(nums)
    err := multiply(11)
    fmt.Println(<-nums, err) // 110 <nil>
    err = multiply(12)
    fmt.Println(<-nums, err) // 120 <nil>
    err = multiply(13)
    err = multiply(14)
    fmt.Println(err)         // busy
}

Pipelines

Een pipeline is een reeks bewerkingen waarbij elke stap inputdata ontvangt, deze op een specifieke manier verwerkt en vervolgens outputt. De input en output van elke operatie is een channel.

Een typische pipeline ziet er als volgt uit:

• Reader: Leest inputdata uit een bestand, database of netwerk.
• N processors: Transformeren, filteren, aggregeren of verrijken data met behulp van externe bronnen.
• Writer: Schrijft de verwerkte data naar een bestand, database of netwerk.

func read[T any]() <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {
            // lees data van ergens
            data := // ...
            out <- data
        }
    }()
    return out
}

func process[T any](in <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for inData := range in {
            // verwerk de data
            outData := // ...
            out <- outData
        }
    }()
    return out
}

func write[T any](in <-chan T) <-chan struct{} {
    done := make(chan struct{})
    go func() {
        defer close(done)
        for data := range in {
            // schrijf de data
        }
    }()
    return done
}

Output channel

Een goroutine kan andere goroutines signaleren dat het werk is voltooid met behulp van een output channel:

func generate(start, stop int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

func main() {
    nums := generate(5, 10)
    for n := range nums {
        fmt.Print(n, " ")
    }
}
// Output: 5 6 7 8 9

Done channel

Als een goroutine geen resultaten hoeft te retourneren, kan het de voltooiing signaleren via een done channel:

func work() <-chan struct{} {
    done := make(chan struct{})
    go func() {
        defer close(done)
        fmt.Println("work done")
    }()
    return done
}

func main() {
    done := work()
    <-done
}
// Output: work done

Cancel channel

Om een goroutine vroegtijdig te beëindigen, kan een aanroepende goroutine een cancel channel gebruiken:

func generate(cancel chan struct{}, n int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 1; i <= n; i++ {
            select {
            case out <- i:
            case <-cancel:
                return
            }
        }
    }()
    return out
}

func main() {
    cancel := make(chan struct{})
    defer close(cancel)
    nums := generate(cancel, 10)
    fmt.Println(<-nums)
    fmt.Println(<-nums)
    fmt.Println(<-nums)
}
// Output: 1 2 3

Foutafhandeling

Er zijn drie benaderingen voor foutafhandeling in concurrent pipelines:

1. Return bij de eerste fout:

func process(in <-chan int) (<-chan int, <-chan error) {
    out := make(chan Answer)
    errc := make(chan error, 1)
    go func() {
        defer close(out)
        for n := range in {
            ans, err := fetchAnswer(n)
            if err != nil {
                errc <- err  // retourneer met fout
                return
            }
            out <- ans
        }
        errc <- nil          // retourneer met nil
    }()
    return out, errc
}

1. Gebruik een resultaat-type:

type Result struct {
    answer int
    err    error
}

func calculate(in <-chan int) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for n := range in {
            ans, err := fetchAnswer(n)
            out <- Result{ans, err}  // retourneer antwoord + fout
        }
    }()
    return out
}

1. Fouten apart verzamelen:

func calculate(in <-chan int, errc chan<- error) <-chan int {
    out := make(chan Answer)
    go func() {
        defer close(out)
        for n := range in {
            ans, err := fetchAnswer(n)
            if err == nil {
                out <- ans   // stuur antwoord
            } else {
                errc <- err  // of fout
            }
        }
    }()
    return out
}

Time

Naast het beheren van datum en tijd, biedt het time-pakket tools voor het beheren van tijdgevoelige operaties in concurrent programma's.

After

time.After() retourneert een channel dat aanvankelijk leeg is, maar een waarde ontvangt na de timeout-periode. Dit is nuttig voor het instellen van timeouts voor operaties:

// withTimeout voert een functie uit met een opgegeven timeout.
func withTimeout(timeout time.Duration, fn func()) error {
    done := make(chan struct{})
    go func() {
        defer close(done)
        fn()
    }()
    // blokkeert tot fn voltooid is of de timer verloopt
    select {
    case <-done:
        return nil
    case <-time.After(timeout):
        return errors.New("timeout")
    }
}

func main() {
    var err error
    // voltooit op tijd
    err = withTimeout(
        50*time.Millisecond,
        func() { fmt.Println("work done") },
    )
    fmt.Println("err =", err)
    // wordt geannuleerd bij timeout
    err = withTimeout(
        50*time.Millisecond,
        func() {
            time.Sleep(100 * time.Millisecond)
            fmt.Println("work done")
        },
    )
    fmt.Println("err =", err)
}
// Output:
// work done
// err = <nil>
// err = timeout

Timer

Een timer (time.Timer) is een structuur met een C channel waarnaar de huidige tijd wordt verzonden wanneer deze afgaat (verloopt). Timers zijn handig voor het plannen van toekomstige uitvoeringen:

done := make(chan struct{})
timer := time.NewTimer(50 * time.Millisecond)
go func() {
    eventTime := <-timer.C  // blokkeert 50ms
    fmt.Println("work done at", eventTime)
    close(done)
}()
<-done

Stop() stopt de timer en retourneert true als deze nog niet is verlopen, en false anders:

timer := time.NewTimer(50 * time.Millisecond)
go func() {
    eventTime := <-timer.C
    fmt.Println("work done at", eventTime)
}()
// na 10ms is de timer nog niet verlopen
time.Sleep(10 * time.Millisecond)
if timer.Stop() {
    fmt.Println("execution canceled")
} else {
    fmt.Println("too late to cancel")
}
// Output: execution canceled

Vaak is het handiger om de wrapper-functie time.AfterFunc() te gebruiken. Deze wacht gedurende duur d en voert vervolgens functie f uit:

done := make(chan struct{})
work := func() {
    fmt.Println("work done")
    close(done)
}
// voert work uit na 50ms
time.AfterFunc(50*time.Millisecond, work)
<-done

time.AfterFunc() retourneert een timer die je kunt annuleren voordat de uitvoering begint:

timer := time.AfterFunc(50*time.Millisecond, func() {})
time.Sleep(10 * time.Millisecond)
if timer.Stop() {
    fmt.Println("execution canceled")
}

Als een timer in een loop wordt gebruikt, is het beter om één enkele timer aan te maken en deze te resetten in plaats van bij elke iteratie een nieuwe instantie aan te maken:

func consumer(in <-chan token) {
    const timeout = time.Hour
    timer := time.NewTimer(timeout)
    for {
        timer.Reset(timeout)
        select {
        case <-in:
            // doe iets
        case <-timer.C:
            // log waarschuwing
        }
    }
}

Ticker

Een ticker is als een timer, maar deze blijft afgaan totdat je hem stopt. Tickers zijn handig voor het uitvoeren van periodieke taken:

// gaat elke 50ms af
ticker := time.NewTicker(50 * time.Millisecond)
defer ticker.Stop()
go func() {
    for {
        at := <-ticker.C
        fmt.Println("work done at", at)
    }
}()
time.Sleep(160*time.Millisecond)
ticker.Stop()

NewTicker(d) maakt een ticker die de huidige tijd naar channel C stuurt met interval d. Je moet de ticker uiteindelijk stoppen met Stop() om resources vrij te maken. Als de lezer van het channel de ticker niet kan bijhouden, zal de ticker ticks overslaan.

Context

Het hoofddoel van context is het annuleren van operaties, handmatig of via een timeout/deadline. De functie accepteert een context en gebruikt het Done() channel om te luisteren naar annuleringen:

func work(ctx context.Context) error {
    done := make(chan struct{})
    go func() {
        time.Sleep(50 * time.Millisecond)
        fmt.Println("work done")
        close(done)
    }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

Handmatig annuleren (context.Canceled fout):

func main() {
    ctx := context.Background()
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    done := make(chan struct{})
    go func() {
        err := work(ctx)
        fmt.Println("err =", err)
        close(done)
    }()
    time.Sleep(10 * time.Millisecond)
    cancel()
    <-done
}
// Output: err = context canceled

Annuleren via timeout (context.DeadlineExceeded fout):

func main() {
    ctx := context.Background()
    ctx, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
    defer cancel()
    done := make(chan struct{})
    go func() {
        err := work(ctx)
        fmt.Println("err =", err)
        close(done)
    }()
    <-done
}
// Output: err = context deadline exceeded

Annuleren via deadline (context.DeadlineExceeded fout):

func main() {
    ctx := context.Background()
    deadline := time.Now().Add(10 * time.Millisecond)
    ctx, cancel := context.WithDeadline(ctx, deadline)
    defer cancel()
    done := make(chan struct{})
    go func() {
        err := work(ctx)
        fmt.Println("err =", err)
        close(done)
    }()
    <-done
}
// Output: err = context deadline exceeded

Context is gelaagd. Een context-object is onveranderlijk (immutable). Om nieuwe eigenschappen toe te voegen, wordt een nieuwe (kind) context gemaakt op basis van de oude (ouder) context. De kortste timeout tussen de ouder- en kindcontext wint altijd. Een kindcontext kan de timeout van de ouder alleen inkorten, niet verlengen:

func main() {
    const dur100ms = 100 * time.Millisecond
    parentCtx, cancel := context.WithTimeout(context.Background(), dur100ms)
    defer cancel()

    const dur10ms = 10 * time.Millisecond
    childCtx, cancel := context.WithTimeout(parentCtx, dur10ms)
    defer cancel()

    err := work(childCtx)
    fmt.Println("err =", err)
}
// Output: err = context deadline exceeded

Meerdere annuleringen zijn veilig. Je kunt cancel() zo vaak aanroepen als je wilt; de eerste keer werkt, de rest wordt genegeerd.

Je kunt een aangepaste reden voor annulering opgeven met context.WithCancelCause(), context.WithTimeoutCause() en context.WithDeadlineCause(). Deze reden is toegankelijk via context.Cause():

ctx, cancel := context.WithCancelCause(context.Background())
cancel(errors.New("the night is dark"))
fmt.Println(context.Cause(ctx))
// Output: the night is dark

Je kunt een functie registreren die wordt uitgevoerd wanneer de context wordt geannuleerd met context.AfterFunc():

ctx, cancel := context.WithCancel(context.Background())
cleanup := func() { fmt.Println("cleanup") }
context.AfterFunc(ctx, cleanup)
cancel()
time.Sleep(10 * time.Millisecond)
// Output: cleanup

Context kan aanvullende informatie over een aanroep doorgeven via context.WithValue(), maar het is over het algemeen beter om dit te vermijden. Gebruik liever expliciete parameters of aangepaste structs.

Wait groups

Het sync.WaitGroup-type laat je wachten tot één of meerdere goroutines klaar zijn:

const n = 10
var wg sync.WaitGroup
wg.Add(n)
for range n {
    go func() {
        defer wg.Done()
        fmt.Print(".")
    }()
}
wg.Wait()
// Output: ..........

Een WaitGroup weet niets over de goroutines die hij beheert; hij werkt met een interne teller. wg.Add(1) verhoogt de teller, wg.Done() verlaagt deze, en wg.Wait() blokkeert de aanroepende goroutine tot de teller nul is.

De Go-methode combineert Add, het starten van een goroutine en Done:

var wg sync.WaitGroup
for range 10 {
    wg.Go(func() {
        fmt.Print(".")
    })
}
wg.Wait()

Alle methoden zijn veilig te gebruiken vanuit meerdere goroutines. Normaal gesproken vinden alle Add-aanroepen plaats vóór Wait, maar technisch gezien kunnen sommige Add-aanroepen ook na Wait (vanuit een andere goroutine) plaatsvinden. Je kunt Wait vanuit meerdere goroutines aanroepen; ze zullen allemaal blokkeren tot de teller nul bereikt.

Data races

Een data race ontstaat wanneer meerdere goroutines toegang hebben tot gedeelde data, en ten minste één daarvan de data wijzigt. We moeten de data tegen dit soort gelijktijdige toegang beschermen.

Een data race veroorzaakt niet altijd een runtime panic. Daarom biedt Go een speciale tool: de race detector. Deze kun je inschakelen met de -race vlag bij test, run, build en install.

var total int
var wg sync.WaitGroup
wg.Go(func() { total++ })
wg.Go(func() { total++ })
wg.Wait()
fmt.Println("total:", total)

Uitvoeren met: go run -race main.go zal een WARNING: DATA RACE genereren.

Channels zijn veilig voor gelijktijdig lezen en schrijven en veroorzaken geen data races.

Manieren om data races te voorkomen:

• Vermijd gelijktijdige datamodificatie (meestal door channels te gebruiken).
• Synchroniseer toegang met mutexes.
• Gebruik uitsluitend atomaire operaties.

Race conditions

Een race condition ontstaat wanneer een onvoorspelbare volgorde van operaties vanuit meerdere goroutines leidt tot een incorrecte systeemstatus:

withdraw := func(amount int) {
    if getBalance() < amount {
        return
    }
    time.Sleep(time.Millisecond)
    setBalance(getBalance() - amount)
}
setBalance(50)
var wg sync.WaitGroup
wg.Go(func() { withdraw(40) })
wg.Go(func() { withdraw(40) })
wg.Wait()
fmt.Println("balance:", getBalance())
// Output kan zijn: balance: -30

Als individuele operaties concurrent-safe zijn, vindt de race detector van Go geen problemen. Hierdoor worden race conditions niet gedetecteerd.

Je kunt onzekerheid in een concurrent omgeving niet volledig elimineren, maar je kunt een race condition voorkomen door een samengestelde operatie te beschermen met een mutex:

var mu sync.Mutex
withdraw := func(amount int) {
    mu.Lock()
    defer mu.Unlock()
    if getBalance() < amount {
        return
    }
    time.Sleep(time.Millisecond)
    setBalance(getBalance() - amount)
}
setBalance(50)
var wg sync.WaitGroup
wg.Go(func() { withdraw(40) })
wg.Go(func() { withdraw(40) })
wg.Wait()
fmt.Println("balance:", getBalance())
// Output: balance: 10

Compare-and-set

Soms kun je een race condition voorkomen zonder mutexes door een atomaire compare-and-set operatie te gebruiken:

• CompareAndSet(old, new any) bool: Wijzigt de waarde naar new als de huidige waarde gelijk is aan old.
• CompareAndSwap(old, new any) any: Wijzigt de waarde naar new als de huidige waarde gelijk is aan old en retourneert de oude waarde.
• CompareAndDelete(old any) bool: Verwijdert de waarde als de huidige waarde gelijk is aan old.

Het idee is altijd hetzelfde: controleer of de veronderstelde (oude) status overeenkomt met de realiteit; zo ja, wijzig de status naar new, zo nee, doe niets.

Mutexes

Het sync.Mutex-type beschermt gedeelde data en delen van je code tegen gelijktijdige toegang:

var total int
var mu sync.Mutex
var wg sync.WaitGroup
for range 100 {
    wg.Go(func() {
        mu.Lock()
        time.Sleep(time.Millisecond)
        total++
        mu.Unlock()
    })
}
wg.Wait()
// Output: total: 100

De mutex garandeert dat slechts één goroutine tegelijk de code tussen Lock() en Unlock() kan uitvoeren. Een mutex wordt gebruikt in deze situaties:

• Wanneer meerdere goroutines dezelfde data wijzigen.
• Wanneer één goroutine data wijzigt en anderen deze lezen.
• Als alle goroutines alleen data lezen, is een mutex niet nodig.

TryLock

De TryLock-methode probeert de mutex te vergrendelen. Als dit niet lukt, retourneert hij direct false in plaats van de goroutine te blokkeren.

RWMutex

Het sync.RWMutex-type maakt onderscheid tussen lezers en schrijvers:

• Lock / Unlock: Vergrendelt de mutex voor zowel lezen als schrijven.
• RLock / RUnlock: Vergrendelt de mutex alleen voor lezen.

Dit werkt als volgt:

• Als een goroutine Lock() gebruikt, worden andere goroutines geblokkeerd bij Lock() of RLock().
• Als een goroutine RLock() gebruikt, kunnen andere goroutines ook RLock() gebruiken zonder te worden geblokkeerd.
• Als ten minste één goroutine RLock() heeft, worden andere goroutines geblokkeerd bij Lock().

Dit creëert een "single writer, multiple readers" opstelling.

Locker

Zowel sync.Mutex als sync.RWMutex implementeren de sync.Locker interface:

type Locker interface {
    Lock()
    Unlock()
}

Door Locker te gebruiken in plaats van een specifiek mutex-type, kun je componenten bouwen die niet afhankelijk zijn van een specifieke lock-implementatie.

Channel als mutex

Je kunt ook een channel gebruiken als mutex om gedeelde data te beschermen:

var total int
lock := make(chan struct{}, 1)
var wg sync.WaitGroup
wg.Go(func() {
    lock <- struct{}{}
    defer func() { <-lock }()
    total++
})
wg.Go(func() {
    lock <- struct{}{}
    defer func() { <-lock }()
    total++
})
wg.Wait()
// Output: total: 2

Semaphores

Een semafoor is als een container met N beschikbare slots en twee operaties: acquire om een slot te nemen en release om een slot vrij te geven.

• acquire neemt een vrij slot. Als er geen slots zijn, blokkeert de goroutine.
• release maakt een slot vrij. Als er goroutines blokkeerden op acquire, zal er één direct het slot innemen.

Je kunt een eenvoudige semafoor implementeren met een buffered channel:

const nConc = 4
const nCalls = 100
sema := make(chan struct{}, nConc)
var wg sync.WaitGroup
for range nCalls {
    sema <- struct{}{} // acquire
    wg.Go(func() {
        defer func() { <-sema }() // release
        time.Sleep(time.Millisecond)
    })
}
wg.Wait()

Voor complexere situaties kan het pakket golang.org/x/sync/semaphore worden gebruikt.

Rendezvous

Een rendezvous laat twee goroutines op elkaar wachten:

1. Twee goroutines (G1 en G2) kunnen signaleren dat ze klaar zijn.
2. Als G1 signaleert maar G2 nog niet, blokkeert G1.
3. Als G2 signaleert maar G1 nog niet, blokkeert G2.
4. Wanneer beiden hebben gesignaleerd, gaan ze allebei verder.

Dit kan worden geïmplementeerd met een wait group:

var rend sync.WaitGroup
rend.Add(2)
var wg sync.WaitGroup
wg.Go(func() {
    fmt.Println("before rendezvous")
    rend.Done()
    rend.Wait()
    fmt.Println("after rendezvous")
})
wg.Go(func() {
    fmt.Println("before rendezvous")
    rend.Done()
    rend.Wait()
    fmt.Println("after rendezvous")
})
wg.Wait()

Barrier

Een barrier is een algemenere vorm van een rendezvous en laat N goroutines op elkaar wachten:

1. De barrier heeft een teller en een drempelwaarde N.
2. Elke goroutine die de barrier bereikt, verhoogt de teller met 1.
3. De barrier blokkeert elke goroutine die hem bereikt.
4. Zodra de teller N bereikt, worden alle wachtende goroutines ontblokkeerd.

Signaling

Het sync.Cond-type (conditionele variabele) laat één goroutine signaleren aan een andere dat hij klaar is, en laat de andere goroutine wachten op dat signaal. Een Cond bevat een mutex en heeft twee methoden:

• Wait: Ontgrendelt de mutex en zet de goroutine in rust tot er een signaal komt. Bij het ontwaken wordt de mutex opnieuw vergrendeld.
• Signal: Wekt één goroutine die wacht op Wait.
• Broadcast: Wekt alle goroutines die wachten op Wait.

cond := sync.NewCond(&sync.Mutex{})
done := false
var wg sync.WaitGroup
wg.Go(func() {
    cond.L.Lock()
    fmt.Println("G1 is ready to signal")
    done = true
    cond.Signal()
    cond.L.Unlock()
})
wg.Go(func() {
    cond.L.Lock()
    for !done {
        cond.Wait()
    }
    fmt.Println("G2 received the signal")
    cond.L.Unlock()
})
wg.Wait()

Je kunt ook signaleren met een channel:

• Signal: signal := make(chan struct{}, 1) → signal <- struct{}{}.
• Broadcast: broadcast := make(chan struct{}) → close(broadcast).

Voor een vollediger publish/subscribe systeem kun je een structuur bouwen met een lijst van subscription channels en een mutex voor statusbescherming.

Run once

Het sync.Once-type zorgt ervoor dat een gegeven functie slechts één keer wordt uitgevoerd. Als meerdere goroutines gelijktijdig Once.Do aanroepen, voert er slechts één de functie uit, terwijl de anderen wachten tot deze retourneert:

total := 0
initState := func() {
    total += 1
}
var once sync.Once
var wg sync.WaitGroup
wg.Go(func() {
    once.Do(initState)
})
wg.Go(func() {
    once.Do(initState)
})
wg.Wait()
// Output: total: 1

Naast het Once-type biedt het sync-pakket ook:

• OnceFunc(f func()) func(): Retourneert een functie die f slechts één keer aanroept.
• OnceValue[T any](f func() T) func() T: Retourneert een functie die f één keer aanroept en de waarde retourneert.
• OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2): Idem voor twee retourwaarden.

Object pool

Het sync.Pool-type helpt bij het hergebruiken van geheugen in plaats van het telkens opnieuw te alloceren, wat de belasting op de garbage collector vermindert:

pool := sync.Pool{
    New: func() any {
        buf := make([]byte, 1024)
        return &buf
    },
}

var wg sync.WaitGroup
for range 4 {
    wg.Go(func() {
        for range 1000 {
            buf := pool.Get().(*[]byte)
            sink = buf
            pool.Put(buf)
        }
    })
}
wg.Wait()

Belangrijke punten:

• New moet een pointer retourneren om geheugen-copying en extra allocaties te vermijden.
• De pool heeft geen groottebeperking.
• Na het teruggeven van een item via Put, mag je het item niet meer gebruiken.

Atomics

Een operatie zonder synchronisatie kan alleen echt atomair zijn als deze vertaald wordt naar een enkele processorinstructie. Deze operaties hebben geen locks nodig. In Go zijn deze te vinden in het sync/atomic pakket:

• Int32, Int64, Uint32, Uint64, Bool, Value, Pointer.

Elk atomair type biedt:

• Load: Leest de waarde.
• Store: Zet een nieuwe waarde.
• Swap: Zet een nieuwe waarde en retourneert de oude.
• CompareAndSwap: Zet een nieuwe waarde alleen als de huidige waarde overeenkomt met de verwachting.

Numerieke types bieden ook een Add-methode.

Let op: De compositie van atomaire operaties is niet atomair. Als je meerdere atomaire stappen moet uitvoeren als één geheel, moet je een mutex gebruiken om race conditions te voorkomen.

Testing

Als je concurrent programma gebruik maakt van channels of synchronisatiemethoden, kun je deze direct in je tests gebruiken:

func Test(t *testing.T) {
    got := <-Calc()
    if got != 42 {
        t.Errorf("got: %v; want: 42", got)
    }
}

Als er geen geschikte synchronisatie-handles zijn, kun je het synctest pakket gebruiken. synctest.Test start een geïsoleerde "bubble" met een fake clock. synctest.Wait blokkeert totdat alle goroutines in de bubble zijn beëindigd of duurzaam geblokkeerd zijn.

Hierdoor kunnen tijdgevoelige tests (zoals timeouts van 3 seconden) direct worden uitgevoerd zonder echt te wachten.

Scheduling

Op hardwareniveau voeren CPU-cores parallelle taken uit. Op OS-niveau is een thread de basiseenheid van uitvoering. Op Go-runtime niveau is een goroutine de basiseenheid.

De Go-runtime scheduler voert een vast aantal OS-threads uit (vaak één per CPU-core) en verdeelt de goroutines over deze threads.

Het algoritme in eenvoudige termen:

1. Is er een vrije thread? Wijs er een goroutine aan vanuit de wachtrij.
2. Raakt een goroutine geblokkeerd (bijv. bij een channel)? Zet hem terug in de wachtrij en wijs een andere goroutine toe.
3. Raakt een goroutine vast in een syscall? Start een nieuwe thread voor andere goroutines.
4. Controleer elke 10ms de draaiende goroutines; preempteer langlopende goroutines om starvation te voorkomen.

Het aantal threads wordt beheerd via de GOMAXPROCS omgevingsvariabele. Een goroutine start met ongeveer 2 KB geheugen voor zijn stack, wat kan groeien indien nodig.

Diagnostics

Voor troubleshooting in productie gebruiken we metrieken, profiling en tracing.

• Metrics: Tonen de prestaties van de runtime (bijv. heap-geheugen, GC-pauzes). Gebruik het runtime/metrics pakket of tools als Prometheus en OpenTelemetry.
• Profiling: Helpt begrijpen wat het programma doet en welke resources het gebruikt. Go gebruikt een sampling profiler. De meest gebruikte zijn CPU- en heap-profiles. Gebruik net/http/pprof en het go tool pprof commando.
• Tracing: Legt specifieke events vast, vooral rondom concurrency en geheugen. Gebruik go tool trace om resultaten te bekijken. "Flight recording" kan worden ingesteld om altijd een recente trace beschikbaar te hebben.

Slotbeschouwingen

We hebben de volgende Go-tools voor concurrent programmeren behandeld:

• Goroutines voor het parallel uitvoeren van taken.
• Channels en select als flexibele communicatiemiddelen.
• Timers en tickers voor tijdbeheer.
• Context voor het annuleren van operaties.
• Wait groups voor synchronisatie van goroutines.
• Mutexes om race conditions te voorkomen.
• Condition variables voor het signaleren van events.
• Once voor veilige eenmalige initialisatie.
• Pools om de belasting van de garbage collector te verminderen.
• Atomaire operaties.