GoSudaKit de démarrage sur la concurrence en Go
Aperçu
Brève introduction
Le langage Go dispose de nombreux outils pour la gestion de la concurrence. Dans cet article, nous allons présenter certains d'entre eux ainsi que quelques astuces.
Goroutines ?
Une goroutine est un nouveau modèle de concurrence pris en charge par le langage Go. En général, pour effectuer plusieurs tâches simultanément, un programme reçoit des threads OS du système d'exploitation et exécute des tâches en parallèle à raison du nombre de cœurs. Et pour réaliser une concurrence à plus petite échelle, on crée des threads verts au niveau utilisateur (userland), de sorte que plusieurs threads verts s'exécutent à tour de rôle au sein d'un même thread OS. Cependant, les goroutines rendent cette forme de threads verts plus légère et plus efficace. Ces goroutines utilisent moins de mémoire que les threads et peuvent être créées et commutées plus rapidement que ces derniers.
Pour utiliser une goroutine, il suffit d'employer le mot-clé go. Cela permet, lors de la rédaction d'un programme, d'exécuter intuitivement du code synchrone sous forme de code asynchrone.
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch := make(chan struct{})
10 go func() {
11 defer close(ch)
12 time.Sleep(1 * time.Second)
13 fmt.Println("Hello, World!")
14 }()
15
16 fmt.Println("Waiting for goroutine...")
17 for range ch {}
18}
Ce code transforme simplement un code synchrone, qui attend une seconde puis affiche Hello, World!, en un flux asynchrone. Si l'exemple actuel est simple, la transformation d'un code synchrone un peu plus complexe en code asynchrone améliore la lisibilité, la visibilité et la compréhension du code, par rapport aux approches classiques telles que async/await ou promise.
Toutefois, dans de nombreux cas, des goroutines mal conçues sont produites lorsqu'on ne comprend pas le simple appel asynchrone d'un code synchrone, ni les flux de type fork & join (semblables à une logique de diviser pour régner). Pour se prémunir contre ces situations, cet article présentera quelques méthodes et techniques.
Gestion de la concurrence
context
Il peut sembler surprenant que context soit la première technique présentée. Pourtant, en Go, context joue un rôle remarquable dans la gestion de l'arbre complet des tâches, bien au-delà de la simple annulation. Pour ceux qui ne le connaissent pas, je vais brièvement présenter ce package.
1package main
2
3func main() {
4 ctx, cancel := context.WithCancel(context.Background())
5 defer cancel()
6
7 go func() {
8 <-ctx.Done()
9 fmt.Println("Context is done!")
10 }()
11
12 time.Sleep(1 * time.Second)
13
14 cancel()
15
16 time.Sleep(1 * time.Second)
17}
Le code ci-dessus utilise context pour afficher Context is done! après une seconde. context permet de vérifier si une annulation a eu lieu via la méthode Done(), et fournit diverses méthodes d'annulation telles que WithCancel, WithTimeout, WithDeadline et WithValue.
Créons un exemple simple. Supposons que vous écriviez un code qui utilise le motif aggregator pour récupérer user, post et comment, et que toutes les requêtes doivent être effectuées dans un délai de 2 secondes. On peut alors écrire le code suivant.
1package main
2
3func main() {
4 ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
5 defer cancel()
6
7 ch := make(chan struct{})
8 go func() {
9 defer close(ch)
10 user := getUser(ctx)
11 post := getPost(ctx)
12 comment := getComment(ctx)
13
14 fmt.Println(user, post, comment)
15 }()
16
17 select {
18 case <-ctx.Done():
19 fmt.Println("Timeout!")
20 case <-ch:
21 fmt.Println("All data is fetched!")
22 }
23}
Si toutes les données ne sont pas récupérées en 2 secondes, le code ci-dessus affiche Timeout!, et s'il les récupère toutes, il affiche All data is fetched!. De cette manière, en utilisant context, on peut gérer facilement l'annulation et les délais d'expiration, même dans un code où plusieurs goroutines fonctionnent.
Les différentes fonctions et méthodes liées à context sont consultables sur godoc context. J'espère que vous pourrez apprendre les éléments simples et les utiliser avec aisance.
channel
unbuffered channel
channel est un outil de communication entre goroutines. Un channel se crée avec make(chan T). Ici, T désigne le type des données que ce channel transmettra. On envoie et reçoit des données avec <-, et on peut fermer un channel avec close.
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
Le code ci-dessus utilise channel pour afficher 1 et 2. Cet exemple se contente de montrer l'envoi et la réception de valeurs dans un channel, mais channel offre davantage de fonctionnalités. Commençons par examiner buffered channel et unbuffered channel. Avant de commencer, l'exemple présenté plus haut est un unbuffered channel : l'action d'envoyer des données dans le canal et celle de les recevoir doivent avoir lieu simultanément. Si ces actions ne se produisent pas en même temps, un deadlock peut survenir.
buffered channel
Et si, au lieu d'un simple affichage, le code précédent effectuait deux processus lourds ? Si le second processus, qui lit et traite les données, est retardé pendant longtemps, le premier processus sera également bloqué pendant ce temps. Pour éviter cette situation, on peut utiliser buffered channel.
1package main
2
3func main() {
4 ch := make(chan int, 2)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
Le code ci-dessus utilise buffered channel pour afficher 1 et 2. Dans ce code, grâce à buffered channel, l'action d'envoyer des données dans le canal et celle de les recevoir n'ont pas besoin d'avoir lieu simultanément. Ainsi, en ajoutant un tampon au canal, on dispose d'une marge correspondant à sa longueur, ce qui permet d'éviter les retards de travail causés par l'influence des tâches en aval.
select
Lorsqu'on manipule plusieurs canaux, la syntaxe select permet de mettre en œuvre facilement une structure fan-in.
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch1 := make(chan int, 10)
10 ch2 := make(chan int, 10)
11 ch3 := make(chan int, 10)
12
13 go func() {
14 for {
15 ch1 <- 1
16 time.Sleep(1 * time.Second)
17 }
18 }()
19 go func() {
20 for {
21 ch2 <- 2
22 time.Sleep(2 * time.Second)
23 }
24 }()
25 go func() {
26 for {
27 ch3 <- 3
28 time.Sleep(3 * time.Second)
29 }
30 }()
31
32 for i := 0; i < 3; i++ {
33 select {
34 case v := <-ch1:
35 fmt.Println(v)
36 case v := <-ch2:
37 fmt.Println(v)
38 case v := <-ch3:
39 fmt.Println(v)
40 }
41 }
42}
Le code ci-dessus crée trois canaux qui transmettent périodiquement 1, 2 et 3, puis utilise select pour recevoir les valeurs de ces canaux et les afficher. De cette manière, select permet de recevoir simultanément des données de plusieurs canaux et de traiter chaque valeur dès qu'elle est reçue.
for range
On peut facilement recevoir des données d'un channel à l'aide de for range. Lorsque for range est appliqué à un canal, la boucle s'exécute à chaque ajout de données dans ce canal, et se termine lorsque le canal est fermé.
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
Le code ci-dessus utilise channel pour afficher 1 et 2. Dans cet exemple, for range reçoit et affiche les données à chaque ajout dans le canal, puis termine la boucle lorsque le canal est fermé.
Comme on l'a fait à plusieurs reprises plus haut, cette syntaxe peut aussi servir de simple mécanisme de synchronisation.
1package main
2
3func main() {
4 ch := make(chan struct{})
5 go func() {
6 defer close(ch)
7 time.Sleep(1 * time.Second)
8 fmt.Println("Hello, World!")
9 }()
10
11 fmt.Println("Waiting for goroutine...")
12 for range ch {}
13}
Le code ci-dessus attend une seconde puis affiche Hello, World!. Dans ce code, un channel est utilisé pour transformer un code synchrone en code asynchrone. De cette manière, en employant channel, on peut facilement transformer un code synchrone en code asynchrone et définir un point de join.
etc
- Envoyer ou recevoir des données sur un nil channel peut provoquer une boucle infinie et un deadlock.
- Envoyer des données après la fermeture d'un channel provoque une panic.
- Il n'est pas nécessaire de fermer explicitement un channel : le GC le collecte et le ferme alors.
mutex
spinlock
spinlock est une méthode de synchronisation qui tente en boucle d'acquérir le verrou en continu. En Go, il est facile d'implémenter un spinlock à l'aide de pointeurs.
1package spinlock
2
3import (
4 "runtime"
5 "sync/atomic"
6)
7
8type SpinLock struct {
9 lock uintptr
10}
11
12func (s *SpinLock) Lock() {
13 for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
14 runtime.Gosched()
15 }
16}
17
18func (s *SpinLock) Unlock() {
19 atomic.StoreUintptr(&s.lock, 0)
20}
21
22func NewSpinLock() *SpinLock {
23 return &SpinLock{}
24}
Le code ci-dessus est l'implémentation du package spinlock. Ici, SpinLock est implémenté à l'aide du package sync/atomic. La méthode Lock tente d'acquérir le verrou via atomic.CompareAndSwapUintptr, et la méthode Unlock le libère via atomic.StoreUintptr. Comme cette approche tente d'acquérir le verrou sans relâche, elle consomme continuellement du CPU jusqu'à l'obtention du verrou, et peut donc entrer dans une boucle infinie. Par conséquent, il est préférable d'utiliser spinlock pour une synchronisation simple ou seulement pendant de courtes durées.
sync.Mutex
mutex est un outil de synchronisation entre goroutines. Le mutex fourni par le package sync propose les méthodes Lock, Unlock, RLock, RUnlock, etc. Un mutex se crée avec sync.Mutex, et il est également possible d'utiliser un verrou de lecture/écriture avec sync.RWMutex.
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 var mu sync.Mutex
9 var count int
10
11 go func() {
12 mu.Lock()
13 count++
14 mu.Unlock()
15 }()
16
17 mu.Lock()
18 count++
19 mu.Unlock()
20
21 println(count)
22}
Dans le code ci-dessus, deux goroutines accèdent presque simultanément à la même variable count. En utilisant mutex pour transformer le code qui accède à la variable count en section critique, on peut empêcher l'accès concurrent à count. Ainsi, ce code affichera toujours 2, quel que soit le nombre d'exécutions.
sync.RWMutex
sync.RWMutex est un mutex qui permet de distinguer et d'utiliser séparément le verrou de lecture et le verrou d'écriture. On peut poser et libérer un verrou de lecture à l'aide des méthodes RLock et RUnlock.
1package cmap
2
3import (
4 "sync"
5)
6
7type ConcurrentMap[K comparable, V any] struct {
8 sync.RWMutex
9 data map[K]V
10}
11
12func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
13 m.RLock()
14 defer m.RUnlock()
15
16 value, ok := m.data[key]
17 return value, ok
18}
19
20func (m *ConcurrentMap[K, V]) Set(key K, value V) {
21 m.Lock()
22 defer m.Unlock()
23
24 m.data[key] = value
25}
Le code ci-dessus implémente ConcurrentMap à l'aide de sync.RWMutex. Dans ce code, la méthode Get pose un verrou de lecture et la méthode Set pose un verrou d'écriture, ce qui permet d'accéder et de modifier la map data en toute sécurité. Le verrou de lecture est utile lorsque les opérations de simple lecture sont nombreuses : en ne posant qu'un verrou de lecture, sans verrou d'écriture, plusieurs goroutines peuvent lire simultanément. Ainsi, lorsqu'aucune modification d'état n'est nécessaire et qu'un verrou d'écriture n'est pas requis, n'utiliser qu'un verrou de lecture permet d'améliorer les performances.
fakelock
fakelock est une astuce simple consistant à implémenter sync.Locker. Cette structure fournit les mêmes méthodes que sync.Mutex, mais n'effectue aucune opération réelle.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
Le code ci-dessus est l'implémentation du package fakelock. Ce package implémente sync.Locker et fournit les méthodes Lock et Unlock, mais en réalité il ne fait absolument rien. Je reviendrai sur la raison de l'existence de ce code lorsque l'occasion se présentera.
waitgroup
sync.WaitGroup
sync.WaitGroup est un outil qui attend que toutes les tâches des goroutines soient terminées. Il fournit les méthodes Add, Done et Wait : Add ajoute le nombre de goroutines, Done signale qu'une goroutine a terminé sa tâche, et Wait attend que toutes les goroutines aient terminé leur travail.
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
Le code ci-dessus utilise sync.WaitGroup pour que 100 goroutines ajoutent simultanément une valeur à la variable c. Ici, sync.WaitGroup attend que toutes les goroutines soient terminées, puis affiche la valeur accumulée dans c. Pour un fork & join de quelques tâches seulement, l'utilisation des canaux suffit, mais pour un fork & join d'un grand nombre de tâches, sync.WaitGroup est également un très bon choix.
with slice
Combiné à un slice, waitgroup peut devenir un excellent outil pour gérer des tâches exécutées simultanément sans verrou.
1package main
2
3import (
4 "fmt"
5 "sync"
6 "rand"
7)
8
9func main() {
10 var wg sync.WaitGroup
11 arr := [10]int{}
12
13 for i := 0; i < 10; i++ {
14 wg.Add(1)
15 go func(id int) {
16 defer wg.Done()
17
18 arr[id] = rand.Intn(100)
19 }(i)
20 }
21
22 wg.Wait()
23 fmt.Println("Done")
24
25 for i, v := range arr {
26 fmt.Printf("arr[%d] = %d\n", i, v)
27 }
28}
Le code ci-dessus utilise uniquement waitgroup pour que chaque goroutine génère simultanément un entier aléatoire parmi 10 et le stocke à l'indice qui lui a été attribué. Ici, waitgroup attend que toutes les goroutines soient terminées, puis affiche Done. De cette manière, avec waitgroup, plusieurs goroutines exécutent leurs tâches simultanément, stockent les données sans verrou jusqu'à la fin de toutes les goroutines, puis effectuent un post-traitement global une fois le travail terminé.
golang.org/x/sync/errgroup.ErrGroup
errgroup est un package qui étend sync.WaitGroup. Contrairement à sync.WaitGroup, errgroup annule toutes les goroutines et renvoie l'erreur dès qu'une des tâches des goroutines produit une erreur.
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/errgroup"
7)
8
9func main() {
10 g, ctx := errgroup.WithContext(context.Background())
11 _ = ctx
12
13 for i := 0; i < 10; i++ {
14 i := i
15 g.Go(func() error {
16 if i == 5 {
17 return fmt.Errorf("error")
18 }
19 return nil
20 })
21 }
22
23 if err := g.Wait(); err != nil {
24 fmt.Println(err)
25 }
26}
Le code ci-dessus utilise errgroup pour créer 10 goroutines, et déclenche volontairement une erreur dans la cinquième goroutine afin de montrer le cas où une erreur survient. En pratique, il suffit d'utiliser errgroup pour créer des goroutines et de mettre en place divers post-traitements lorsqu'une erreur survient dans chacune d'elles.
once
Il s'agit d'un outil permettant d'exécuter du code qui ne doit être exécuté qu'une seule fois. Le code associé peut être exécuté à l'aide des constructeurs suivants.
1func OnceFunc(f func()) func()
2func OnceValue[T any](f func() T) func() T
3func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)
OnceFunc
OnceFunc garantit simplement que la fonction concernée ne sera exécutée qu'une seule fois sur l'ensemble du programme.
1package main
2
3import "sync"
4
5func main() {
6 once := sync.OnceFunc(func() {
7 println("Hello, World!")
8 })
9
10 once()
11 once()
12 once()
13 once()
14 once()
15}
Le code ci-dessus utilise sync.OnceFunc pour afficher Hello, World!. Dans ce code, la fonction once est créée avec sync.OnceFunc, et même si elle est appelée plusieurs fois, Hello, World! n'est affiché qu'une seule fois.
OnceValue
OnceValue ne se contente pas d'exécuter la fonction une seule fois sur l'ensemble du programme : il mémorise la valeur de retour de la fonction et renvoie cette valeur mémorisée lors des appels suivants.
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValue(func() int {
8 c += 1
9 return c
10 })
11
12 println(once())
13 println(once())
14 println(once())
15 println(once())
16 println(once())
17}
Le code ci-dessus utilise sync.OnceValue pour incrémenter la variable c de 1. Dans ce code, la fonction once est créée avec sync.OnceValue et, même si elle est appelée plusieurs fois, la variable c n'est incrémentée qu'une seule fois, et la valeur 1 est renvoyée.
OnceValues
OnceValues fonctionne de manière identique à OnceValue, mais peut renvoyer plusieurs valeurs.
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValues(func() (int, int) {
8 c += 1
9 return c, c
10 })
11
12 a, b := once()
13 println(a, b)
14 a, b = once()
15 println(a, b)
16 a, b = once()
17 println(a, b)
18 a, b = once()
19 println(a, b)
20 a, b = once()
21 println(a, b)
22}
Le code ci-dessus utilise sync.OnceValues pour incrémenter la variable c de 1. Dans ce code, la fonction once est créée avec sync.OnceValues et, même si elle est appelée plusieurs fois, la variable c n'est incrémentée qu'une seule fois, et la valeur 1 est renvoyée.
atomic
Le package atomic fournit des opérations atomiques. Il propose les méthodes Add, CompareAndSwap, Load, Store, Swap, etc., mais ces dernières années, l'utilisation des types Int64, Uint64, Pointer, etc. est recommandée.
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
C'est l'exemple utilisé précédemment. Il utilise le type atomic.Int64 pour incrémenter la variable c de manière atomique. Les méthodes Add et Load permettent d'incrémenter la variable de manière atomique et de lire sa valeur. De plus, la méthode Store enregistre une valeur, Swap remplace une valeur, et CompareAndSwap compare une valeur et la remplace si elle correspond.
cond
sync.Cond
Le package cond fournit des variables de condition. Il se crée avec sync.Cond et propose les méthodes Wait, Signal et Broadcast.
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 c := sync.NewCond(&sync.Mutex{})
9 ready := false
10
11 go func() {
12 c.L.Lock()
13 ready = true
14 c.Signal()
15 c.L.Unlock()
16 }()
17
18 c.L.Lock()
19 for !ready {
20 c.Wait()
21 }
22 c.L.Unlock()
23
24 println("Ready!")
25}
Le code ci-dessus utilise sync.Cond pour attendre que la variable ready devienne true. Dans ce code, sync.Cond permet d'attendre que ready devienne true, puis affiche Ready!. En utilisant sync.Cond de cette manière, plusieurs goroutines peuvent attendre simultanément qu'une condition particulière soit satisfaite.
En s'appuyant sur ce principe, on peut implémenter une simple queue.
1package queue
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8type Node[T any] struct {
9 Value T
10 Next *Node[T]
11}
12
13type Queue[T any] struct {
14 sync.Mutex
15 Cond *sync.Cond
16 Head *Node[T]
17 Tail *Node[T]
18 Len int
19}
20
21func New[T any]() *Queue[T] {
22 q := &Queue[T]{}
23 q.Cond = sync.NewCond(&q.Mutex)
24 return q
25}
26
27func (q *Queue[T]) Push(value T) {
28 q.Lock()
29 defer q.Unlock()
30
31 node := &Node[T]{Value: value}
32 if q.Len == 0 {
33 q.Head = node
34 q.Tail = node
35 } else {
36 q.Tail.Next = node
37 q.Tail = node
38 }
39 q.Len++
40 q.Cond.Signal()
41}
42
43func (q *Queue[T]) Pop() T {
44 q.Lock()
45 defer q.Unlock()
46
47 for q.Len == 0 {
48 q.Cond.Wait()
49 }
50
51 node := q.Head
52 q.Head = q.Head.Next
53 q.Len--
54 return node.Value
55}
Ainsi, en utilisant sync.Cond, on peut attendre efficacement, au lieu de consommer beaucoup de CPU avec un spin-lock, et reprendre l'exécution dès que la condition est satisfaite.
semaphore
golang.org/x/sync/semaphore.Semaphore
Le package semaphore fournit des sémaphores. Il se crée avec golang.org/x/sync/semaphore.Semaphore et propose les méthodes Acquire, Release et TryAcquire.
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/semaphore"
7)
8
9func main() {
10 s := semaphore.NewWeighted(1)
11
12 if s.TryAcquire(1) {
13 fmt.Println("Acquired!")
14 } else {
15 fmt.Println("Not Acquired!")
16 }
17
18 s.Release(1)
19}
Le code ci-dessus crée un sémaphore avec semaphore, acquiert le sémaphore avec la méthode Acquire puis le libère avec la méthode Release. Ce code montre comment acquérir et libérer un sémaphore à l'aide de semaphore.
Conclusion
Ces éléments de base devraient suffire pour l'instant. J'espère qu'à partir du contenu de cet article, vous pourrez comprendre comment gérer la concurrence avec les goroutines et l'utiliser concrètement. J'espère que cet article vous aura été utile. Merci.