select

select é uma estrutura que pode monitorar simultaneamente múltiplos estados de canais. Sua sintaxe é semelhante ao switch.

import (
  "context"
  "log/slog"
  "os"
  "os/signal"
  "time"
)

func main() {
  finished := make(chan struct{})
  ctx, stop := signal.NotifyContext(context.Background(), os.Kill, os.Interrupt)
  defer stop()
  slog.Info("running")

  go func() {
    time.Sleep(time.Second * 2)
    finished <- struct{}{}
  }()

  select {
  case <-ctx.Done():
    slog.Info("shutting down")
  case <-finished:
    slog.Info("finished")
  }
}

Este código implementa uma lógica simples de saída suave do programa combinando o uso de context, canal e select. O select no código monitora simultaneamente os dois canais ctx.Done e finished. Há duas condições para sua saída: primeiro, o sistema operacional envia um sinal de saída; segundo, o canal finished tem mensagem para ler, ou seja, a tarefa do código do usuário foi concluída. Assim, podemos fazer trabalho de finalização ao sair do programa.

Como é sabido, select tem duas características muito importantes: primeira, não bloqueante. No código-fonte de envio e recebimento de canais, pode-se ver que há tratamento para select, permitindo julgar se o canal está disponível em situação não bloqueante. Segunda, aleatoriedade. Se houver múltiplos canais disponíveis, ele seleciona um aleatoriamente para executar, não seguindo a ordem estabelecida. Isso permite que cada canal seja executado de forma relativamente justa, caso contrário, em situações extremas, alguns canais podem nunca ser processados. Como todo seu trabalho está relacionado a canais, recomenda-se primeiro ler o artigo chan. Entender canais antes de entender select facilitará muito.

Estrutura

Em tempo de execução, há apenas uma estrutura runtime.scase representando o branch do select. Cada case em tempo de execução é representado por scase.

type scase struct {
  c    *hchan         // chan
  elem unsafe.Pointer // data element
}

Onde c refere-se ao canal, elem é o ponteiro do elemento recebido ou enviado. Na verdade, a palavra-chave select refere-se à função runtime.selectgo.

Princípio

O uso de select é dividido pelo Go em quatro situações para otimização. Isso pode ser visto na função cmd/compile/internal/walk.walkSelectCases, onde há lógica de tratamento para estas quatro situações.

func walkSelectCases(cases []*ir.CommClause) []ir.Node {
  ncas := len(cases)
  sellineno := base.Pos

  // optimization: zero-case select
  if ncas == 0 {
    return []ir.Node{mkcallstmt("block")}
  }

  // optimization: one-case select: single op.
  if ncas == 1 {
    ...
  }

  // optimization: two-case select but one is default: single non-blocking op.
  if ncas == 2 && dflt != nil {
    ...
  }

  ...
  return init
}

Otimização

O compilador otimiza as três primeiras situações. A primeira situação é quando a quantidade de cases é 0, ou seja, um select vazio. Todos sabemos que uma instrução select vazia causará bloqueio permanente da goroutine atual.

select{}

A razão do bloqueio é que o compilador o traduz em uma chamada direta à função runtime.block.

func block() {
  gopark(nil, nil, waitReasonSelectNoCases, traceBlockForever, 1) // forever
}

E a função block chama a função runtime.gopark, fazendo com que a goroutine atual mude para o estado _Gwaiting e entre em bloqueio permanente, nunca mais sendo agendada.

Segunda situação, há apenas um case e não é default. Neste caso, o compilador o traduz diretamente em uma operação de envio/recebimento de canal, e ainda é bloqueante. Por exemplo, o código abaixo:

func main() {
  ch := make(chan int)
  select {
  case <-ch:
        // do something
  }
}

Ele será traduzido em uma chamada direta à função runtime.chanrecv1, o que pode ser visto no código assembly:

TEXT  main.main(SB), ABIInternal, $2
...
LEAQ  type:chan int(SB), AX
XORL  BX, BX
PCDATA  $1, $0
CALL  runtime.makechan(SB)
XORL  BX, BX
NOP
CALL  runtime.chanrecv1(SB)
ADDQ  $16, SP
POPQ  BP
...

No caso de haver apenas um case, enviar dados para o canal segue o mesmo princípio. Será traduzido em uma chamada direta à função runtime.chansend1, também bloqueante.

Terceira situação, há dois cases e um deles é default:

