cond

sync.Cond est la variable conditionnelle de la bibliothèque standard Go. C'est le seul outil de synchronisation qui nécessite une initialisation manuelle. Contrairement aux autres primitives de synchronisation, sync.Cond nécessite de passer un verrou mutuel (sync.Mutex) pour protéger l'accès aux ressources partagées. Elle permet aux coroutines d'entrer en attente jusqu'à ce qu'une condition soit satisfaite, puis d'être réveillées.

Exemple de code

package main

import (
    "fmt"
    "sync"
    "time"
)

var i = 0

func main() {
    var mu sync.Mutex
    var wg sync.WaitGroup

    // Crée une variable conditionnelle et lui passe un verrou mutuel
    cd := sync.NewCond(&mu)

    // Ajoute 4 coroutines en attente
    wg.Add(4)

    // Crée 3 coroutines, chacune attendant que la condition soit satisfaite
   	for j := range 3 {
		go func() {
			defer wg.Done()

			mu.Lock()
			for i <= 100 {
                 // Quand la condition n'est pas satisfaite, la coroutine se bloque ici
				cd.Wait()
			}
			fmt.Printf("%d wake up\n", j)
			mu.Unlock()
		}()
	}

    // Crée une coroutine pour mettre à jour la condition et réveiller les autres coroutines
    go func() {
        defer wg.Done()
        for {
            mu.Lock()
            i++ // Met à jour la variable partagée
            mu.Unlock()
            if i > 100 {
                cd.Broadcast() // Réveille toutes les coroutines en attente quand la condition est satisfaite
                break
            }
            time.Sleep(time.Millisecond * 10) // Simule une charge de travail
        }
    }()

    // Attend que toutes les coroutines terminent
    wg.Wait()
}

Dans l'exemple ci-dessus, la variable partagée i est accédée et modifiée par plusieurs coroutines de manière concurrente. Le verrou mutuel mu garantit que l'accès à i est sûr en conditions concurrentes. Ensuite, via sync.NewCond(&mu), une variable conditionnelle cd est créée, qui dépend du verrou mu pour garantir que l'accès aux ressources partagées est synchronisé pendant l'attente.

