Go: Processos e Concorrência

Afluentes: Sistemas Distribuídos e Mobile

Concorrência

A linguagem de programação Go trabalha com concorrência usando as gorotines. Elas exencutam funções paralelamente. É mais ou menos o conceito de threads, mas são mais leves, pois o custo da criação de gorotines é menor do que se comparado a uma thread.

Vantagens das Gorotines

• Mais leves computacionamente, ocupando menos memória na pilha e a pilha pode alterar seu tamanho conforme a necessidade da aplicação. As threads possuem tamanho de pilha fixo;
• As gorotines são multiplexadas para um número menor de threads do Sistema Operacional. Ou seja, pode haver uma única thread no programa com várias gorotines executando. Caso uma thread seja bloqueada por uma goritne, as outras gorotines que estão nela podem ser movidas para outra thread;
• Goroutines se comunicam por meio de canais, que previnem que race conditions aconteçam ao acessar memória compartilhada.

Para chamar uma função de forma concorrente no Go, basta chamá-la com a palavra chave go na frente:

 1package main
 2 
 3import (  
 4    "fmt"
 5)
 6 
 7func hello() {  
 8    fmt.Println("Hello world goroutine")
 9}
10func main() {  
11    go hello()
12    // Vamos dar uma pausa aqui senão o programa pode terminar
13    // antes de executar a gorotine
14    time.Sleep(1 * time.Second)
15    fmt.Println("main function")
16}

Subprocessos

Estou falando em subprocessos, mas estamos implementando gorotines na linguagem go.

Veja que no exemplo, vamos criar uma gorotine que será filha do processo principal. Nas duas execuções, colocamos algumas linhas de execução e um pequeno delay aleatório.

O problema é que caso o programa termine sem ter terminado a gorotine chamada por ele, ela fecha também. Para contornar esse problema, vamos usar um sistema de sincronização via WAIT.

A gente cria uma variável que será um objeto do tipo sync.WaitGroup e para cada gorotine instanciada, a gente faz um incremento no objeto.

var wg sync.WaitGroup
wg.Add(1)
go child(&wg) // child é nossa função para paralelizar

De dentro da função child, que irá executar em paralelo, a gente já de início dá um defer para executar um Done no objeto WaitGroup. Cada done feito no objeto, é decrementado em um uma variável interna que controla a quantidade de gorotines executando.

defer wg.Done()

Não podemos esquecer que vamos ter que passar o objeto do WaitGroup como parâmetro de referência, então nossa função tem que estar preparada pra isso.