func main() {
  ch := make(chan int)
  select {
  case ch <- 1:
        // do something
  default:
        // do something
  }
}

Nesta situação, será traduzido em uma instrução if com chamada à função runtime.selectnbsend, como abaixo:

if selectnbsend(ch, 1) {
  // do something
} else {
  // do something
}

Se for receber dados do canal, será traduzido em uma chamada à função runtime.selectnbrecv:

ch := make(chan int)
select {
  case x, ok := <-ch:
      // do something
  default:
      // do something
}

if selected, ok = selectnbrecv(&v, c); selected {
  // do something
} else {
  // do something
}

Vale notar que, nesta situação, o envio ou recebimento do canal é não bloqueante. Podemos ver claramente que o parâmetro block é false.

func selectnbsend(c *hchan, elem unsafe.Pointer) (selected bool) {
  return chansend(c, elem, false, getcallerpc())
}

func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected, received bool) {
  return chanrecv(c, elem, false)
}

E tanto para enviar quanto para receber dados do canal, quando block é false, há um caminho rápido que pode julgar sem bloqueio se é possível enviar ou receber dados. Como mostrado abaixo:

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  if !block && empty(c) {
        if atomic.Load(&c.closed) == 0 {
      return
    }
    return true, false
  }
  ...
}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    if !block && c.closed == 0 && full(c) {
    return false
  }
    ...
}

Ao ler do canal, se o canal estiver vazio, retorna diretamente. Ao escrever no canal, se o canal não estiver fechado e já estiver cheio, também retorna diretamente. Em situações gerais, isso causaria bloqueio da goroutine, mas combinado com select não ocorre.

Tratamento

As três situações acima são apenas otimizações para casos especiais. A palavra-chave select em uso normal será traduzida em uma chamada à função runtime.selectgo, cuja lógica de tratamento tem mais de 400 linhas.

func selectgo(cas0 *scase, order0 *uint16, pc0 *uintptr, nsends, nrecvs int, block bool) (int, bool)

O compilador coleta todas as instruções case em um array scase e passa para a função selectgo. Após o processamento, retorna dois valores:

1. O primeiro é o índice do canal selecionado aleatoriamente, indicando qual canal foi processado. Se não houver, retorna -1.
2. O segundo indica se a operação de leitura do canal foi bem-sucedida.

Aqui está uma explicação simples dos parâmetros:

• cas0, ponteiro inicial do array scase. A primeira metade armazena cases de escrita de canal, a segunda metade armazena cases de leitura de canal, diferenciados por nsends.
• order0, seu comprimento é o dobro do array scase. A primeira metade é alocada para o array pollorder, a segunda metade para o array lockorder.
• nsends e nrecvs indicam a quantidade de cases de leitura/escrita de canal. A soma dos dois é o total de cases.
• block indica se é bloqueante. Se houver case default, representa não bloqueante, seu valor é false, caso contrário é true.
• pc0, aponta para o cabeçalho de um array [ncases]uintptr, usado para análise de race condition. Pode ser ignorado posteriormente, não ajuda no entendimento de select.

Suponha que haja o código abaixo:

func main() {
  ch := make(chan int)
  select {
  case ch <- 1:
    println(1)
  case ch <- 2:
    println(2)
  case ch <- 3:
    println(3)
  case ch <- 4:
    println(4)
  default:
    println("default")
  }
}

Visualizando sua forma assembly, aqui omitimos parte do código para facilitar o entendimento:

