Concurrence avancée

Composer des étapes : le pipeline

Vous savez lancer des goroutines et les faire communiquer. L'étape d'après, c'est de les composer : enchaîner des étapes (stages) reliées par des channels, comme une chaîne de montage. Chaque étape reçoit sur un channel d'entrée, travaille, et envoie sur un channel de sortie.

Le vocabulaire officiel (Go blog) : une source (ou producteur) démarre la chaîne, des étapes intermédiaires transforment, un sink (ou consommateur) termine.

// Étape source : émet 1, 2, 3...
func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)            // on ferme quand on a fini d'émettre
    }()
    return out
}

// Étape intermédiaire : élève au carré
func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {   // range lit jusqu'à la fermeture de in
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    for n := range sq(gen(2, 3, 4)) {  // sink : on consomme
        fmt.Println(n)                  // 4, 9, 16
    }
}

Fan-out / fan-in : paralléliser une étape lente

Si une étape est lente (un appel réseau, un calcul lourd), on la démultiplie : plusieurs goroutines lisent le même channel d'entrée (fan-out), puis on fusionne leurs résultats dans un seul channel (fan-in). C'est le worker pool.

// fan-in : fusionne plusieurs channels en un seul
func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, c := range cs {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                out <- n
            }
        }(c)
    }
    go func() {
        wg.Wait()   // goroutine à part : Wait() bloque jusqu'à la fin
        close(out)  // sinon merge ne pourrait jamais retourner out
    }()
    return out
}

func main() {
    in := gen(2, 3, 4, 5, 6)
    // fan-out : 3 workers lisent le même channel in
    out := merge(sq(in), sq(in), sq(in))
    for n := range out {   // fan-in : on consomme la fusion
        fmt.Println(n)
    }
}

Annuler proprement : context.Context

Comment dire à des goroutines « arrêtez-vous, on n'a plus besoin de vous » ? Un utilisateur ferme l'onglet, un timeout expire, une requête est abandonnée. La réponse standard en Go, c'est context.Context : un signal d'annulation qu'on propage à toutes les goroutines concernées.

func main() {
    // Annule automatiquement au bout de 2 secondes
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()  // libère les ressources, toujours

    resultat := make(chan int)
    go travail(ctx, resultat)

    select {
    case r := <-resultat:
        fmt.Println("Résultat :", r)
    case <-ctx.Done():               // le contexte est annulé
        fmt.Println("Abandonné :", ctx.Err())  // context deadline exceeded
    }
}

func travail(ctx context.Context, out chan<- int) {
    select {
    case <-time.After(5 * time.Second):  // travail trop long
        out <- 42
    case <-ctx.Done():                   // on écoute l'annulation
        return                          // on s'arrête proprement
    }
}

Le piège : les fuites de goroutines

Une goroutine bloquée sur un channel que personne ne lira jamais ne se termine jamais. Elle reste en mémoire, pour toujours. Répétez l'erreur à chaque requête, et la mémoire grimpe lentement jusqu'au crash. C'est une fuite de goroutine.

func chercher() int {
    ch := make(chan int)        // channel NON bufferisé
    go func() {
        ch <- travailLong()     // ⛔ bloque ici pour toujours...
    }()
    return <-ch                 // si on ne lit qu'une fois, ok ;
}                               // mais avec un timeout qui part avant,
                                // la goroutine reste bloquée sur ch <-

Le scénario classique : on ajoute un select avec timeout côté lecteur, on part sur le timeout, et la goroutine reste coincée à vouloir envoyer dans ch que plus personne ne lit. Deux parades :

• donner au channel un buffer (make(chan int, 1)) pour que l'envoi ne bloque pas ;
• passer un context et faire écouter ctx.Done() à la goroutine pour qu'elle abandonne.

Pour aller plus loin sur la détection (pprof, goleak) et les 4 patterns qui fuient le plus, j'ai écrit un article dédié : Goroutine leaks en Go : détecter, comprendre, corriger.

À toi de jouer : répare la fuite

Le code ci-dessous lance une goroutine qui envoie dans un channel non bufferisé, mais le select peut partir sur le timeout : la goroutine reste alors bloquée à jamais. Corrige-le (un buffer suffit ici), puis exécute :

package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan int) // ← à corriger : make(chan int, 1)
    go func() {
        time.Sleep(50 * time.Millisecond)
        ch <- 42 // si le select est déjà parti, bloque pour toujours
    }()

    select {
    case r := <-ch:
        fmt.Println("Résultat :", r)
    case <-time.After(100 * time.Millisecond):
        fmt.Println("Timeout")
    }
}