GoSuda

Kit Inicial de Concorrência em Go

By snowmerak views ...

Visão geral

Breve introdução

A linguagem Go possui muitas ferramentas para o gerenciamento de concorrência. Neste artigo, apresentaremos algumas delas e alguns truques.

Goroutines?

A goroutine é um novo modelo de concorrência suportado pela linguagem Go. Geralmente, para executar várias tarefas simultaneamente, um programa recebe threads do SO do sistema operacional e executa trabalhos em paralelo, na quantidade de núcleos disponíveis. Para realizar concorrência em uma escala menor, são criadas threads verdes (green threads) no espaço do usuário (userland), de modo que várias threads verdes se revezem na execução dentro de uma única thread do SO. No entanto, as goroutines tornaram essa forma de thread verde ainda menor e mais eficiente. Essas goroutines utilizam menos memória do que as threads e podem ser criadas e alternadas mais rapidamente do que elas.

Para usar goroutines, basta utilizar a palavra-chave go. Isso permite que, no processo de escrita de um programa, código síncrono seja executado de forma intuitiva como código assíncrono.

 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}

Este código transforma um código síncrono, que espera 1 segundo e então imprime Hello, World!, em um fluxo assíncrono. Embora o exemplo atual seja simples, ao converter código síncrono mais complexo em código assíncrono, a legibilidade, a visibilidade e a compreensão do código tornam-se melhores do que com métodos existentes como async/await ou promise.

No entanto, em muitos casos, sem compreender o fluxo de simplesmente chamar código síncrono de forma assíncrona e fluxos como fork & join (semelhantes a dividir para conquistar), podem ser criados códigos de goroutine inadequados. Para lidar com essas situações, apresentaremos neste artigo alguns métodos e técnicas.

Gerenciamento de concorrência

context

Pode ser surpreendente que o context seja a primeira técnica de gerenciamento apresentada. Porém, na linguagem Go, o context vai além de uma simples função de cancelamento e desempenha um papel excelente no gerenciamento de toda a árvore de tarefas. Para quem não conhece, explicaremos brevemente esse pacote.

 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}

O código acima utiliza context para imprimir Context is done! após 1 segundo. O context permite verificar se houve cancelamento por meio do método Done() e oferece diversos métodos de cancelamento, como WithCancel, WithTimeout, WithDeadline e WithValue.

Vamos criar um exemplo simples. Suponha que você escreva um código que utilize o padrão aggregator para buscar user, post e comment, e que todas as requisições devam ser concluídas em 2 segundos. Nesse caso, pode-se escrever da seguinte forma.

 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}

O código acima imprime Timeout! caso não consiga buscar todos os dados em 2 segundos, e imprime All data is fetched! caso consiga buscar todos os dados. Dessa forma, ao utilizar context, é possível gerenciar facilmente cancelamentos e timeouts mesmo em códigos com várias goroutines em execução.

Diversas funções e métodos relacionados ao context podem ser consultados em godoc context. Espero que, após estudar o básico, você consiga utilizá-los com facilidade.

channel

unbuffered channel

channel é uma ferramenta para comunicação entre goroutines. Um channel pode ser criado com make(chan T). Aqui, T é o tipo de dado que o channel transmitirá. Dados são enviados e recebidos com <-, e o channel pode ser fechado com 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}

O código acima utiliza channel para imprimir 1 e 2. Este exemplo mostra apenas o envio e o recebimento simples de valores pelo channel, mas o channel oferece mais funcionalidades do que isso. Primeiro, vamos conhecer o buffered channel e o unbuffered channel. Antes de começar, o exemplo acima é um unbuffered channel, em que a ação de enviar dados ao canal e a ação de receber dados devem ocorrer simultaneamente. Se essas ações não ocorrerem ao mesmo tempo, pode ocorrer um deadlock.

buffered channel

E se, em vez de apenas imprimir, o código acima executasse dois processos pesados? Se o segundo processo, ao ler e processar dados, demorar muito tempo, o primeiro processo também ficará parado durante esse período. Para evitar essa situação, podemos usar um 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}

O código acima utiliza buffered channel para imprimir 1 e 2. Neste código, por meio do buffered channel, não é necessário que a ação de enviar dados ao canal e a ação de receber dados ocorram simultaneamente. Ao colocar um buffer no canal, há folga equivalente ao seu tamanho, o que permite evitar atrasos de trabalho causados pela influência de tarefas posteriores.

select

Ao lidar com vários canais, é possível implementar facilmente a estrutura fan-in com a sintaxe select.

 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}