0x0000 00000 TEXT  main.main(SB), ABIInterna
...
0x0023 00035 CALL  runtime.makechan(SB)
0x0028 00040 MOVQ  $1, main..autotmp_2+72(SP) // variáveis temporárias 1 2 3 4
0x0031 00049 MOVQ  $2, main..autotmp_4+64(SP)
0x003a 00058 MOVQ  $3, main..autotmp_6+56(SP)
0x0043 00067 MOVQ  $4, main..autotmp_8+48(SP)
...
0x00c8 00200 CALL  runtime.selectgo(SB) // chama a função runtime.selectgo
0x00cd 00205 TESTQ  AX, AX
0x00d0 00208 JLT  352 // pula para o branch default
0x00d6 00214 PCDATA  $1, $-1
0x00d6 00214 JEQ  320 // pula para o branch 4
0x00d8 00216 CMPQ  AX, $1
0x00dc 00220 JEQ  288 // pula para o branch 3
0x00de 00222 NOP
0x00e0 00224 CMPQ  AX, $2
0x00e4 00228 JNE  258 // pula para o branch 2
0x00e6 00230 PCDATA  $1, $0
0x00e6 00230 CALL  runtime.printlock(SB)
0x00eb 00235 MOVL  $3, AX
0x00f0 00240 CALL  runtime.printint(SB)
0x00f5 00245 CALL  runtime.printnl(SB)
0x00fa 00250 CALL  runtime.printunlock(SB)
0x00ff 00255 NOP
0x0100 00256 JMP  379
0x0102 00258 CALL  runtime.printlock(SB)
0x0107 00263 MOVL  $4, AX
0x010c 00268 CALL  runtime.printint(SB)
0x0111 00273 CALL  runtime.printnl(SB)
0x0116 00278 CALL  runtime.printunlock(SB)
0x011b 00283 JMP  379
0x011d 00285 NOP
0x0120 00288 CALL  runtime.printlock(SB)
0x0125 00293 MOVL  $2, AX
0x012a 00298 CALL  runtime.printint(SB)
0x012f 00303 CALL  runtime.printnl(SB)
0x0134 00308 CALL  runtime.printunlock(SB)
0x0139 00313 JMP  379
0x013b 00315 NOP
0x0140 00320 CALL  runtime.printlock(SB)
0x0145 00325 MOVL  $1, AX
0x014a 00330 CALL  runtime.printint(SB)
0x014f 00335 CALL  runtime.printnl(SB)
0x0154 00340 CALL  runtime.printunlock(SB)
0x0159 00345 JMP  379
0x015b 00347 NOP
0x0160 00352 CALL  runtime.printlock(SB)
0x0165 00357 LEAQ  go:string."default\n"(SB)
0x016c 00364 MOVL  $8, BX
0x0171 00369 CALL  runtime.printstring(SB)
0x0176 00374 CALL  runtime.printunlock(SB)
0x017b 00379 PCDATA  $1, $-1
0x017b 00379 ADDQ  $160, SP
0x0182 00386 POPQ  BP
0x0183 00387 RET

Podemos ver que há uma lógica de julgamento + salto após chamar a função selectgo. Através disso, não é difícil deduzir sua forma original:

casei, ok := runtime.selectgo()
if casei == -1 {
    println("default")
} else if casei == 3 {
    println(4)
} else if casei == 2 {
    println(3)
} else if casei == 1 {
    println(2)
} else {
    println(1)
}

O código real gerado pelo compilador pode ser diferente deste, mas o significado geral é semelhante. Portanto, o compilador, após chamar a função selectgo, usa instruções if para julgar qual canal será executado. E antes da chamada, o compilador também gera um loop for para coletar o array scase, mas aqui omitimos.

Após saber como usar externamente a função selectgo, vamos entender como a função selectgo funciona internamente. Primeiro, ela inicializa vários arrays. nsends+nrecvs representa o total de cases. Do código abaixo também podemos ver que o valor máximo de cases é 1 << 16. pollorder determina a ordem de execução dos canais, lockorder determina a ordem de bloqueio dos canais.

cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
// Seu comprimento é o dobro do array scase, a primeira metade é alocada para o array pollorder, a segunda metade para o array lockorder.
order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))

ncases := nsends + nrecvs
scases := cas1[:ncases:ncases]
pollorder := order1[:ncases:ncases]
lockorder := order1[ncases:][:ncases:ncases]

Em seguida, inicializa o array pollorder, que armazena os índices do array sacses dos canais aguardando execução:

norder := 0
for i := range scases {
    cas := &scases[i]

    // Omit cases without channels from the poll and lock orders.
    if cas.c == nil {
       cas.elem = nil // allow GC
       continue
    }

    j := fastrandn(uint32(norder + 1))
    pollorder[norder] = pollorder[j]
    pollorder[j] = uint16(i)
    norder++
}
pollorder = pollorder[:norder]
lockorder = lockorder[:norder]

Ele percorre todo o array scases, depois gera um número aleatório entre [0, i] através de runtime.fastrandn, e o troca com i. Durante o processo, pula cases onde o canal é nil. Após completar a travessia, obtém-se um array pollorder com elementos embaralhados, como mostrado na figura abaixo:

Depois, ordena o array pollorder pelo endereço dos canais usando heap sort, obtendo o array lockorder. Em seguida, chama runtime.sellock para bloqueá-los em ordem:

