Utilizando uma estrutura concorrente para o disparo de emails

Recentemente aqui na OLX no time de accounts, recebemos uma solicitação para o disparo de emails para alguns clientes com o intuito de enviar um formulário para a obtenção de dados que possam ajudar a melhorar a experiencia do usuário.

Como essa solicitação era um envio de formulário para um grupo de usuários muito específico, não havíamos um serviço que atendesse essa demanda, o que tínhamos em mãos era um relatório contendo todos os emails desse grupo para os quais deveriam ser entregues os emails.

Aqui na OLX possuímos um serviço interno que já realiza envio de emails através de uma API, facilitando o disparo de emails, dessa forma o processo basicamente se limita a ler de um arquivo que contem todos emails que precisam ser enviados e o consumo dessa API. Simples não?

A resposta é sim, o envio é muito simples. O problema é que a base de envio desses emails é o relatório descrito anteriormente que possui aproximadamente 200 mil registros. Isso significa que para cada registro é necessário disparar um email. Acontece que o algoritmo de envio é simples, basta percorrer todos os emails e para cada email consumir a API e entregar o email solicitado e seguir para o próximo email, o problema é que seguir com esse algoritmo bloqueante irá demorar e muito o processamento dessa solicitação e tínhamos uma necessidade de entregar todos esses emails em um período curto.

Para isso foi concebido uma prova de conceito de um algoritmo para termos ideia do tempo que iria demorar através desse algoritmo simples, e para construir esse algoritmo utilizei o golang como ferramenta, pois sabia do potencial de concorrência da linguagem.

Todo os códigos apresentados nesse material encontra-se no github.

Iniciando o Projeto

Para executar e entendermos esse projeto, será necessário clonar o projeto do github, abra um terminal e execute:

git clone https://github.com/MAAARKIN/notification-async.git

Estrutura

A estrutura do projeto estará descrita dessa forma

notification-async/
┣ basic - Algoritmo bloqueante
┃ ┗ basic.go
┣ concurrent - Algoritmo concorrente
┃ ┗ concurrent.go
┣ model - Modelo que armazenará os dados de entrada
┃ ┗ option.go
┣ MOCK_DATA.csv - arquivo de mock com 1000 registros de email
┣ MOCK_DATA100.csv - arquivo de mock com 100 registros de email
┣ README.md
┣ base.csv
┣ go.mod
┗ main.go - simples CLI que designará para qual algoritmo será direcionado

Main.go

Nosso main.go será o entrypoint do projeto, tendo o papel de ser o responsável decidir se o código a ser executado será o bloqueante ou o concorrente a partir de dados que serão digitados pelo usuário.

Além disso, por ser o entrypoint do nosso projeto, será ele o responsável por criar um contexto que será responsável por finalizar nosso processo bloqueante ou concorrente caso venhamos a querer o fim do programa ANTES do termino de leitura do arquivo.

package main

import (
    "context"
    "flag"
    "log"
    "os"
    "os/signal"

    "github.com/MAAARKIN/notification-async/basic"
    "github.com/MAAARKIN/notification-async/concurrent"
    "github.com/MAAARKIN/notification-async/model"
)

func main() {
    // create a context
    ctx, cancel := context.WithCancel(context.Background()) //1
    defer cancel() //2
    // that cancels at ctrl+C
    go onSignal(os.Interrupt, cancel) //3

    // parse command line arguments
    op := new(model.Options)

    flag.StringVar(&op.Filename, "filename", "", "src file")
    flag.StringVar(&op.Event, "event", "", "event name")
    numberOfWorkers := flag.Int("workers", 2, "concurrent workers")
    flag.BoolVar(&op.Async, "async", false, "if the process will be async")
    flag.Parse() //4

    // check arguments
    if op.Filename == "" {
        log.Fatal("filename required")
    }

    if op.Event == "" {
        log.Fatal("event required")
    }

    if op.Async {
        // check arguments
        if numberOfWorkers == nil {
            log.Fatal("workers required")
        }
        concurrent.Start(ctx, *op, *numberOfWorkers) //5
    } else {
        basic.Start(ctx, *op)
    }

}

func onSignal(s os.Signal, cancel func()) {
    c := make(chan os.Signal, 1)
    signal.Notify(c, s)
    <-c
    cancel()
}

