MFormations
Modern Go Engineering

Chapitre 2

02 - Concurrence en Go

02 - Concurrence en Go

Cours 02 : Concurrence en Go

1. Goroutines

1.1 Concept et modèle M:N

Diagramme en cours de génération...

Le modèle M:N signifie que N goroutines sont multiplexées sur M threads OS.

// Lancement d'une goroutine
go func() {
    fmt.Println("Exécuté en parallèle")
}()

// Lancement avec paramètres
msg := "hello"
go func(s string) {
    fmt.Println(s)
}(msg)

1.2 Stack dynamique

func main() {
    // Une goroutine démarre avec ~2KB de stack
    // Le stack grandit/rétrécit automatiquement
    var stackSize int
    go func() {
        // Appels récursifs qui augmentent le stack
        var bigBuf [10000]byte
        _ = bigBuf
    }()
    
    // Vérifier la taille minimale de stack
    _ = stackSize
}

1.3 Goroutine lifecycle

func main() {
    done := make(chan bool)
    
    go func() {
        fmt.Println("Working...")
        time.Sleep(time.Second)
        done <- true
    }()
    
    <-done // Attend que la goroutine finisse
    fmt.Println("Done")
}

2. Channels

2.1 Création et types

// Channel non-buffered (synchrone)
ch := make(chan int)

// Channel buffered (asynchrone)
chBuf := make(chan string, 10)

// Channel en lecture seule
var readOnly <-chan int = ch

// Channel en écriture seule
var writeOnly chan<- int = ch

2.2 Opérations de base

func main() {
    ch := make(chan int)

    // Envoi (bloque jusqu'à réception)
    go func() {
        ch <- 42
    }()

    // Réception (bloque jusqu'à émission)
    val := <-ch
    fmt.Println(val)

    // Channel buffered
    bufCh := make(chan int, 3)
    bufCh <- 1
    bufCh <- 2
    bufCh <- 3
    // bufCh <- 4 // BLOCK: buffer plein

    fmt.Println(<-bufCh) // 1
    fmt.Println(<-bufCh) // 2
    fmt.Println(<-bufCh) // 3
}

2.3 Range et Close

func producer(ch chan<- int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch) // Important : le sender ferme le channel
}

func consumer(ch <-chan int) {
    for val := range ch { // Lit jusqu'à fermeture
        fmt.Printf("Reçu: %d\n", val)
    }
}

func main() {
    ch := make(chan int, 5)
    go producer(ch)
    consumer(ch)
}

2.4 Select

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "un"
    }()

    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "deux"
    }()

    select {
    case msg1 := <-ch1:
        fmt.Println("Reçu de ch1:", msg1)
    case msg2 := <-ch2:
        fmt.Println("Reçu de ch2:", msg2)
    case <-time.After(500 * time.Millisecond):
        fmt.Println("Timeout!")
    default:
        fmt.Println("Aucun channel prêt")
    }
}

2.5 Pattern de sélection multiple

func main() {
    ch := make(chan int, 1)
    
    for i := 0; i < 10; i++ {
        select {
        case ch <- i:
            // Envoyé
        default:
            // Channel plein, on passe
        }
    }
}

3. Patterns de concurrence

3.1 Generator Pattern

func fibonacci(n int) <-chan int {
    ch := make(chan int)
    go func() {
        a, b := 0, 1
        for i := 0; i < n; i++ {
            ch <- a
            a, b = b, a+b
        }
        close(ch)
    }()
    return ch
}

func main() {
    for fib := range fibonacci(10) {
        fmt.Println(fib)
    }
}

3.2 Fan-In (Multiplexage)

func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    
    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for val := range c {
                out <- val
            }
        }(ch)
    }
    
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

func main() {
    c1 := generator(1, 3, 5, 7, 9)
    c2 := generator(2, 4, 6, 8, 10)
    
    for val := range fanIn(c1, c2) {
        fmt.Println(val)
    }
}

3.3 Fan-Out (Distribution)

func fanOut(in <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    
    for i := 0; i < workers; i++ {
        out := make(chan int)
        go func(id int) {
            for val := range in {
                fmt.Printf("Worker %d traite %d\n", id, val)
                out <- val * 2
            }
            close(out)
        }(i)
        channels[i] = out
    }
    
    return channels
}

3.4 Pipeline Pattern

func pipeline(stages ...func(<-chan int) <-chan int) <-chan int {
    var ch <-chan int
    
    // Stage 0 : source
    ch = stages[0](nil)
    
    // Stages intermédiaires
    for _, stage := range stages[1:] {
        ch = stage(ch)
    }
    
    return ch
}

// Étapes du pipeline
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func filterOdd(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            if n%2 == 0 {
                out <- n
            }
        }
        close(out)
    }()
    return out
}

func main() {
    // Pipeline: generate → square → filterOdd
    for result := range pipeline(generate, square, filterOdd) {
        fmt.Println(result)
    }
}

3.5 Worker Pool

type Job struct {
    ID   int
    Data string
}

type Result struct {
    JobID int
    Output string
}

func worker(id int, jobs <-chan Job, results chan<- Result) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job.ID)
        results <- Result{
            JobID:  job.ID,
            Output: fmt.Sprintf("Processed: %s", job.Data),
        }
    }
}

func main() {
    const numJobs = 10
    const numWorkers = 3

    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)

    // Démarrer les workers
    for w := 1; w <= numWorkers; w++ {
        go worker(w, jobs, results)
    }

    // Envoyer les jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- Job{ID: j, Data: fmt.Sprintf("data-%d", j)}
    }
    close(jobs)

    // Collecter les résultats
    for r := 1; r <= numJobs; r++ {
        result := <-results
        fmt.Printf("Result: %+v\n", result)
    }
}