func sellock(scases []scase, lockorder []uint16) {
  var c *hchan
  for _, o := range lockorder {
    c0 := scases[o].c
    if c0 != c {
      c = c0
      lock(&c.lock)
    }
  }
}

Vale notar que ordenar os canais por endereço é para evitar deadlocks, pois a operação select em si não permite concorrência com bloqueio. Suponha que o bloqueio seja feito em ordem aleatória de pollorder. Considere a situação do código abaixo:

ch1 := make(chan int)
ch2 := make(chan int)
ch3 := make(chan int)
ch4 := make(chan int)

poll := func() {
    select {
    case ch1 <- 1:
        println(1)
    case ch2 <- 2:
        println(2)
    case ch3 <- 3:
        println(3)
    case ch4 <- 4:
        println(4)
    default:
        println("default")
    }
}

// A
go poll()
// B
go poll()
// C
go poll()

Três goroutines ABC chegam todas a este passo de bloqueio, e suas ordens de bloqueio são aleatórias e diferentes umas das outras. Pode ocorrer a seguinte situação, como mostrado na figura abaixo:

Suponha que a ordem de bloqueio de ABC seja igual à da figura acima, então a possibilidade de deadlock é muito grande. Por exemplo, A primeiro mantém o bloqueio de ch2, depois tenta obter o bloqueio de ch1. Mas suponha que ch1 já tenha sido bloqueado pela goroutine B, e a goroutine B tenta obter o bloqueio de ch2, então isso causa deadlock.

Se todas as goroutines bloquearem na mesma ordem, não ocorrerá deadlock. Esta é a razão fundamental pela qual lockorder deve ser ordenado por endereço.

Após bloquear, começa a fase real de tratamento. Primeiro, percorre o array pollorder, acessando os canais na ordem embaralhada anterior, percorrendo um por um para encontrar um canal disponível:

for _, casei := range pollorder {
    casi = int(casei)
    cas = &scases[casi]
    c = cas.c

    if casi >= nsends { // canal de leitura
        sg = c.sendq.dequeue()
        if sg != nil {
            goto recv
        }
        if c.qcount > 0 {
            goto bufrecv
        }
        if c.closed != 0 {
            goto rclose
        }
    } else { // canal de escrita
        if c.closed != 0 {
            goto sclose
        }
        sg = c.recvq.dequeue()
        if sg != nil {
            goto send
        }
        if c.qcount < c.dataqsiz {
            goto bufsend
        }
    }
}

Podemos ver que aqui há tratamento para 6 situações de canais de leitura/escrita. Abaixo explicamos cada uma. Primeira situação: ler do canal e há remetente aguardando envio. Aqui vai para a função runtime.recv, cuja função já foi explicada. Eventualmente, despertará a goroutine do remetente. Antes de despertar, a função de callback desbloqueia todos os canais.