1. Criaremos um contexto que terá como responsabilidade a propagação de um possível cancelamento precoce, caso o usuário não queira aguardar a leitura de todo o arquivo. Mais pra frente detalhar sobre o contexto;
2. A criação do contexto a partir da função .WithCancel retorna dois valores, o primeiro será o contexto criado, com o qual iremos propagar nos nossos algoritmos e o segundo é uma função que ao ser chamada, emite um evento de cancelamento através de um canal, permitindo que possamos escutar esse sinal e possamos finalizar nosso programa;
3. A função onSignal terá o papel de ser executada de forma assíncrona onde iremos informar que caso o sinal de os.Interrupt (um Ctrl + C, irá emitir esse sinal de interrupção) seja executado, iremos chamar a função de cancelamento. E nos dois algoritmo iremos escutar esse evento de forma que possamos interromper os algoritmos e finalizar o programa. Vale salientar que iremos descrever com mais calma o papel da palavra reservada go mais a frente no processo concorrente;
4. O flag.Parse() tem o papel de levar os dados inseridos da linha de comando para os parâmetros dentro da variável op e dessa forma utilizar a estrutura de modelo Options;
5. O usuário que informar na chamada do comando de inicialização do sistema utilizar o valor do parametro de -async como verdadeiro, será chamado o algoritmo concorrente, caso contrário iremos consumir o algoritmo bloqueante.

Bloqueando o processo

Tanto nosso algoritmo bloqueante como o concorrente possuirá duas funcionalidades, o worker e o start.

O worker terá como papel ser a funcionalidade que possuirá o bloco que desejamos executar a chamada, ou seja, caso você queira realizar qualquer outro processo, seu bloco de código deve estar dentro do worker. No nosso caso, iremos construir um payload padrão com o dado do evento que será passado na chamada do terminal e iremos colocar um temporizador de 100 milli segundos para simular uma chamada a uma API, optei por não colocar a chamada via API para não ficar mais verboso nem expor dados de endpoint internos da OLX, mas um http client em go é fácil de encontrar na internet.

O start terá o papel de ser a funcionalidade que inicializará o bloco de código respectivo a cada algoritmo, que no nosso caso será bloqueante e concorrente. No caso do algoritmo bloqueante iremos contabilizar o tempo total de processamento, inicializar o arquivo que será passado via linha de comando e percorrer todo o arquivo de forma bloqueante e entregar ao worker cada linha lida para que a função possa realizar seu papel.

package basic

import (
    "bufio"
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/MAAARKIN/notification-async/model"
)

func worker(url string, event string) {
    payload := map[string]interface{}{
        "event":       event,
        "destination": map[string]string{"email": url},
        "source":      event,
    }

    data, _ := json.Marshal(payload)

    //my http client post here
    time.Sleep(100 * time.Millisecond) //to simulate the api.

    out := fmt.Sprintf("send event to %v, payload to send %v", url, string(data)) // do somethingg useful.
    fmt.Println(out)
}

func Start(ctx context.Context, op model.Options) {
    start := time.Now()

    csvfile, err := os.Open(op.Filename) //1
    if err != nil {
        log.Fatal(err) //2
    }

    defer csvfile.Close() //3

    scanner := bufio.NewScanner(csvfile) //4
    scanner.Split(bufio.ScanLines) //5

    for scanner.Scan() { //6
        select { //7
        case <-ctx.Done(): //8
            fmt.Println("worker finish")
            os.Exit(0) //9
        default:
            worker(scanner.Text(), op.Event) //10
        }
    }

    fmt.Printf("\n%2fs", time.Since(start).Seconds())
    fmt.Println()
}

