cond

sync.Cond é a variável de condição da biblioteca padrão do Go, e é o único utilitário de sincronização que requer inicialização manual. Diferentemente de outras primitivas de sincronização, sync.Cond requer a passagem de um mutex (sync.Mutex) para proteger o acesso a recursos compartilhados. Ela permite que goroutines entrem em estado de espera até que uma condição seja satisfeita e sejam despertadas quando a condição for atendida.

Código de Exemplo

package main

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

var i = 0

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

    // Cria uma variável de condição e passa o mutex
    cd := sync.NewCond(&mu)

    // Adiciona 4 goroutines para processar
    wg.Add(4)

    // Cria 3 goroutines, cada uma espera a condição ser satisfeita
   	for j := range 3 {
		go func() {
			defer wg.Done()

			mu.Lock()
			for i <= 100 {
                 // Quando a condição não é satisfeita, a goroutine fica bloqueada aqui
				cd.Wait()
			}
			fmt.Printf("%d wake up\n", j)
			mu.Unlock()
		}()
	}

    // Cria uma goroutine para atualizar a condição e despertar outras goroutines
    go func() {
        defer wg.Done()
        for {
            mu.Lock()
            i++ // Atualiza variável compartilhada
            mu.Unlock()
            if i > 100 {
                cd.Broadcast() // Desperta todas as goroutines em espera quando a condição é satisfeita
                break
            }
            time.Sleep(time.Millisecond * 10) // Simula carga de trabalho
        }
    }()

    // Aguarda todas as goroutines completarem
    wg.Wait()
}

No exemplo acima, a variável compartilhada i é acessada e modificada concorrentemente por múltiplas goroutines. O mutex mu garante que o acesso a i seja seguro sob condições concorrentes. Em seguida, através de sync.NewCond(&mu) cria-se uma variável de condição cd, que depende do lock mu para garantir que o acesso ao recurso compartilhado seja síncrono durante a espera.

• Três goroutines em espera: Cada goroutine se bloqueia através de cd.Wait() até que a condição seja satisfeita (i > 100). Estas goroutines permanecem em estado bloqueado até que o valor da variável i seja atualizado.
• Uma goroutine que atualiza a condição e desperta outras goroutines: Quando a condição é satisfeita (ou seja, i > 100), esta goroutine desperta todas as goroutines em espera através de cd.Broadcast(), permitindo que continuem executando.

Estrutura

type Cond struct {
	// L é mantido enquanto observa ou muda a condição
	L Locker

	notify  notifyList
}

type notifyList struct {
	// wait é o número do ticket do próximo waiter. É incrementado
	// atomicamente fora do lock.
	wait atomic.Uint32

	notify uint32

	// Lista de waiters estacionados.
	lock mutex
	head *sudog
	tail *sudog
}

Sua estrutura não é complexa:

• L, mutex, aqui o tipo é a interface Locker, não o tipo concreto de lock
• notify, lista de notificação de goroutines em espera

A estrutura runtime.notifyList é mais importante:

• wait, valor atômico, registra quantas goroutines estão em espera
• notify, aponta para a próxima goroutine a ser despertada, começa em 0 e incrementa
• lock, mutex, não é o lock que passamos, mas um lock implementado internamente no runtime
• head, tail, ponteiros de lista encadeada

Ela tem apenas três métodos:

• Wait, bloqueia e espera
• Signal, desperta uma goroutine em espera
• Broadcast, desperta todas as goroutines em espera

A maior parte de sua implementação está oculta sob a biblioteca runtime. Estas implementações estão localizadas no arquivo runtime/sema.go, de modo que seu código na biblioteca padrão é muito curto. Seu princípio básico é essencialmente uma fila bloqueante com lock.

Wait

O método Wait faz com que a própria goroutine entre em estado de espera bloqueante até ser despertada.

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

Primeiro adiciona a si mesma ao notifyList, mas na verdade apenas incrementa notifyList.wait em um. A operação aqui é equivalente a len(notifyList)-1, obtendo o índice do último elemento.

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

A operação real de adição é completada na função notifyListWait:

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

Nesta função, primeiro aplica lock na lista encadeada, depois verifica rapidamente se a goroutine atual já foi despertada. Se já foi despertada, retorna diretamente sem precisar bloquear e esperar.

lockWithRank(&l.lock, lockRankNotifyList)
// Retorna imediatamente se este ticket já foi notificado.
if less(t, l.notify) {
	unlock(&l.lock)
	return
}

Se não foi despertada, constrói um sudog e adiciona à fila, depois suspende através de 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)

Após ser despertada, libera a estrutura sudog:

releaseSudog(s)

Signal

Signal desperta as goroutines bloqueadas na ordem FIFO (primeiro a entrar, primeiro a sair) da fila.

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

Seu fluxo é o seguinte:

1. Verifica sem lock se l.wait é igual a l.notify. Se forem iguais, indica que todas as goroutines já foram despertadas.

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

2. Após aplicar lock, verifica novamente se todas já foram despertadas.

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

3. Incrementa l.notify em um.

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

4. Percorre a lista encadeada, encontra a goroutine a ser despertada, e finalmente desperta a goroutine através de 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 desperta todas as goroutines bloqueadas.

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

Seu fluxo é basicamente o mesmo:

1. Verificação sem lock para ver se todas já foram despertadas.

   // Fast-path: se não há novos waiters desde a última notificação
   // não precisamos adquirir o lock.
   if l.wait.Load() == atomic.Load(&l.notify) {
   	return
   }

2. Aplica lock, limpa a lista encadeada, depois libera o lock. Goroutines novas que chegarem posteriormente serão adicionadas ao cabeçalho da lista.

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

3. Percorre a lista encadeada, despertando todas as goroutines.

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

Resumo

Os cenários de uso mais comuns do sync.Cond são quando é necessário sincronizar certas condições entre múltiplas goroutines, tipicamente aplicado em modelos produtor-consumidor, agendamento de tarefas e outros cenários. Nestes cenários, múltiplas goroutines precisam esperar que certas condições sejam satisfeitas antes de continuar executando, ou precisam notificar múltiplas goroutines quando uma condição mudar. Ele fornece uma maneira flexível e eficiente de gerenciar sincronização entre goroutines. Ao usar em conjunto com mutex, sync.Cond pode garantir acesso seguro a recursos compartilhados e pode controlar a ordem de execução das goroutines quando condições específicas forem satisfeitas. Entender seus princípios de implementação interna ajuda a dominar melhor técnicas de programação concorrente, especialmente ao lidar com sincronização de condições complexas.