recv:
  // can receive from sleeping sender (sg)
  recv(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
  recvOK = true
  goto retc

Segunda situação: ler do canal, não há remetente aguardando, quantidade de elementos no buffer maior que 0. Aqui lê diretamente os dados do buffer. Sua lógica é completamente consistente com runtime.chanrecv, depois desbloqueia.

bufrecv:
  recvOK = true
  qp = chanbuf(c, c.recvx)
  if cas.elem != nil {
    typedmemmove(c.elemtype, cas.elem, qp)
  }
  typedmemclr(c.elemtype, qp)
  c.recvx++
  if c.recvx == c.dataqsiz {
    c.recvx = 0
  }
  c.qcount--
  selunlock(scases, lockorder)
  goto retc

Terceira situação: ler do canal, mas o canal já está fechado e não há elementos restantes no buffer. Aqui primeiro desbloqueia e depois retorna diretamente.

rclose:
  selunlock(scases, lockorder)
  recvOK = false
  if cas.elem != nil {
    typedmemclr(c.elemtype, cas.elem)
  }
  goto retc

Quarta situação: enviar dados para canal fechado. Aqui primeiro desbloqueia e depois ocorre panic.

sclose:
  selunlock(scases, lockorder)
  panic(plainError("send on closed channel"))

Quinta situação: há receptor bloqueado aguardando. Aqui chama a função runitme.send e eventualmente desperta a goroutine do receptor. Antes de despertar, a função de callback desbloqueia todos os canais.

send:
  send(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
  goto retc

Sexta situação: não há goroutine de receptor aguardando. Coloca os dados a serem enviados no buffer, depois desbloqueia.

bufsend:
  typedmemmove(c.elemtype, chanbuf(c, c.sendx), cas.elem)
  c.sendx++
  if c.sendx == c.dataqsiz {
    c.sendx = 0
  }
  c.qcount++
  selunlock(scases, lockorder)
  goto retc

Depois, todas as situações acima finalmente chegam ao branch retc, que só precisa retornar o índice do canal selecionado casi e recvOk representando se a leitura foi bem-sucedida.

retc:
    return casi, recvOK

Sétima situação: não encontrou canal disponível e o código contém branch default. Então desbloqueia os canais e retorna diretamente. Aqui o casi retornado é -1, indicando que não há canal disponível.

if !block {
    selunlock(scases, lockorder)
    casi = -1
    goto retc
}

Última situação: não encontrou canal disponível e o código não contém branch default. Então a goroutine atual entra em estado bloqueante. Antes disso, selectgo adiciona a goroutine atual às filas recvq/sendq de todos os canais monitorados:

gp = getg()
nextp = &gp.waiting
for _, casei := range lockorder {
    casi = int(casei)
    cas = &scases[casi]
    c = cas.c
    sg := acquireSudog()
    sg.g = gp
    sg.isSelect = true
    sg.elem = cas.elem
    sg.releasetime = 0
    sg.c = c
    *nextp = sg
    nextp = &sg.waitlink

    if casi < nsends {
        c.sendq.enqueue(sg)
    } else {
        c.recvq.enqueue(sg)
    }
}

Aqui cria vários sudog e os vincula aos canais correspondentes, como mostrado na figura abaixo:

Depois, bloqueia através de runtime.gopark. Antes de bloquear, desbloqueia os canais. O trabalho de desbloqueio é completado pela função runtime.selparkcommit, que é passada como callback para gopark.

gp.param = nil
// Signal to anyone trying to shrink our stack that we're about
// to park on a channel. The window between when this G's status
// changes and when we set gp.activeStackChans is not safe for
// stack shrinking.
gp.parkingOnChan.Store(true)
gopark(selparkcommit, nil, waitReasonSelect, traceBlockSelect, 1)
gp.activeStackChans = false

A primeira coisa após ser despertado é remover a vinculação entre sudog e os canais:

sellock(scases, lockorder)

gp.selectDone.Store(0)
sg = (*sudog)(gp.param)
gp.param = nil

casi = -1
cas = nil
caseSuccess = false
sglist = gp.waiting
// Clear all elem before unlinking from gp.waiting.
for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
    sg1.isSelect = false
    sg1.elem = nil
    sg1.c = nil
}
gp.waiting = nil

Depois remove o sudog das filas de espera dos canais anteriores:

for _, casei := range lockorder {
    k = &scases[casei]
    if sg == sglist {
        // sg has already been dequeued by the G that woke us up.
        casi = int(casei)
        cas = k
        caseSuccess = sglist.success
        if sglist.releasetime > 0 {
            caseReleaseTime = sglist.releasetime
        }
    } else {
        c = k.c
        if int(casei) < nsends {
            c.sendq.dequeueSudoG(sglist)
        } else {
            c.recvq.dequeueSudoG(sglist)
        }
    }
    sgnext = sglist.waitlink
    sglist.waitlink = nil
    releaseSudog(sglist)
    sglist = sgnext
}

No processo acima, certamente encontrará o canal processado pela goroutine que despertou. Depois, faz o tratamento final baseado em caseSuccess. Para operações de escrita, sg.success sendo false representa que o canal já está fechado. E em todo o runtime Go, apenas a função close define ativamente este campo como false, indicando que a goroutine atual foi despertada pelo despertador através da função close. Para operações de leitura, se foi despertado pelo remetente, a operação de leitura de dados já foi completada pelo remetente através da função runtime.send antes de despertar. Seu valor é true. Se foi despertado pela função close, como antes, retorna diretamente.

c = cas.c

if casi < nsends {
    if !caseSuccess {
       goto sclose
    }
} else {
    recvOK = caseSuccess
}

selunlock(scases, lockorder)
goto retc

Até aqui, toda a lógica de select foi esclarecida. Acima dividimos em várias situações. Pode-se ver que o tratamento de select é relativamente complexo.