1. Utilizamos a funcionalidade para inicializar e ter acesso ao arquivo desejado;
2. Caso a leitura do arquivo tenha algum problema, iremos finalizar o programa, não faz sentido continuar se não houver dados a processar ou se o arquivo apresentar problemas;
3. A palavra reservada defer no golang te garantirá que seu código seja executado ANTES que o bloco termine, garantindo assim que fechemos o arquivo ao final do processo;
4. O Scanner é será uma estrutura de armazenamento dos dados do arquivo num buffer para que possamos realizar a leitura;
5. Queremos que o scanner realize o split (a quebra dos dados) a partir de cada linha do arquivo;
6. a funcionalidade Scan() retorna um booleano caso haja linhas a serem lidas, parando somente quando não houver mais linhas, retornando assim um valor false e consequentemente finalizando o laço;
7. A palavra reservada select é uma uma palavra da linguagem utilizada para se trabalhar com channel, nesse caso ele realiza o bloqueio do código aguardando algum dos casos serem executados, similar ao switch case de outras linguagens mas voltado para canais. Vale salientar que se não existir o default o select mantem o código bloqueado até o canal receber informação, e caso o default exista e nenhum dos cases for atendido, então o default será executado, permitindo assim que o código continue. Enquanto o evento de cancelamento não for disparado, o default será executado;
8. Canal do contexto que escutará o evento de cancelamento ser executado para processar o bloco de código;
9. Funcionalidade do go para realizar a finalização do código;
10. Execução da função worker junto do scanner.Text() que é utilizado para obter a linha atual do arquivo.

Dentro do projeto notification-async teremos dois arquivos distintos de .csv com emails falsos que foram criados para simular um cenário de envio de emails, para nossos exemplos, utilizaremos o MOCK_DATA100.csv que contem 100 registros de emails falsos para testarmos nosso código.

Para executar nosso algoritmo executaremos o seguinte código no terminal

go run main.go -filename=MOCK_DATA100.csv -event=some_event -async=false

Esse projeto recebe como parametro alguns dados, como o arquivo que iremos ler, um nome de evento que irá estar presente no conteúdo que simulará o body do nosso serviço (que substituímos por um temporizador de 100ms) e o parametro -async serve pra designar para que nosso código main solicite para que o algoritmo bloqueante seja executado.

Resultado Bloqueante

Todos os testes realizados em minha maquina pessoal estão sempre em uma media de 10 a 11 segundos. Vale salientar que isso pode mudar de acordo com a maquina mas caso ocorra a repetição dos testes, será possível você identificar que o valor sempre irá diferenciar nas casas dos milli segundos, mantendo um valor muito próximo dos segundos.

...
send event to brackstraw2n@xinhuanet.com, payload to send {"destination":{"email":"brackstraw2n@xinhuanet.com"},"event":"teste","source":"teste"}
send event to nnosworthy2o@github.io, payload to send {"destination":{"email":"nnosworthy2o@github.io"},"event":"teste","source":"teste"}
send event to esmiley2p@opensource.org, payload to send {"destination":{"email":"esmiley2p@opensource.org"},"event":"teste","source":"teste"}
send event to gkording2q@desdev.cn, payload to send {"destination":{"email":"gkording2q@desdev.cn"},"event":"teste","source":"teste"}
send event to hgunthorpe2r@blogs.com, payload to send {"destination":{"email":"hgunthorpe2r@blogs.com"},"event":"teste","source":"teste"}

10.429105s

Context

Como pode ser observado dentro da função start o primeiro parâmetro, o context.Context será utilizado nos algoritmos para realizar o papel de "olheiro" do nosso sistema, ou seja, através dele teremos a possibilidade de emitir um sinal a partir da linha de comando para que possamos parar o programa caso a gente queira. Este evento de parada poderá ser executado através do 'Ctrl+C' no terminal para que por algum motivo precisemos parar o sistema. Não iremos explicar a fundo sobre o context pois este não é o objetivo, mas saiba que através do canal context.Done() é possível escutar o evento de cancelamento.

Processo Concorrente

Assim como no bloqueante, teremos duas funcionalidades, worker e start. Diferentemente do bloqueante, o worker aqui será executado através de goroutine que é uma funcionalidade de execução assíncrona, disponibilizada pela linguagem.

Por esse motivo, você verá que o worker terá alguns parâmetros diferentes, que são os canais e o contexto. Agora o processo de escutar o evento de cancelamento e parar a execução do código estará presente dentro da função worker e iremos explicar o motivo quando detalharmos o código.

Para evitar copiar códigos já mostrados anteriormente optarei por utilizar reticencias para descrever que é o mesmo código do bloqueante, sendo assim, basta verificar o código anterior. Também irei ignorar os imports pois eles podem ser visualizados no repositório do github. E irei modificar a ordem das funções para uma melhor didática.

