Model Go CSP

Idź objąć Komunikowanie procesów sekwencyjnych (CSP), model formalny zaproponowana przez Tony'ego Hoare'a w 1978. Podstawowa zasada: zamiast wielu goroutines które czytają i zapisują do tej samej zmiennej (z muteksem), goroutines komunikują się za pośrednictwem kanał — potoki typu, które przenoszą własność danych.

Środowisko wykonawcze Go zarządza planowaniem goroutines w puli wątków systemu operacyjnego (domyślnie nawet do liczby dostępnych rdzeni, kontrolowanych przez GOMAXPROCS). Gorutyna jest automatycznie zawieszane podczas operacji blokowania (we/wy, odbiór kanału), umożliwiając innym Goroutine, aby przejść dalej.

Goroutine: anatomia i cykl życia

package main

import (
    "fmt"
    "time"
)

func worker(id int, done chan struct{}) {
    fmt.Printf("Worker %d started\n", id)
    time.Sleep(100 * time.Millisecond)     // simula lavoro
    fmt.Printf("Worker %d finished\n", id)
    done <- struct{}{}                       // segnala completamento
}

func main() {
    done := make(chan struct{}, 5)  // channel bufferizzato per 5

    // Lancia 5 goroutine concorrentemente
    for i := 0; i < 5; i++ {
        go worker(i, done)           // "go" avvia la goroutine
    }

    // Aspetta che tutte completino
    for i := 0; i < 5; i++ {
        <-done                       // riceve dal channel (blocca se vuoto)
    }

    fmt.Println("All workers done")
}

// Costo di una goroutine: ~2-8KB stack (cresce dinamicamente fino a 1GB)
// vs ~1MB per un OS thread
// Go può avere milioni di goroutine attive contemporaneamente

Kanał: Rodzaje i wzorce

Kanał niebuforowany i buforowany

// Channel unbuffered: sincronizzazione diretta
// Send blocca finché un receiver è pronto
ch := make(chan int)         // unbuffered
go func() { ch <- 42 }()  // blocca finché qualcuno riceve
v := <-ch                   // sblocca il sender

// Channel buffered: coda FIFO con capacità fissata
// Send blocca SOLO se il buffer è pieno
buffered := make(chan int, 10)   // buffer da 10 elementi
buffered <- 1                   // non blocca (buffer ha spazio)
buffered <- 2                   // non blocca
v := <-buffered                 // riceve 1 (FIFO)

// Directional channel types: sicurezza a compile time
func producer(ch chan<- int) { // solo write
    ch <- 42
}

func consumer(ch <-chan int) { // solo read
    v := <-ch
    fmt.Println(v)
}

Wzór: rurociąg

// Pipeline: catena di goroutine connesse da channel
// Ogni fase legge dall'input channel e scrive sull'output channel

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)    // IMPORTANTE: chiudi quando finisci
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {  // range su channel riceve finché chiuso
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    // Compone la pipeline
    c := generate(2, 3, 4, 5)
    out := square(c)

    // Consuma l'output
    for v := range out {
        fmt.Println(v)  // 4, 9, 16, 25
    }
}

Wybierz: Multipleksowanie na wielu kanałach

select to najpotężniejsze słowo kluczowe w Go for the Competition: czekaj wielu kanałów jednocześnie i wykonuje przypadek gotowości kanału. Jeśli wiele kanałów są gotowe, wybiera losowo. 

// Timeout pattern con select
import "time"

func fetchWithTimeout(url string, timeout time.Duration) (string, error) {
    resultCh := make(chan string, 1)
    errCh := make(chan error, 1)

    go func() {
        result, err := fetch(url)
        if err != nil {
            errCh <- err
            return
        }
        resultCh <- result
    }()

    select {
    case result := <-resultCh:
        return result, nil
    case err := <-errCh:
        return "", err
    case <-time.After(timeout):
        return "", fmt.Errorf("timeout after %v", timeout)
    }
}