O código acima cria três canais que periodicamente enviam 1, 2 e 3, e utiliza select para receber valores desses canais e imprimi-los. Dessa forma, ao usar select, é possível receber dados de vários canais simultaneamente e processar os valores na ordem em que chegam.

for range

É possível receber dados de um channel de forma simples com for range. Quando for range é usado com um canal, ele é executado sempre que um dado é adicionado ao canal e encerra o loop quando o canal é fechado.

 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}

O código acima utiliza channel para imprimir 1 e 2. Neste código, for range é usado para receber e imprimir dados sempre que um dado é adicionado ao canal. Quando o canal é fechado, o loop é encerrado.

Como foi escrito algumas vezes acima, essa sintaxe também pode ser usada como um meio simples de sincronização.

 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}

O código acima espera 1 segundo e então imprime Hello, World!. Neste código, o channel é utilizado para transformar código síncrono em código assíncrono. Dessa forma, ao usar channel, é possível converter facilmente código síncrono em assíncrono e definir pontos de join.

etc

  1. Ao enviar ou receber dados em um nil channel, pode ocorrer um loop infinito e um deadlock.
  2. Ao enviar dados depois de fechar um canal, ocorre um panic.
  3. Mesmo sem fechar explicitamente um canal, ele será fechado pelo GC quando for coletado.

mutex

spinlock

spinlock é um método de sincronização que tenta continuamente adquirir o lock em um loop. Na linguagem Go, é possível implementar um spinlock de forma simples utilizando ponteiros.

 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}

O código acima é a implementação do pacote spinlock. Nele, o SpinLock foi implementado com o pacote sync/atomic. O método Lock tenta adquirir o lock usando atomic.CompareAndSwapUintptr, e o método Unlock libera o lock usando atomic.StoreUintptr. Como esse método tenta adquirir o lock sem parar, ele consome CPU continuamente até que o lock seja obtido, podendo entrar em um loop infinito. Portanto, o spinlock é recomendado apenas para sincronizações simples ou para uso durante períodos curtos.

sync.Mutex

mutex é uma ferramenta para sincronização entre goroutines. O mutex fornecido pelo pacote sync oferece os métodos Lock, Unlock, RLock, RUnlock e outros. Um mutex pode ser criado com sync.Mutex, e também é possível utilizar um lock de leitura/escrita com 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}

No código acima, duas goroutines acessam quase simultaneamente a mesma variável count. Nesse caso, ao transformar o código que acessa a variável count em uma região crítica com mutex, é possível impedir o acesso concorrente à variável count. Assim, este código sempre imprimirá 2, independentemente de quantas vezes for executado.

sync.RWMutex

sync.RWMutex é um mutex que permite distinguir e utilizar lock de leitura e lock de escrita. É possível adquirir e liberar o lock de leitura com os métodos RLock e 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}

O código acima é a implementação de ConcurrentMap utilizando sync.RWMutex. Nele, o método Get adquire o lock de leitura e o método Set adquire o lock de escrita, permitindo acessar e modificar o mapa data de forma segura. O motivo de usar lock de leitura é que, em casos com muitas operações de leitura simples, não se bloqueia com lock de escrita; apenas o lock de leitura é adquirido, permitindo que várias goroutines executem operações de leitura simultaneamente. Assim, nos casos em que não há alteração de estado e, portanto, não é necessário o lock de escrita, pode-se melhorar o desempenho utilizando apenas o lock de leitura.

fakelock

fakelock é um truque simples que implementa sync.Locker. Esta estrutura fornece os mesmos métodos que sync.Mutex, mas não realiza nenhuma operação real.

1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}

O código acima é a implementação do pacote fakelock. Esse pacote implementa sync.Locker e fornece os métodos Lock e Unlock, mas, na prática, não executa nenhuma operação. Explicarei por que esse código é necessário em outra oportunidade.

waitgroup

sync.WaitGroup

sync.WaitGroup é uma ferramenta que aguarda até que todas as tarefas das goroutines sejam concluídas. Ela fornece os métodos Add, Done e Wait: o método Add adiciona a quantidade de goroutines, o método Done sinaliza que a tarefa de uma goroutine foi concluída, e o método Wait aguarda até que as tarefas de todas as goroutines sejam concluídas.

 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}

O código acima utiliza sync.WaitGroup para que 100 goroutines somem valores simultaneamente à variável c. Neste código, sync.WaitGroup é utilizado para aguardar até que todas as goroutines terminem e, em seguida, imprimir o valor acumulado em c. Em casos simples de fork & join com poucas tarefas, basta usar apenas canais, mas para fork & join com grande quantidade de tarefas, usar sync.WaitGroup também é uma boa escolha.

