Vamos Lá: Paralelismo em Golang, Parte 2

Portuguese (Português) translation by Erick Patrick (you can also view the original English article)

Visão Geral

Uma dos recursos únicos da Go é o uso de canais para comunicação segura entre goroutines. Nesse artigo, aprenderemos o que são canais, como usá-los efetivamente e alguns padrões comuns.

O Que É Um Canal?

Um canal é uma fila sincronizada em memória que goroutines e funções padrões podem usar para enviar e receber valores tipados. Comunicação é seralizada pelo canal.

Podemos cria rum canal usando make() especificando o tipo de valores que o canal aceita:

ch := make(chan int)

Go provê a sintaxe de de flexa para enviar para e receber de canais:

    // send value to a channel
    ch <- 5

    // receive value from a channel
    x := <- ch

Não temos de consumir o valor. Tudo bem removermos um valor do canal:

<-ch

Canais são bloqueantes por padrão. Se enviarmos um valor a umcanal, o bloquearemos até ele ser recebido. Igualmente, se recebermos de um canal, bloquearemo-no até alguém enviar um valor para o canal.

O programa a seguir demonstra isso. A função main() cria um canal e começa uma goroutine que imprime "start", lê um valor do canal e o imprime também. main() começa outra goroutine que imprime um traço ("-") a cada segundo. Então, dorme por 2.5 segundos, enviar um valor para o canal e dorme mais 3 segundos para deixar todas goroutines terminarem.

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan int)

    // Start a goroutine that reads a value from a channel and prints it
    go func(ch chan int) {
        fmt.Println("start")
        fmt.Println(<-ch)
    }(ch)

    // Start a goroutine that prints a dash every second
    go func() {
        for i := 0; i < 5; i++ {
            time.Sleep(time.Second)
            fmt.Println("-")
        }
    }()

    // Sleep for two seconds
    time.Sleep(2500 * time.Millisecond)

    // Send a value to the channel
    ch <- 5

    // Sleep three more seconds to let all goroutines finish
    time.Sleep(3 * time.Second)
}

Esse programa demonstra muito bem a natureza bloqueante do canal. A primeira goroutine imprime "start" de primeira, mas e bloqueada na tentativa de receber do canal até a função main(), que dorme por 2.5s, envie o valor. A outra goroutine apenas provê uma indicação visual do fluxo de tempo, imprimindo um traço a cada segundo.

Eis o retorno:

1	start
2	-
3	-
4	5
5	-
6	-
7	-

Canais de Dados Temporários

Esse comportamento liga, fortemente, remetente e recebedor, e, algumas vezes, não é o que queremos. Go provê vários mecanismos para resolver isso.

Canais de dados temporários são canais que guardam um certo número (pré-definido) de valores que o remetente não bloqueia até que esteja cheio, mesmo que ninguém receba.

Para criar um canal de dados temporário, apenas adicione a capacidade como segundo argumento:

ch := make(chan int, 5)

O programa a seguir ilustra o comportamento de canais de dados temporários: main() define um desse canais com capacidade 3. Então começa uma goroutine que lê os dados tmeporários do canal a cada segundo e imprime-os, e outr goroutine que apenas imprime um traço a cada segundo, dando indicação visual de progressão de tempo. Então, ele envia sinco valores ao canal.

import (
    "fmt"
    "time"
)


func main() {
    ch := make(chan int, 3)

    // Start a goroutine that reads a value from the channel every second and prints it
    go func(ch chan int) {
        for {
            time.Sleep(time.Second)
            fmt.Printf("Goroutine received: %d\n", <-ch)
        }

    }(ch)

    // Start a goroutine that prints a dash every second
    go func() {
        for i := 0; i < 5; i++ {
            time.Sleep(time.Second)
            fmt.Println("-")
        }
    }()

    // Push values to the channel as fast as possible
    for i := 0; i < 5; i++ {
        ch <- i
        fmt.Printf("main() pushed: %d\n", i)
    }

    // Sleep five more seconds to let all goroutines finish
    time.Sleep(5 * time.Second)
}

O que aconteceu em tempo de execução? Os 3 primeiros valores foram salvos temporariamente pelo canal, de forma imediata, e main() bloqueou. Após um segundo, o valor é recebido pela goroutine e main() pode enviar outro valor. Outro segundo passa e a goroutina recebe outro valor e main() pode enviar o último valor. Nessa altura, a goroutine continua recebendo valores do canal a cada segundo.

Eis o retorno:

main() pushed: 0
main() pushed: 1
main() pushed: 2
-
Goroutine received: 0
main() pushed: 3
-
Goroutine received: 1
main() pushed: 4
-
Goroutine received: 2
-
Goroutine received: 3
-
Goroutine received: 4

Seleção

Canais de dados temporário podem (desde que haja espaço) resolver o problema de flutuações temporárias onde não há receptor suficiente para processar todas as mensagens enviadas. Mas há, também, o problem contrário de receptores bloqueados esperando mensagens para processar. Go tem a solução.