package concurrent

func Start(ctx context.Context, op model.Options, numberOfWorkers int) {
    ...

    scanner := bufio.NewScanner(csvfile)
    scanner.Split(bufio.ScanLines)

    src := make(chan string) //1
    out := make(chan string) //2

    // use a waitgroup to manage synchronization
    var wg sync.WaitGroup //3

    // declare the workers
    for i := 0; i < numberOfWorkers; i++ { //4
        wg.Add(1) //5
        go func() { //6
            defer wg.Done() //7
            worker(ctx, out, src, op.Event) //8
        }()
    }

    go func() { //9
        for scanner.Scan() {
            src <- scanner.Text() //10
        }
        close(src) //11
    }()

    // drain the output
    go func() { //12
        for res := range out {
            fmt.Println(res)
        }
    }()

    // wait for worker group to finish and close out
    wg.Wait() //13
    close(out) //14

    fmt.Printf("\n%2fs", time.Since(start).Seconds())
    fmt.Println()
}

func worker(ctx context.Context, out chan<- string, src <-chan string, event string) {
    for { //15
        select { //16
        case url, ok := <-src: //17
            if !ok {
                return
            }

            payload := map[string]interface{}{
                "event":       event,
                "destination": map[string]string{"email": url},
                "source":      event,
            }
            ...
            out <- fmt.Sprintf("send event to %v, payload to send %v", url, string(data)) // do somethingg useful.
        case <-ctx.Done(): //18
            fmt.Println("worker finish")
            return
        }
    }
}

1. Para realizar a comunicação das linhas que serão lidas e entregues ao worker utilizaremos um canal chamado src. Vale salientar que a partir de agora o start estará sendo executado em uma rotina e o worker estará sendo executada em outra, por isso, necessitaremos dos canais.
2. Um canal para obter as respostas do worker de quais foram os registros enviados.
3. No golang para que as rotinas que foram criadas em paralelo sejam processadas, a rotina principal precisa estar ativa, ou seja, precisamos manter em algum momento um cenário onde falamos para que o nosso código aguarde que as rotinas executadas, sejam aguardadas até que finalizem. E para isso utilizaremos o sync.WaitGroup que tem o papel de conhecer quantas rotinas serão observadas para que possamos aguardar a finalização delas.
4. Iremos iterar num total de workers que serão definidos na entrada principal, ou seja, caso digamos que teremos um total de 5 workers trabalhando em paralelo, então esse laço irá iterar 5 vezes criando 5 workers.
5. Através da funcionalidade .Add(1) estamos informando para nosso WaitGroup que teremos mais uma goroutine a ser aguardada.
6. o go func é uma forma de criar uma goroutine de forma "embarcada" ou seja, podemos invocar uma função anônima de forma concorrente escrevendo a palavra reservada go + func() {...}.
7. Através do defer wg.Done() estamos informando para o WaitGroup que ao termino dessa nossa goroutine queremos que o WaitGroup seja notificado que uma das rotinas que estão sendo processadas já foi finalizada.
8. A chamada do worker com os novos parâmetros dentro de uma goroutine, ou seja esse bloco de código será executado de forma concorrente
9. Agora com os workers já criados e iniciados, iremos criar uma nova goroutine para percorrer o arquivo
10. Antes entregávamos a leitura da linha para dentro do worker, agora iremos entregar a leitura da linha para dentro de um canal, ou seja, iremos escrever essa linha atual que estamos iterando para dentro do canal através do símbolo ←, ou seja, se esse símbolo estiver a direita do nome do canal, significa que você estará escrevendo algo para dentro do canal, se estiver a esquerda do nome do canal, significa que você estará lendo os dados desse canal.
11. Após percorrer todo o arquivo e escrevermos tudo dentro do canal src, não iremos mais utilizar esse canal, sendo assim, devemos fechar esse canal através da funcionalidade close() que pertence a própria linguagem;
12. Iremos criar uma outra goroutine somente para ficar aguardando a leitura do canal de impressão dos dados processados dentro do worker, esse laço continuará repetindo enquanto houver dados a serem lidos do canal.
13. Através da funcionalidade .Wait() o WaitGroup bloqueia o código, aguardando que todos as funcionalidades wg.Done() sejam executadas, garantindo assim, que só iremos finalizar nossa rotina principal, após todas rotinas serem finalizadas;
14. Como chegamos ao final da leitura dos dados de saída do worker, iremos fechar o canal out. Vale salientar que esse bloco de código não será executado enquanto todas as goroutines declaradas no WaitGroup tenham sido processadas e o método .Wait() permita que esse código seja executado;
15. Dentro do worker iremos manter um laço "eterno" para ficar sempre escutando os canais.
16. Através do select iremos bloquear essa goroutine de forma que o bloco de código só seguirá se chegar uma mensagem no canal src ou se o contexto for cancelado. Vale salientar que o for não será iterado enquanto nenhum dos canais forem lidos, como falei anteriormente o select é bloqueante sem um default;
17. Através do símbolo ← a esquerda do canal src, estamos informando para o select que nesse case iremos aguardar a LEITURA do canal src, ou seja, toda vez que uma linha do arquivo for lida e escrita nesse canal, esse case será processado e executado, permitindo assim que o select seja desbloqueado e que possa ocorrer a próxima iteração do laço for;
18. A escuta do cancelamento do contexto foi transferida para dentro do worker pois caso o cancelamento ocorra, precisamos notificar TODOS as rotinas que foram criadas para que sejam canceladas através do return, que ao ser executado o return a função worker será finalizada, consequentemente finalizando também nosso laço "eterno". Permitindo assim que o defer wg.Done() ocorra e consequentemente desbloqueando o .Wait() e assim permitindo a finalização do nosso bloco de código do start;