func child(wg *sync.WaitGroup) { //...

Ao final do programa principal, a gente usa um WAIT para agardar até que a gorotine tenha terminado.

wg.Wait()

Para só então dar sequência e fechar o programa.

 1package main
 2
 3import (
 4	"fmt"
 5	"math/rand"
 6	"sync"
 7	"time"
 8)
 9
10func child(wg *sync.WaitGroup) {
11	defer wg.Done()                   // decrementa o contador de processo
12	fmt.Println("Luke: Noooooo.....") // imprime na tela
13	for i := 0; i < 5; i++ {          // executa o laço 5 vezes
14		fmt.Println("Luke: No!")                        // imprime n a tela
15		d := rand.Intn(2000)                            // retorna um número randômico de 0 a 1999
16		time.Sleep(time.Duration(d) * time.Millisecond) // aguarda um tempo
17	}
18	fmt.Println("Luke Skywalker falls.") // imprime na tela
19}
20
21func main() {
22	var wg sync.WaitGroup                           // wg é tilizado para sincronizar os processos
23	wg.Add(1)                                       // incrementa o contador de processos
24	go child(&wg)                                   // chama a rotina child paralelamente
25	fmt.Println("Vader: Luke! I am your father...") // imprime na tela
26	for i := 0; i < 5; i++ {                        // executa o laço 5 vezes
27		fmt.Println("Vader: Yes!")                      // imprime na tela
28		d := rand.Intn(2000)                            // retorna um número randômico de 0 a 1999
29		time.Sleep(time.Duration(d) * time.Millisecond) // aguarda um tempo
30	}
31	wg.Wait()                         // aguarda os subprocessos terminarem
32	fmt.Println("Dart Vader leaves.") // imprime na tela
33}

Muitos Filhos

Quando um processo principal (pai) cria vários filhos, pode ser necessário aguardar que todos eles terminem seu processamento. Para isso, usamos uma forma de sincronização para o pai aguardar os filhos fecharem.

O exemplo abaixo cria n filhos e aguarda eles terminarem. Ele utiliza o sync.WaitGroup para sincronizar os processos.

 1package main
 2
 3import (
 4	"fmt"
 5	"math/rand"
 6	"sync"
 7	"time"
 8)
 9
10func child(wg *sync.WaitGroup, id int) { // função do filho
11	defer wg.Done()          // decrementa o sincronizador
12	for i := 0; i < 5; i++ { // conta até 5
13		fmt.Println("CHILD[", id, "] ", i)              // imprime i
14		d := rand.Intn(2000)                            // pega um valor randômico
15		time.Sleep(time.Duration(d) * time.Millisecond) // pausa um tempo aleatório
16	}
17	fmt.Println("CHILD[", id, "] done...") // imprime a finalização
18}
19
20func main() { // programa principal
21	fmt.Println("START PROGRAM...") // imprime na tela
22	var wg sync.WaitGroup           // inicializa o sincronizador
23	for i := 0; i < 5; i++ {        // cria 5 filhos
24		wg.Add(1)        // adiciona um no sincronizador
25		go child(&wg, i) //cria filho
26	}
27	wg.Wait()                     // aguarda todos os filhos terminarem
28	fmt.Println("END PROGRAM...") // imprime na tela
29}

Comunicação entre Processos

A comunicação entre processos no go é feito por canais. Eles são comparados aos pipes na linguagem C. Cada canal tem um tipo associado. chan T é um canal do tipo T.

// Canal a do tipo int
a := make(chan int)

Para usar o canal, usamos uma seta direcional

// estamos lendo do canal a e jogando para a variável data
data := <- a

// estamos escrevendo as informações de data para o canal a
a <- data

Envios e recebimentos de informações pelos canais são bloqueantes, ou seja. Quando você envia algo, a função aguarda até que outra gorotine receba o dado. Da mesma forma funciona o envio. Isso também é últim para sincronização de gorotines.

Exemplo de comunicação usando canais

 1package main
 2
 3import (
 4	"fmt"
 5	"time"
 6)
 7
 8func child(fc chan string, cc chan string) {
 9	msg := <-cc                                   // aguarda a mensagem do pai
10	fmt.Println("CHILD: Father says " + msg)      // imprime na tela
11	time.Sleep(2 * time.Second)                   // aguarda 2 segundos
12	fmt.Println("CHILD: I will response just ok") // imprime na tela
13	fc <- "ok!"                                   // envia a resposta pelo canal do pai
14}
15
16func main() {
17	fatherChannel := make(chan string)                      // canal de comunicação do pai
18	childCHannel := make(chan string)                       // canal de comunicação do filho
19	go child(fatherChannel, childCHannel)                   // chama a rotina child paralelamente
20	fmt.Println("FATHER: I will say hello to my child")     // imprime na tela
21	msg := "hello"                                          // cria a variável msg e atribui a string hello
22	childCHannel <- msg                                     // envia a mensagem hello pelo canal do filho
23	msg = <-fatherChannel                                   // aguarda a resposta do filho
24	fmt.Println("FATHER: I receive " + msg + " from child") // imprime na tela
25}

Canais Unidirecionais

Os canais unidirecionais são apenas para envio ou recebimento. Pode-se converter um bidirecional para unidirecional, mas não o contrário.

sendChannel := make(chan<- string) // canal apenas de envio

func send(ch chan<- string) // diz que o canal é só de envio

Fechando Canais

 1package main
 2
 3import (
 4	"fmt"
 5)
 6
 7func child(c chan int) {
 8	defer close(c) // Fecha o canal quando terminar a função
 9	for i := 0; i < 3; i++ {
10		c <- i // Envia informações algumas vezes pelo canal
11	}
12}
13func main() {
14	ch := make(chan int)
15	go child(ch)
16	for {
17		res, ok := <-ch  // res recebe a resposta e ok se o canal está aberto
18		if ok == false { // Se ok é falso, então o canal foi fechado
19			fmt.Println("Canal fechado", ok)
20			break
21		}
22		fmt.Println("Recebi", res, ok) // Vai imprimindo o que recebe
23	}
24}

Canais com buffer

Os canais com buffer são bloqueados caso a capacidade tenha sido excedida. Por exemplo. Se o seu buffer tem capacidade de 5 inteiros, e você quer enviar 10. Se seu programa não conseguiu fazer qualquer leitura, o próximo envio ficará bloqueado até que libere um espaço.

 1package main
 2
 3import (
 4	"fmt"
 5	"time"
 6)
 7
 8func child(ch chan int) {
 9	defer close(ch)
10	// Já vamos tentar enviar 5, mas somente pode ter 2 inteiros no buffer
11	for i := 0; i < 5; i++ {
12		ch <- i // Caso o buffer esteja cheio, essa linha fica bloqueada
13		fmt.Println("enviei", i, "com sucesso para o canal")
14	}
15}
16func main() {
17	// Canal com capacidade de 2 inteiros.
18	ch := make(chan int, 2) //Veja que temos um parâmetro extra
19	go child(ch)
20	time.Sleep(2 * time.Second)
21	for v := range ch {
22		fmt.Println("recebi o valor", v, "do canal")
23		time.Sleep(2 * time.Second)
24
25	}
26}

Usando select nos canais

podemos usar select (sintaxe tipo o switch-case) para múltiplas operações de enviar e receber dados via canais. O select bloqueia até alguma das operações recebeu algo ou enviou (de forma síncrona). Se mais de uma rotina de mensagem terminou, escolhe entre uma aleatoriamente.

Veja que no nosso exemplo vai sempre imprimir apenas o resultado do child2. Isso porque o child1 leva mais tempo para execução e o select pega o primeiro. Daí o programa fecha.

Ok, mas pra que eu usaria isso?

Existem várias situações em que precisamos atividades competitivas, por exemplo testar qual caminho de roteadores até um determinado site está mais rápido. O que retornar primeiro é o escolhido. Ou mesmo imagine que uma tabela de banco de dados está replicada em dois ou mais servidores. Pode-se fazer uma requisição de teste para cada servidor, o que retornar antes será a tabela a ser acessada.

 1package main
 2
 3import (
 4	"fmt"
 5	"time"
 6)
 7
 8func child1(ch chan string) {
 9	time.Sleep(6 * time.Second)
10	ch <- "do child1"
11}
12func child2(ch chan string) {
13	time.Sleep(3 * time.Second)
14	ch <- "do child2"
15
16}
17func main() {
18	ch1 := make(chan string)
19	ch2 := make(chan string)
20	go child1(ch1)
21	go child2(ch2)
22	// Podemos usar select
23	select {
24	case s1 := <-ch1:
25		fmt.Println(s1)
26	case s2 := <-ch2:
27		fmt.Println(s2)
28	}
29}

Deadlocks ¯\_(ツ)_/¯

Quando usamos canais, é importante ficarmos atentos para não gerar deadlocks. Se uma gorotine enviar dados para um canal esperando que outra gorotine a receba, mas essa outra rotina não recebe, o programa entrará em estado de deadlock. Veja o exemplo a baixo:

 1package main
 2
 3import (
 4	"fmt"
 5)
 6
 7func child(c chan bool) {
 8	a := <-c
 9	fmt.Printf("Recebi %t e estou enviando o contrário\n", a)
10	c <- !a
11}
12func main() {
13	ch := make(chan bool)
14	go child(ch)
15	response := <-ch
16	fmt.Printf("Acho que nunca vou receber porque não enviei nada %t\n", response)
17	fmt.Println("main function")
18}

se executarmos o programa, ele vai retornar algo como:

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
main.main()
        /home/saulo/dev/golang/src/aulas.com/distribuidos/deadlock/deadlock.go:15 +0x85

goroutine 6 [chan receive]:
main.hello(0x0)
        /home/saulo/dev/golang/src/aulas.com/distribuidos/deadlock/deadlock.go:8 +0x2c
created by main.main
        /home/saulo/dev/golang/src/aulas.com/distribuidos/deadlock/deadlock.go:14 +0x6f
exit status 2

Exclusão Múltua - Mutex

Como mecanismo de bloqueio, as gorotines utilizam o Mutex. Ele tenta garantir que apenas uma gorotine esteja executando a região crítica de um código em um dado momento, evitando que outras gorotines também tentem. Isso evita que a race condition ocorra.

Uma gorotine pode bloquear e desbloquear uma região crítica. essa região ficará bloqueada até que ela seja desbloqueada.

Exemplo

 1package main
 2
 3import (
 4	"fmt"
 5	"sync"
 6)
 7
 8// x é uma região crítica
 9var x = 0
10
11func increment(wg *sync.WaitGroup, m *sync.Mutex) {
12	// Estamos entrando em uma região crítica, então bloqueamos
13	m.Lock()
14	x = x + 1
15	// Ao sair da região crítica desbloqueamos
16	m.Unlock()
17	wg.Done()
18}
19func main() {
20	var w sync.WaitGroup
21	var m sync.Mutex
22	for i := 0; i < 1000; i++ {
23		w.Add(1)
24		go increment(&w, &m)
25	}
26	w.Wait()
27	fmt.Println("Valor final de x", x)
28}

Referências

• http://cocic.cm.utfpr.edu.br/progconcorrente/doku.php?id=go