// Fan-out/fan-in con select e done channel
func merge(channels ...<-chan int) <-chan int {
    merged := make(chan int)
    var wg sync.WaitGroup

    multiplex := func(ch <-chan int) {
        defer wg.Done()
        for v := range ch {
            merged <- v
        }
    }

    wg.Add(len(channels))
    for _, ch := range channels {
        go multiplex(ch)
    }

    go func() {
        wg.Wait()
        close(merged)
    }()

    return merged
}

kontekst.Kontekst: Propagowane usuwanie

Pakiet context to standardowy mechanizm Go do propagowania usuwania poprzez łańcuch goroutin. Każda funkcja wykonująca operacje we/wy lub długą pracę powinna to zaakceptować a context.Context jako pierwszy parametr: 

import (
    "context"
    "fmt"
    "time"
)

// Funzione che rispetta la cancellazione
func longOperation(ctx context.Context, id int) error {
    for i := 0; i < 10; i++ {
        select {
        case <-ctx.Done():              // controlla se il context è cancellato
            return ctx.Err()           // errore: context.Canceled o DeadlineExceeded
        default:
            fmt.Printf("Step %d/%d\n", i+1, 10)
            time.Sleep(100 * time.Millisecond)
        }
    }
    return nil
}

func main() {
    // Context con timeout di 500ms
    ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
    defer cancel()  // SEMPRE defer cancel() per liberare le risorse

    err := longOperation(ctx, 1)
    if err != nil {
        fmt.Printf("Cancelled: %v\n", err)  // context deadline exceeded
    }
}

// Context si propaga attraverso le chiamate:
// Handler HTTP -> Service -> Repository -> Database
// Se il client disconnette, il context si cancella e risale tutta la catena

WaitGroup: wzorzec synchronizacji

import "sync"

func processAll(items []Item) {
    var wg sync.WaitGroup
    results := make([]Result, len(items))

    for i, item := range items {
        wg.Add(1)
        go func(i int, item Item) {   // passa i e item come parametri!
            defer wg.Done()
            results[i] = process(item) // scrivere indici diversi è safe
        }(i, item)
    }

    wg.Wait()           // blocca finché tutti Done() sono stati chiamati
    fmt.Println(results)
}

// ERRORE COMUNE: closure su variabile del loop (prima di Go 1.22)
// In Go 1.22+ il loop variable è scoped per iterazione -- no problema
for _, item := range items {
    go func() {
        process(item)    // Go 1.22+: safe. Go <1.22: BUG! usa go func(i Item) invece
    }()
}

Detektor wyścigów: znajdowanie danych o wyścigach

// Compila ed esegui con il race detector integrato
go run -race main.go
go test -race ./...

// Esempio di data race che il detector trova:
var counter int

func increment() {
    counter++  // DATA RACE! lettura + scrittura non atomica
}

// go run -race stamperà:
// WARNING: DATA RACE
// Write at 0x... by goroutine 6:
//   main.increment()
// Previous write at 0x... by goroutine 5:
//   main.increment()

// Fix con atomic o mutex:
import "sync/atomic"
var atomicCounter int64
atomic.AddInt64(&atomicCounter, 1)  // atomico, thread-safe

Wyciek Goroutine: najczęstszy błąd w Go

Gorutyna, która blokuje się na kanale odbiorczym (nikt nigdy nie wysyła) lub nie odbiera nigdy nie usuwanie kontekstu jest wyciekiem goroutine. Zgromadź pamięć i procesor. Zawsze używaj context w przypadku długich operacji i upewnij się, że każdy kanał ma nadawcę i odbiornika lub użyj kanałów buforowanych z jawnym czyszczeniem.

Wnioski

Model współbieżności Go jest jednym z najskuteczniejszych w przypadku usług backendowych: goroutine ultralekkie kanały, które z założenia zapobiegają warunkom wyścigowym, select do multipleksowania tj context za propagowane anulowanie. Większość konkurencyjnego kodu Go nie potrzebuje wyraźnych muteksów.

Następny artykuł dotyczy Pythona: asyncio.Grupa zadań i zorganizowaną konkurencję w Pythonie 3.11+, który wreszcie zapewnia światu bezpieczeństwo semantyczne porównywalne z CSP Go asynchronicznie/czekaj.

Poprzedni artykuł ← JavaScript pętli zdarzeń
Następny w serii Synchronizacja Pythona Zaawansowane →