with slice

Quando usado junto com slices, o waitgroup pode se tornar uma ferramenta excelente para gerenciar execuções concorrentes sem locks.

 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}

O código acima utiliza apenas waitgroup para que cada goroutine gere simultaneamente um inteiro aleatório e o armazene no índice que lhe foi atribuído. Neste código, waitgroup é utilizado para aguardar até que todas as goroutines terminem e, em seguida, imprimir Done. Dessa forma, ao usar waitgroup, várias goroutines podem executar tarefas simultaneamente, armazenar dados sem locks até que todas terminem e, após a conclusão, realizar o pós-processamento de forma conjunta.

golang.org/x/sync/errgroup.ErrGroup

errgroup é um pacote que estende o sync.WaitGroup. Diferentemente do sync.WaitGroup, o errgroup cancela todas as goroutines e retorna o erro caso ocorra um erro em qualquer uma das tarefas.

 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}

O código acima utiliza errgroup para criar 10 goroutines, gerando um erro na quinta goroutine. Propositalmente, um erro foi gerado na quinta goroutine para demonstrar o caso em que ocorre um erro. Na prática, ao usar errgroup, basta criar as goroutines e aplicar diversos tipos de pós-processamento para os casos em que ocorram erros em cada uma delas.

once

É uma ferramenta para executar código que deve ser executado apenas uma vez. O código relacionado pode ser executado pelos construtores abaixo.

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 simplesmente garante que a função correspondente seja executada apenas uma vez em todo o programa.

 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}

O código acima utiliza sync.OnceFunc para imprimir Hello, World!. Neste código, sync.OnceFunc é usado para criar a função once, e mesmo que ela seja chamada várias vezes, Hello, World! é impresso apenas uma vez.

OnceValue

OnceValue não apenas garante que a função seja executada uma única vez em todo o programa, mas também armazena o valor de retorno da função e, nas chamadas seguintes, retorna o valor armazenado.

 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}

O código acima utiliza sync.OnceValue para incrementar a variável c em 1. Neste código, sync.OnceValue é usado para criar a função once, e mesmo que ela seja chamada várias vezes, a variável c é incrementada apenas uma vez e retorna 1.

OnceValues

OnceValues funciona da mesma forma que OnceValue, mas pode retornar múltiplos valores.

 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}

O código acima utiliza sync.OnceValues para incrementar a variável c em 1. Neste código, sync.OnceValues é usado para criar a função once, e mesmo que ela seja chamada várias vezes, a variável c é incrementada apenas uma vez e retorna 1.

atomic

O pacote atomic fornece operações atômicas. Ele oferece métodos como Add, CompareAndSwap, Load, Store e Swap, mas recentemente é recomendado o uso de tipos como Int64, Uint64 e Pointer.

 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}

Este é o exemplo utilizado anteriormente. É o código que incrementa atomicamente a variável c usando o tipo atomic.Int64. Com os métodos Add e Load, é possível incrementar a variável atomicamente e ler seu valor. Além disso, o método Store armazena um valor, o método Swap substitui um valor, e o método CompareAndSwap compara um valor e, se for adequado, realiza a substituição.

cond

sync.Cond

O pacote cond fornece variáveis de condição. Ele pode ser criado com sync.Cond e oferece os métodos Wait, Signal e 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}

O código acima utiliza sync.Cond para aguardar até que a variável ready se torne true. Neste código, sync.Cond é utilizado para aguardar até que ready seja true e, em seguida, imprimir Ready!. Dessa forma, ao usar sync.Cond, é possível fazer com que várias goroutines aguardem até que determinada condição seja satisfeita.

Com isso, é possível implementar uma queue simples.

 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}

Ao utilizar sync.Cond dessa forma, em vez de consumir muita CPU com spin-lock, é possível aguardar de maneira eficiente e retomar a execução quando a condição for satisfeita.

semaphore

golang.org/x/sync/semaphore.Semaphore

O pacote semaphore fornece semáforos. Ele pode ser criado com golang.org/x/sync/semaphore.Semaphore e oferece os métodos Acquire, Release e 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}

O código acima cria um semáforo com semaphore e demonstra como adquirir o semáforo com o método Acquire e liberá-lo com o método Release. Neste código, foi mostrado como adquirir e liberar um semáforo usando semaphore.

Considerações finais

Parece que o conteúdo básico é suficiente até aqui. Espero que, com base neste artigo, você compreenda como gerenciar a concorrência usando goroutines e consiga aplicá-la na prática. Espero que este artigo tenha sido útil para você. Obrigado.