• Trois coroutines en attente : Chaque coroutine se bloque via cd.Wait() jusqu'à ce que la condition soit satisfaite (i > 100). Ces coroutines restent bloquées tant que la valeur de la ressource partagée i n'est pas mise à jour.
• Une coroutine mettant à jour la condition et réveillant les autres : Quand la condition est satisfaite (c'est-à-dire i > 100), cette coroutine réveille toutes les coroutines en attente via cd.Broadcast() pour qu'elles continuent leur exécution.

Structure

type Cond struct {
	// L is held while observing or changing the condition
	L Locker

	notify  notifyList
}

type notifyList struct {
	// wait is the ticket number of the next waiter. It is atomically
	// incremented outside the lock.
	wait atomic.Uint32

	notify uint32

	// List of parked waiters.
	lock mutex
	head *sudog
	tail *sudog
}

Sa structure n'est pas complexe :

• L, verrou mutuel, le type ici est l'interface Locker, pas un type de verrou spécifique
• notify, liste chaînée de notification des coroutines en attente

La structure runtime.notifyList est importante :

• wait, valeur atomique, enregistre combien de coroutines sont en attente
• notify, pointe vers la prochaine coroutine à réveiller, incrémenté à partir de 0
• lock, verrou mutuel, ce n'est pas le verrou que nous avons passé, mais un verrou implémenté en interne dans runtime
• head, tail, pointeurs de liste chaînée

Elle n'a que trois méthodes :

• Wait, attendre en se bloquant
• Signal, réveiller une coroutine en attente
• Broadcast, réveiller toutes les coroutines en attente

La plupart de son implémentation est cachée dans la bibliothèque runtime, ces implémentations se trouvent dans le fichier runtime/sema.go, de sorte que dans la bibliothèque standard son code est très court. Son principe de base est une file d'attente bloquante avec verrou.

Wait

La méthode Wait met la coroutine en attente bloquante jusqu'à ce qu'elle soit réveillée.

func (c *Cond) Wait() {
    t := runtime_notifyListAdd(&c.notify)
    c.L.Unlock()
    runtime_notifyListWait(&c.notify, t)
    c.L.Lock()
}

Elle ajoute d'abord elle-même à notifyList, mais en fait elle ne fait qu'incrémenter notifyList.wait de un. Cette opération équivaut à len(notifyList)-1, obtenant l'indice du dernier élément.

func notifyListAdd(l *notifyList) uint32 {
	return l.wait.Add(1) - 1
}

La véritable opération d'ajout est effectuée dans la fonction notifyListWait.

func notifyListWait(l *notifyList, t uint32) {
	...
}

Dans cette fonction, elle verrouille d'abord la liste chaînée, puis vérifie rapidement si la coroutine courante a déjà été réveillée. Si c'est le cas, elle retourne directement sans avoir besoin d'attendre.

lockWithRank(&l.lock, lockRankNotifyList)
// Return right away if this ticket has already been notified.
if less(t, l.notify) {
	unlock(&l.lock)
	return
}

Si elle n'a pas été réveillée, elle construit un sudog et l'ajoute à la file, puis se suspend via gopark.

s := acquireSudog()
s.g = getg()
s.ticket = t
s.releasetime = 0
if l.tail == nil {
	l.head = s
} else {
	l.tail.next = s
}
l.tail = s
goparkunlock(&l.lock, waitReasonSyncCondWait, traceBlockCondWait, 3)

Après avoir été réveillée, elle libère la structure sudog.

releaseSudog(s)

Signal

Signal réveille les coroutines bloquées dans l'ordre FIFO de la file.

func (c *Cond) Signal() {
	runtime_notifyListNotifyOne(&c.notify)
}

Son processus est le suivant :

1. Vérifie sans verrouiller si l.wait est égal à l.notify, si égal cela signifie que toutes les coroutines ont été réveillées

   if l.wait.Load() == atomic.Load(&l.notify) {
   	return
   }

2. Après verrouillage, vérifie à nouveau si toutes ont été réveillées

   lockWithRank(&l.lock, lockRankNotifyList)
   t := l.notify
   if t == l.wait.Load() {
   	unlock(&l.lock)
   	return
   }

3. Incrémente l.notify de un

   atomic.Store(&l.notify, t+1)

4. Parcourt la liste chaînée, trouve la coroutine à réveiller, puis la réveille via runtime.goready.

   for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {
   	if s.ticket == t {
   		n := s.next
   		if p != nil {
   			p.next = n
   		} else {
   			l.head = n
   		}
   		if n == nil {
   			l.tail = p
   		}
   		unlock(&l.lock)
   		s.next = nil
   		readyWithTime(s, 4)
   		return
   	}
   }
   unlock(&l.lock)

Broadcast

Broadcast réveille toutes les coroutines bloquées.

func (c *Cond) Broadcast() {
    runtime_notifyListNotifyAll(&c.notify)
}

Son processus est globalement similaire :

1. Vérification sans verrou, si toutes ont déjà été réveillées

   // Fast-path: if there are no new waiters since the last notification
   // we don't need to acquire the lock.
   if l.wait.Load() == atomic.Load(&l.notify) {
   	return
   }

2. Verrouille, vide la liste chaînée, puis libère le verrou. Les nouvelles coroutines arrivées seront ajoutées en tête de liste

   lockWithRank(&l.lock, lockRankNotifyList)
   s := l.head
   l.head = nil
   l.tail = nil
   atomic.Store(&l.notify, l.wait.Load())
   unlock(&l.lock)

3. Parcourt la liste chaînée, réveille toutes les coroutines

   for s != nil {
   	next := s.next
   	s.next = nil
   	readyWithTime(s, 4)
   	s = next
   }

Résumé

Le cas d'utilisation le plus courant de sync.Cond est la synchronisation de certaines conditions entre plusieurs coroutines, généralement appliqué aux modèles producteur-consommateur, ordonnancement de tâches, etc. Dans ces scénarios, plusieurs coroutines doivent attendre que certaines conditions soient satisfaites avant de continuer leur exécution, ou doivent être notifiées quand les conditions changent. Elle fournit un moyen flexible et efficace de gérer la synchronisation entre coroutines. Utilisée en combinaison avec un verrou mutuel, sync.Cond peut garantir la sécurité d'accès aux ressources partagées et contrôler l'ordre d'exécution des coroutines quand des conditions spécifiques sont satisfaites. Comprendre ses principes d'implémentation interne nous aide à mieux maîtriser les techniques de programmation concurrente, en particulier lors de la synchronisation de conditions complexes.