E se quisermos que a goroutine faça algo quando não houver mensagens a processar no canal? Um bom exemplo é se o receptor esperar por mensagens de múltiplos canais. Não queremos bloquear o canal A se o canal B tiver mensagens. O programa a seguir tenta computar a soma de 3 e 5 usando o poder total da máquina.

A ideia é simular uma operação complexa (exemplo: consulta remota a BD distribuído) com redundância. A função sum() (note sua definição aninhada em main()) aceita dois parâmetros inteiros e retorna um canal de inteiros. Uma goroutine interna e anônima dorme por tempo aleatório menor que 1 segundo e, então, escreve a soma no canal, fecha e retorna-o.

Agora, main invoca sum(3,5) por quatro vezes e salva os canais resultantes em variáveis, ch1 a ch4. As quatro invocações retornam imediatamente porque o sono aleatório acontece dentro da goroutine que cada sum() invoca.

Eis a parte legal. A declaração select permitr main() esperar por todos os canais e responder o primeiro que retornar algo. A declaração select opera como uma declaração switch.

func main() {
    r := rand.New(rand.NewSource(time.Now().UnixNano()))

    sum := func(a int, b int) <-chan int {
        ch := make(chan int)
        go func() {
            // Random time up to one second
            delay := time.Duration(r.Int()%1000) * time.Millisecond
            time.Sleep(delay)
            ch <- a + b
            close(ch)
        }()
        return ch
    }

    // Call sum 4 times with the same parameters
    ch1 := sum(3, 5)
    ch2 := sum(3, 5)
    ch3 := sum(3, 5)
    ch4 := sum(3, 5)

    // wait for the first goroutine to write to its channel
    select {
    case result := <-ch1:
        fmt.Printf("ch1: 3 + 5 = %d", result)
    case result := <-ch2:
        fmt.Printf("ch2: 3 + 5 = %d", result)
    case result := <-ch3:
        fmt.Printf("ch3: 3 + 5 = %d", result)
    case result := <-ch4:
        fmt.Printf("ch4: 3 + 5 = %d", result)
    }
}

Por vezes, não queremos main() bloqueada, nem que seja pela primeira goroutine a terminar. Aqui, podemos usar um caso padrão que executará se todos os canais estão bloqueados.

Exemplo de Rastreador Web

No artigo anterior, mostramos uma solução de um exercício de rastreador web do Tour de Go. Usamos goroutines e mapas sincronizados. Também resolvemos o exercício usando canais. O código completo das soluções está disponível no GitHub.

As partes relevantes: Primeiro, uma estrutura que será enviada ao canal sempre que uma goroutina analisar uma página. Ela contém a profunidade atual e todas as URLs da página.

1	type links struct {
2	urls []string
3	depth int
4	}

A função fetchURL() aceita uma URL, profundidade e canal de saída. Ela usa o buscador (provido pelo exercício) para obter URLs de todos os links da página. Ela envia a lista de URL como mensagem única ao canal candidato como uma estrutura links com profundidade menor. A profundidade representa o quanto se deverá rastrear. Quando chegar a 0, não é preciso mais processamento.

func fetchURL(url string, depth int, candidates chan links) {
    body, urls, err := fetcher.Fetch(url)
    fmt.Printf("found: %s %q\n", url, body)

    if err != nil {
        fmt.Println(err)
    }

    candidates <- links{urls, depth - 1}
}

A função ChannelCrawl() coordena tudo. Ela registra todas as URL que já foram buscadas em um mapa. Não é preciso sincronizar o acesso porque nenhuma outra função ou goroutina usará. Também define um canal candidato em que todas as goroutines escreverão seus resultados.

Então, começa a invocar parseUrl como goroutines para cada nova URL. A lógica registra quantas goroutines foram invocadas, usando um contador. Sempre que um valor é lido do canal, o contador é reduzido (a goroutine remetente termina após o envio), e sempre que uma nova goroutine é lançada, o contador é incrementado. Se profundidade chegar a zero, nenhuma outra goroutine será invocada e a função principal, continuará lendo do canal até todas goroutines terminarem.

// ChannelCrawl crawls links from a seed url
func ChannelCrawl(url string, depth int, fetcher Fetcher) {
    candidates := make(chan links, 0)
    fetched := make(map[string]bool)
    counter := 1

    // Fetch initial url to seed the candidates channel
    go fetchURL(url, depth, candidates)

    for counter > 0 {
        candidateLinks := <-candidates
        counter--
        depth = candidateLinks.depth
        for _, candidate := range candidateLinks.urls {
            // Already fetched. Continue...
            if fetched[candidate] {
                continue
            }

            // Add to fetched mapped
            fetched[candidate] = true

            if depth > 0 {
                counter++
                go fetchURL(candidate, depth, candidates)
            }
        }
    }

Conclusão

Os canais da Go proveem várias opções para comunicação segura entre goroutines. O suporte à sintaxe é tanto concisa quanto ilustrativa. É um benefício real ao criar algortimos concorrentes. Há muitos mais sobre canais que o mostrado aqui. Encorajamos a ver mais e familiarizar-se com os vários padrões de paralelismo que eles permitem.