Para executar nosso algoritmo executaremos o seguinte código no terminal

go run main.go -filename=MOCK_DATA100.csv -event=teste -async=true -workers=2

Resultado Concorrente

• Testes com 2 workers
  

  go run main.go -filename=MOCK_DATA100.csv -event=teste -async=true -workers=2
  ...
  send event to massandri2i@cocolog-nifty.com, payload to send {"destination":{"email":"massandri2i@cocolog-nifty.com"},"event":"teste","source":"teste"}
  send event to tberthon2k@photobucket.com, payload to send {"destination":{"email":"tberthon2k@photobucket.com"},"event":"teste","source":"teste"}
  send event to estruthers2l@marriott.com, payload to send {"destination":{"email":"estruthers2l@marriott.com"},"event":"teste","source":"teste"}
  send event to brackstraw2n@xinhuanet.com, payload to send {"destination":{"email":"brackstraw2n@xinhuanet.com"},"event":"teste","source":"teste"}
  send event to adepinna2m@microsoft.com, payload to send {"destination":{"email":"adepinna2m@microsoft.com"},"event":"teste","source":"teste"}
  send event to esmiley2p@opensource.org, payload to send {"destination":{"email":"esmiley2p@opensource.org"},"event":"teste","source":"teste"}
  send event to nnosworthy2o@github.io, payload to send {"destination":{"email":"nnosworthy2o@github.io"},"event":"teste","source":"teste"}
  send event to gkording2q@desdev.cn, payload to send {"destination":{"email":"gkording2q@desdev.cn"},"event":"teste","source":"teste"}
  send event to hgunthorpe2r@blogs.com, payload to send {"destination":{"email":"hgunthorpe2r@blogs.com"},"event":"teste","source":"teste"}
  
  5.133309s
  

