Goroutines semaphores e pipeline pattern

O semáforo no Go é como se uma flag, que vai ser um channel que vai se comunicar com as routines, quando o semáforo apontar que deu tudo certo a ideia é q parem os canais.

func main() {
    channel := make(chan int)
    ok := make(chan bool)

    go func() {
        for i := 0; i < 10; i++ {
            channel <- i
        }
        ok <- true
    }()

    go func() {
        for i := 0; i < 10; i++ {
            channel <- i
        }
        ok <- true
    }()

    go func() {
        <-ok
        <-ok
        close(channel)
    }()

    for number := range channel {
        fmt.Println(number)
    }
}

Nesse exemplo a variável ok serve como o semáforo, perceba que na terceira função eu espero que a variável ok tenha sido esvaziada duas vezes para só então poder fechar o channel

Pipeline pattern

Até agora todas as vezes que usei o channel tinham sido de forma isolada, com a pipeline fazemos funções que retornam um channel e outras funções que recebem esse channel retornado e continuam o processamento. Aprendi a trabalhar com pipelines através da seguinte abordagem.

primeiro criei uma função generate que recebe numeros inteiros e retorna um channel de inteiros

func generate(numbers ...int) chan int {
    channel := make(chan int)

    go func() {
        for _, number := range numbers {
            channel <- number
        }
    }()

    return channel
}

E depois uma segunda função para dividir que recebe um channel de inteiros e retorna outro channel de inteiros

func divide(input chan int) chan int {
    channel := make(chan int)

    go func() {
        for number := range input {
            channel <- number / 2
        }
        close(channel)
    }()

    return channel
}

porem no caso dessa função enquanto eu vou atribuindo o valor para o channel eu já atribuo ele dividido por 2.

Minha função main ficou da seguinte forma

func main() {
    numbers := generate(2, 4, 6)
    result := divide(numbers)

    fmt.Println(<-result)
    fmt.Println(<-result)
    fmt.Println(<-result)
}

Onde em ordem os prints exibem 1 2 3. Mas o que tá acontecendo de fato é. A função generate recebe o numero 2, atribui ele ao seu channel, e tenta atribuir o valor 4 em seguida, não consegue pois o channel ainda não foi esvaziado, nesse momento a função divide inicia e recebe o valor do channel da primeira função e atribui ao seu channel, ele também vai tentar atribuir o próximo numero e não vai conseguir, então ele chama o println para poder esvaziar o seu channel e isso roda 3 vezes até os 3 números terem sido processados

Fun in

Imagine que temos 3 processos rodando em paralelo cada um com o seu próprio canal, agora temos uma segunda função que recebe os canais dos 3 processos e retorna um novo canal mandando os resultados dos 3 processos

O fun in é um afunilamento que faz com que não precisemos olhar para os 3 channels e sim para apenas 1 e termos o resultado dos 3

func main() {
    x := funnel(generateMsg("Hello"), generateMsg("World"))

    for i := 0; i < 10; i++ {
        fmt.Println(<-x)
    }
}

func generateMsg(s string) <-chan string {
    channel := make(chan string)
    go func() {
        for i := 0; ; i++ {
            channel <- fmt.Sprintf("String %s - Value: %d", s, i)
            time.Sleep(time.Duration(rand.Intn(255)) * time.Millisecond)
        }
    }()

    return channel
}

func funnel(channel1, channel2 <-chan string) <-chan string {
    channel := make(chan string)

    go func() {
        for {
            channel <- <-channel1
        }
    }()

    go func() {
        for {
            channel <- <-channel2
        }
    }()

    return channel
}

ps: o operador <- <- me deixou um pouco confuso no inicio porem ele significa que estamos pegando o valor do channel1 e passando para a variavel channel

o output dessa função vai ser

❯ go run main.go
String World - Value: 0
String Hello - Value: 0
String World - Value: 1
String Hello - Value: 1
String World - Value: 2
String World - Value: 3
String Hello - Value: 2
String Hello - Value: 3
String World - Value: 4
String World - Value: 5

então somente olhando para a variável x dá para ter o resultado dos dois channels que estão rodando em paralelo

Fan out

Basicamente o Fan out é o processo oposto do fun in, onde vamos ter apenas 1 canal mas queremos distribuir ele pro diversos canais

para fazer o fan out podemos reutilizar o código do algoritmo de divide basta que pegarmos o valor inicial e passarmos para a função divide mais de uma vez, por exemplo:

func main() {
    c := generate(5, 10)

    d1 := divide(c)
    d2 := divide(c)

    fmt.Println(<-d1)
    fmt.Println(<-d2)
}

func generate(numbers ...int) chan int {
    channel := make(chan int)
    go func() {
        for _, n := range numbers {
            channel <- n
        }
        close(channel)
    }()

    return channel
}

func divide(input chan int) chan int {
    channel := make(chan int)

    go func() {
        for number := range input {
            channel <- number / 2
        }
        close(channel)
    }()

    return channel
}