4. Synchronisation

4.1 sync.Mutex

type Counter struct {
    mu    sync.Mutex
    value int
}

func (c *Counter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value++
}

func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

func main() {
    var wg sync.WaitGroup
    counter := Counter{}

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Increment()
        }()
    }

    wg.Wait()
    fmt.Println(counter.Value()) // 1000
}

4.2 sync.RWMutex

type Cache struct {
    mu    sync.RWMutex
    data  map[string]any
}

func (c *Cache) Get(key string) (any, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    val, ok := c.data[key]
    return val, ok
}

func (c *Cache) Set(key string, value any) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = value
}

4.3 sync.WaitGroup

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("Worker %d démarré\n", id)
            time.Sleep(time.Duration(id) * 100 * time.Millisecond)
            fmt.Printf("Worker %d terminé\n", id)
        }(i)
    }

    wg.Wait()
    fmt.Println("Tous les workers terminés")
}

4.4 sync.Once

var (
    config   *Config
    configMu sync.Once
)

func GetConfig() *Config {
    configMu.Do(func() {
        fmt.Println("Initialisation de la configuration...")
        config = loadConfig()
    })
    return config
}

func main() {
    // Appels concurrents, une seule initialisation
    go GetConfig()
    go GetConfig()
    GetConfig()
}

4.5 sync.Cond

type Queue struct {
    items []int
    cond  *sync.Cond
}

func NewQueue() *Queue {
    return &Queue{
        items: make([]int, 0),
        cond:  sync.NewCond(&sync.Mutex{}),
    }
}

func (q *Queue) Enqueue(item int) {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    q.items = append(q.items, item)
    q.cond.Signal() // Réveille un goroutine en attente
}

func (q *Queue) Dequeue() int {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    for len(q.items) == 0 {
        q.cond.Wait() // Attend qu'un item soit disponible
    }
    item := q.items[0]
    q.items = q.items[1:]
    return item
}

4.6 sync.Pool

type BigBuffer struct {
    buf [1024]byte
}

var bufferPool = sync.Pool{
    New: func() any {
        return &BigBuffer{}
    },
}

func processRequest() {
    buf := bufferPool.Get().(*BigBuffer)
    defer bufferPool.Put(buf)
    
    // Utiliser buf
    // Pas besoin de réinitialiser, le pool gère
}

5. Context

5.1 Création de context

func main() {
    // Context vide (racine)
    ctx := context.Background()
    
    // Context TODO (à remplacer)
    ctx = context.TODO()

    // Avec annulation
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    // Avec timeout
    ctx, cancel = context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    // Avec deadline
    ctx, cancel = context.WithDeadline(ctx, time.Now().Add(5*time.Second))
    defer cancel()

    // Avec valeurs
    ctx = context.WithValue(ctx, "key", "value")
}

5.2 Propagation

func handleRequest(ctx context.Context) error {
    // Propagation aux sous-appels
    ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
    defer cancel()

    result, err := fetchData(ctx)
    if err != nil {
        return fmt.Errorf("fetch data: %w", err)
    }

    return processResult(ctx, result)
}

func fetchData(ctx context.Context) (string, error) {
    done := make(chan string, 1)

    go func() {
        time.Sleep(3 * time.Second) // Simulation d'appel long
        done <- "data"
    }()

    select {
    case result := <-done:
        return result, nil
    case <-ctx.Done():
        return "", ctx.Err()
    }
}

5.3 Context avec valeurs

type contextKey string

const (
    UserIDKey    contextKey = "user_id"
    TraceIDKey   contextKey = "trace_id"
    RequestIDKey contextKey = "request_id"
)

func middleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ctx := r.Context()
        ctx = context.WithValue(ctx, TraceIDKey, uuid.New().String())
        ctx = context.WithValue(ctx, UserIDKey, r.Header.Get("X-User-ID"))
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

func handler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    traceID := ctx.Value(TraceIDKey).(string)
    userID := ctx.Value(UserIDKey).(string)
    
    fmt.Printf("Trace: %s, User: %s\n", traceID, userID)
}

6. GOMAXPROCS et Race Detector

6.1 GOMAXPROCS

func main() {
    // Nombre de CPUs logiques
    fmt.Println("CPUs:", runtime.NumCPU())
    
    // Définir le nombre de threads OS
    runtime.GOMAXPROCS(runtime.NumCPU())
    
    // Vérifier
    fmt.Println("GOMAXPROCS:", runtime.GOMAXPROCS(0))
}

6.2 Race Detector

# Activer le race detector
go run -race main.go
go build -race -o program main.go
go test -race ./...

// race_example.go - À EXÉCUTER AVEC -race
func main() {
    counter := 0
    var wg sync.WaitGroup

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter++ // DATA RACE !
        }()
    }

    wg.Wait()
    fmt.Println(counter)
}

7. Modèle GMP

Diagramme en cours de génération...

Résumé

  • Goroutines : légères (~2KB), multiplexées sur threads OS (M:N)
  • Channels : communication synchronisée entre goroutines
  • Patterns : fan-in, fan-out, pipeline, worker pool
  • Synchronisation : Mutex, RWMutex, WaitGroup, Once, Cond, Pool
  • Context : propagation d'annulation, timeout, valeurs
  • Race detector : outil essentiel pour la concurrence
PrécédentChapitre 1 — 01 - Fondamentaux de GoSuivantChapitre 3 — 03 - Go Standard Library