• Testes com 4 workers
  

  go run main.go -filename=MOCK_DATA100.csv -event=teste -async=true -workers=4
  ...
  send event to djuckes2h@dropbox.com, payload to send {"destination":{"email":"djuckes2h@dropbox.com"},"event":"teste","source":"teste"}
  send event to dtippler2j@wsj.com, payload to send {"destination":{"email":"dtippler2j@wsj.com"},"event":"teste","source":"teste"}
  send event to adepinna2m@microsoft.com, payload to send {"destination":{"email":"adepinna2m@microsoft.com"},"event":"teste","source":"teste"}
  send event to brackstraw2n@xinhuanet.com, payload to send {"destination":{"email":"brackstraw2n@xinhuanet.com"},"event":"teste","source":"teste"}
  send event to estruthers2l@marriott.com, payload to send {"destination":{"email":"estruthers2l@marriott.com"},"event":"teste","source":"teste"}
  send event to tberthon2k@photobucket.com, payload to send {"destination":{"email":"tberthon2k@photobucket.com"},"event":"teste","source":"teste"}
  send event to hgunthorpe2r@blogs.com, payload to send {"destination":{"email":"hgunthorpe2r@blogs.com"},"event":"teste","source":"teste"}
  send event to gkording2q@desdev.cn, payload to send {"destination":{"email":"gkording2q@desdev.cn"},"event":"teste","source":"teste"}
  send event to esmiley2p@opensource.org, payload to send {"destination":{"email":"esmiley2p@opensource.org"},"event":"teste","source":"teste"}
  send event to nnosworthy2o@github.io, payload to send {"destination":{"email":"nnosworthy2o@github.io"},"event":"teste","source":"teste"}
  
  2.562593s
  

• Testes com 1 worker
  

  go run main.go -filename=MOCK_DATA100.csv -event=teste -async=true -workers=1
  ...
  send event to massandri2i@cocolog-nifty.com, payload to send {"destination":{"email":"massandri2i@cocolog-nifty.com"},"event":"teste","source":"teste"}
  send event to dtippler2j@wsj.com, payload to send {"destination":{"email":"dtippler2j@wsj.com"},"event":"teste","source":"teste"}
  send event to tberthon2k@photobucket.com, payload to send {"destination":{"email":"tberthon2k@photobucket.com"},"event":"teste","source":"teste"}
  send event to estruthers2l@marriott.com, payload to send {"destination":{"email":"estruthers2l@marriott.com"},"event":"teste","source":"teste"}
  send event to adepinna2m@microsoft.com, payload to send {"destination":{"email":"adepinna2m@microsoft.com"},"event":"teste","source":"teste"}
  send event to brackstraw2n@xinhuanet.com, payload to send {"destination":{"email":"brackstraw2n@xinhuanet.com"},"event":"teste","source":"teste"}
  send event to nnosworthy2o@github.io, payload to send {"destination":{"email":"nnosworthy2o@github.io"},"event":"teste","source":"teste"}
  send event to esmiley2p@opensource.org, payload to send {"destination":{"email":"esmiley2p@opensource.org"},"event":"teste","source":"teste"}
  send event to gkording2q@desdev.cn, payload to send {"destination":{"email":"gkording2q@desdev.cn"},"event":"teste","source":"teste"}
  send event to hgunthorpe2r@blogs.com, payload to send {"destination":{"email":"hgunthorpe2r@blogs.com"},"event":"teste","source":"teste"}
  
  10.241959s
  

Conclusão

É possível visualizar que para esse tipo de cenário a estrutura concorrente diminuiu drasticamente o tempo de processamento do algoritmo mas vale notificar com somente 1 worker não vale a pena utilizar o algoritmo concorrente pois estaremos só tornando nosso código mais verboso para uma estratégia onde somente uma rotina será executada.

Um outro motivo de se utilizar esse padrão para o consumo concorrente é que caso eu não limitasse, poderia gerar um aumento de recurso pois para cada linha estaríamos abrindo rotinas concorrentes, então em um cenário com um arquivo de 1000 linhas, teríamos 1000 rotinas em paralelo e talvez para processar essa quantidade de rotinas seria necessário mais recursos do computador, ou seja, não teríamos limites de processos ocorrendo de forma concorrente.

Sem falar que caso tivéssemos consumindo uma API, poderíamos derrubar ou até mesmo acabar tendo que aumentar os recursos dessa API pelo uso desenfreado de rotinas em paralelo. Essa estratégia de limitar a quantidade de rotinas processando em paralelo é conhecida como Worker Pool, muito utilizada para gerenciamento de conexões de banco de dados.

Se bem utilizado, a programação concorrente pode ajudar e muito nos processos das empresas. E obvio que a linguagem fornece uma facilidade enorme em utilizar esse tipo de programação, tendo em vista que para todo esse nosso estudo, não foi necessário nenhum tipo de serviço ou dependência de uma biblioteca terceira. Ou seja, a própria linguagem irá te fornecer o necessário para transformar teu algoritmo em um cenário concorrente de forma natural.