Para melhor entendimento desse post, indico a leitura de: Olá mundo da programação concorrente.
Este post é uma tradução da parte que fala sobre o módulo Task do livro Concurrent Data Processing in Elixir de Svilen Gospodinov.
Conforme o livro Concurrent Data Processing in Elixir, o processamento de dados é uma parte essencial de muitas aplicações de software. Na verdade, a maioria dos engenheiros nem sequer pensa nisso como algo separado da programação. Mas se você está transformando informações de alguma forma, por exemplo, ao fazer relatórios, agregação de dados ou análises, então você está fazendo processamento de dados.
Graças à Máquina Virtual Erlang (também conhecida como BEAM), todos que usam Elixir se beneficiam de seu incrível modelo de concorrência, sendo particularmente adequado para tarefas concorrentes de longa duração. Como resultado, você verá que a linguagem Elixir oferece mais maneiras de realizar trabalhos concorrentes do que outras linguagens.
Embora isto seja uma coisa boa, também pode ser um desafio encontrar a ferramenta certa para o trabalho. Algumas tarefas são uma excelente escolha para Flow, enquanto outras são perfeitas para Broadway. Às vezes é mais fácil usar apenas o GenStage, ou mesmo o módulo Task.
As técnicas certas o ajudarão a simplificar seu produto, melhorar o desempenho de seu código e tornar sua aplicação mais resistente a erros e ao aumento da carga de trabalho.
Introdução
Desde o início da indústria da computação, os fabricantes de hardware e cientistas da computação têm tentado tornar os computadores mais rápidos na execução de programas. No início, a multithreading era a única maneira de alcançar a concorrência, sendo a capacidade de executar duas ou mais tarefas de programação e alternar entre elas para coletar os resultados. Foi assim que os computadores pareciam estar fazendo muitas coisas ao mesmo tempo, quando, na verdade, eram simplesmente multitarefas.
As CPUs multi-core mudaram isso. Trouxeram paralelismo e permitiram que as tarefas funcionassem lado a lado, independentemente, o que aumentou significativamente o desempenho dos sistemas. Seguiram-se arquiteturas de multiprocessadores, permitindo ainda maior concorrência e paralelismo ao suportar duas ou mais CPUs em uma única máquina. A figura abaixo mostra uma comparação simples entre a concorrência em uma CPU de núcleo único e em uma CPU de núcleo duplo que permite o paralelismo:
Naturalmente, as ferragens de última geração vêm sempre com uma etiqueta de preço elevado. Mas com o advento da computação em nuvem, as coisas mudaram mais uma vez. Hoje em dia, é possível executar código em serviços na nuvem usando máquinas virtuais com dezenas de núcleos de CPU, sem a necessidade de comprar e manter qualquer hardware físico.
Estes avanços são importantes para nós como engenheiros de software. Queremos escrever um software que tenha bom desempenho e funcione rapidamente. Afinal de contas, ninguém gosta de carregar telas e esperar que o computador termine. Entretanto, a execução do código em um sistema de processador multi-core não o torna automaticamente eficiente. De modo a aproveitar ao máximo os recursos computacionais disponíveis, precisamos escrever software tendo em mente a concorrência e o paralelismo. Felizmente, as modernas linguagens de programação tentam nos ajudar o máximo possível, e a linguagem Elixir não é exceção. De fato, graças ao Erlang, a Máquina Virtual Erlang (BEAM) e a Plataforma de Telecomunicações Abertas (OTP - Open Telecom Platform), o Elixir é uma excelente escolha para a construção de aplicações e processamento de dados concorrentes.
Vamos abordar as ferramentas mais populares para a realização de trabalhos concorrentes usando Elixir. Vamos verificar os prós e contras de cada uma e ver como funcionam na prática. Algumas delas, como o módulo Task e o GenServer, já vêm com Elixir. Os outros - GenStage, Flow e Broadway - estão disponíveis como bibliotecas autônomas no registro de pacotes Hex.pm. Saber como utilizar cada uma destas ferramentas ajudará você a alavancar a concorrência da maneira mais eficaz e resolver até mesmo os problemas mais desafiadores. Ao longo do caminho, você também aprenderá como construir aplicações tolerantes a falhas, recuperar-se de falhas, usar a contrapressão para lidar com recursos limitados do sistema e muitas outras técnicas úteis.
Primeiro, vamos analisar o módulo Task, que faz parte da biblioteca padrão da linguagem Elixir. Ele possui um poderoso conjunto de características que o ajudarão a executar o código concorrentemente. Você também vai ver como lidar com erros e evitar que a aplicação trave quando uma tarefa concorrente trava.
Concorrência fácil com o módulo Task
Para executar o código concorrentemente em Elixir, você tem que iniciar um processo e executar seu código dentro desse processo. Você também pode precisar recuperar o resultado e usá-lo para algo mais. Elixir fornece uma função de baixo nível e uma macro para fazer isso - spawn/1 e receive. Entretanto, usá-los pode ser complicado na prática e você provavelmente acabará com um monte de código repetitivo.
Elixir também vem com um módulo chamado Task, que simplifica significativamente o início de processos concorrentes. Ele fornece uma abstração para executar o código concorrentemente, recuperando resultados, manipulando erros e iniciando uma série de processos. Ele possui muitas características com uma API concisa, de modo que raramente (ou nunca) há a necessidade de usar spawn/1 e receive.
Vamos conhecer tudo o que o módulo Task tem a oferecer:
- Como iniciar tarefas;
- Diferentes maneiras de recuperar resultados;
- Como lidar com processamento de grandes listas de dados;
- Como lidar com falhas;
- Como funciona a ligação de processos em Elixir;
- Como usar um dos módulos
Supervisorintegrados para isolar falhas de processo; - Discussão sobre a abordagem de Elixir para o tratamento de erros.
Antes de mergulharmos, vamos criar um projeto Elixir para trabalhar primeiro e nos familiarizarmos com algumas das ferramentas de desenvolvimento que vamos usar ao longo deste post.
O que é um processo de Elixir?
Os processos na Elixir são processos Erlang, uma vez que a Elixir funciona na Máquina Virtual Erlang. Ao contrário dos processos do sistema operacional, eles são muito leves em termos de uso de memória e rápidos de iniciar. A VM de Erlang sabe como executá-los concorrentemente e em paralelo (quando uma CPU multi-core está presente). Como resultado, ao utilizar processos, você obtém concorrência e paralelismo de graça.
Criando o projeto
Vamos criar uma aplicação chamada sender e fingir que estamos enviando e-mails para endereços de e-mail reais. Vamos usar o módulo Task mais tarde para desenvolver algumas de suas funcionalidades.
Primeiramente, vamos usar a ferramenta de linha de comando mista para a montagem de nosso novo projeto:
$ mix new sender --sup
Isto cria um diretório para o nosso projeto sender com um monte de arquivos e pastas dentro. Note que também utilizamos o argumento --sup, que criará uma aplicação com uma árvore de supervisão. Veremos sobre árvores de supervisão mais adiante.
Para manter nosso projeto e nossos exemplos simples, não vamos enviar e-mails de verdade. Entretanto, ainda precisamos de alguma lógica comercial para nossas experiências. Podemos usar a função Process.sleep/1 para fingir que estamos enviando um e-mail, que normalmente é uma operação lenta e pode levar alguns segundos para ser concluída. Quando chamado com um número inteiro, o Process.sleep/1 interrompe o processo atual durante um determinado período em milissegundos. Isto é muito útil, porque você pode usá-lo para simular um código que leva muito tempo para ser completo. Você também pode usá-lo para testar vários casos extremos, como veremos mais tarde. Naturalmente, em aplicações de produção real, você substituirá isto por sua lógica comercial real. Mas por enquanto, vamos fingir que estamos fazendo um trabalho muito intensivo.
Abra seu arquivo sender/lib/sender.ex e adicione o seguinte:
def send_email(email) do
Process.sleep(3000)
IO.puts("Email to #{email} sent")
{:ok, "email_sent"}
end
Ao chamar esta função, a execução será suspensa por três segundos e uma mensagem será impressa, o que será útil para a depuração. Também retorna uma tupla {:ok, "email_sent"} para indicar que o email foi enviado com sucesso.
Agora que tudo está pronto, podemos começar. Sugiro que você mantenha uma sessão de terminal com o IEx aberta e seu editor de texto favorito ao lado, para você poder fazer e executar mudanças à medida que avançamos.
Iniciando Tasks e Recuperando Resultados
Antes de saltar para o módulo Task, vamos ver como as coisas estão funcionando. Vamos chamar a função send_email/1 do IEx, e passar um endereço de email fictício como um argumento.
iex> Sender.send_email("hello@world.com")
Email to hello@world.com sent
{:ok, "email_sent"}
Você notou o atraso/delay? Tivemos que esperar três segundos até vermos a produção e o resultado impresso. Na verdade, mesmo o iex> prompt não estava aparecendo. Vamos adicionar outra função, notify_all/1:
Em sender/lib/sender.ex:
def notify_all(emails) do
Enum.each(emails, &send_email/1)
end
A função notify_all/1 usa Enum.each/2 para iterar sobre os emails variados, que é uma lista de strings. Para cada item da lista, vamos chamar a função send_email/1. Vamos testar esta função no IEx, mas primeiro precisamos de alguns dados de teste. Crie um arquivo .iex.exs na pasta principal do projeto sender no mesmo local onde está o mix.exs. Adicione o seguinte:
Em sender/.iex.exs:
emails = [
"hello@world.com",
"hola@world.com",
"nihao@world.com",
"konnichiwa@world.com",
]
Salve o arquivo, saia do IEx e inicie-o novamente com iex -S mix. Digite emails e pressione enter:
iex(1)> emails
["hello@world.com", "hola@world.com", "nihao@world.com",
"konnichiwa@world.com"]
Isto lhe poupará muita digitação. Todo o código Elixir em .iex.exs será executado quando o IEx iniciar, portanto, isto persistirá entre as sessões do IEx. Agora vamos usar os dados de teste com notify_all/1. Você pode adivinhar quanto tempo levará para enviar todos os emails? Vamos descobrir:
iex> Sender.notify_all(emails)
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent
:ok
Foram necessárias quatro chamadas para send_email/1 e cerca de doze segundos para completar. Só de esperar a saída no IEx parecia uma eternidade. Isto não é nada bom!!! À medida que a nossa base de usuários cresce, levará uma eternidade para enviar nossos emails.
Não se desespere - podemos acelerar significativamente nosso código usando o módulo Task. No entanto, antes de saltarmos para dentro dele, vamos tirar um momento para falar sobre dois conceitos importantes na programação: código síncrono e assíncrono.
Código síncrono e assíncrono
Tanto o código síncrono como assíncrono são executados aos mesmo tempo, a diferença é que o código assíncrono não há uma ordem específica e por isso é considerado concorrente, diferentemente do síncrono que há uma sequencia específica.
Por padrão, quando você executa algum código em Elixir, você tem que esperar que ele seja concluído. Você não pode fazer nada enquanto isso, e você obtém o resultado assim que o código tiver terminado. O código é executado de forma síncrona e é algumas vezes chamado de código bloqueador (blocking code).
O oposto disto é executar o código de forma assíncrona. Neste caso, você pede ao tempo de execução da programação para executar o código, mas continua com o resto do programa. O código assíncrono roda em segundo plano porque a aplicação continua rodando, como se nada tivesse acontecido. Eventualmente, quando o código assíncrono termina, você pode recuperar o resultado. Como o código assíncrono não bloqueia a execução principal do programa, ele é chamado de código não-bloqueador (non-blocking code). O código assíncrono também é concorrente, já que nos permite continuar fazendo outros trabalhos. A figura abaixo ilustra a diferença entre código síncrono e assíncrono em um programa de panificação:
- Mix flour = Misture a farinha
- Rest the dough = Descanse a massa
- Pre-heat oven = Pré-aqueça o forno
- Bake = Assar
Uma vez que apenas a etapa final de cozimento (Bake) requer que o forno seja pré-aquecido (Pre-heat), você pode fazer o pré-aquecimento do forno de forma assíncrona e fazer algo mais enquanto isso. Em comparação com a versão síncrona, isto diminui significativamente o tempo necessário para completar o conjunto de instruções.
Você já deve ter adivinhado que a função notify_all/1 está enviando cada email de forma síncrona. É por isso que está demorando tanto para ser concluída. Para melhorar isto, vamos converter nosso código para ser executado de forma assíncrona. Graças ao módulo Task do Elixir, precisaremos apenas de algumas pequenas mudanças para conseguir isso. Vamos ver como funciona.
Iniciando processos
O módulo Task contém uma série de funções muito úteis para executar código de forma assíncrona e concorrente. Uma delas é start/1. Ele aceita uma função como um argumento, e dentro dessa função devemos fazer todo o trabalho que pretendemos fazer. Vamos experimentar isso rapidamente no IEx:
iex> Task.start(fn -> IO.puts("Hello async world!") end)
Hello async world!
{:ok, #PID<0.266.0>}
Você viu a mensagem impressa instantaneamente porque é uma operação rápida, mas o resultado real foi {:ok, #PID<0.266.0>}. Retornar uma tupla como {:ok, result} ou {:erro, message} é uma prática comum no Elixir. O resultado foi :ok para o sucesso e #PID<0.266.0>. PID significa identificador de processo (process identifier) - um número que identifica de forma única um processo Elixir.
Já temos o send_email/1 pronto, por isso podemos usar o Task.start/1 para chamá-lo. Vamos fazer algumas mudanças:
Em sender/lib/sender.change1.ex:
def notify_all(emails) do
Enum.each(emails, fn email ->
Task.start(fn ->
send_email(email)
end)
end)
end
Em seguida, recompilar e executar notify_all/1:
iex> Sender.notify_all(emails)
:ok
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent
Isto deve ser significativamente mais rápido - de fato, quatro vezes mais rápido! Todas as funções foram chamadas concorrentemente e terminadas ao mesmo tempo, imprimindo a mensagem de sucesso como esperávamos.
Recuperando o resultado de uma tarefa
Task.start/1 tem uma limitação por projeto: não retorna o resultado da função que foi executada. Isto pode ser útil em alguns casos, mas na maioria das vezes você precisa do resultado para alguma outra coisa. Seria ótimo se modificássemos nosso código e retornássemos um resultado significativo quando todos os emails forem enviados com sucesso.
Para recuperar o resultado de uma função, você tem que usar o Task.async/1. Ela retorna uma estrutura %Task{} que você pode atribuir a uma variável para uso posterior. Você pode experimentá-la no IEx assim:
iex> task = Task.async(fn -> Sender.send_email("hello@world.com") end)
%Task{
owner: #PID<0.145.0>,
pid: #PID<0.165.0>,
ref: #Reference<0.713486762.1657274369.63141>
}
O código send_email/1 está agora rodando em segundo plano. Enquanto isso, estamos livres para fazer outros trabalhos, conforme necessário. Você pode adicionar mais lógica comercial ou até mesmo iniciar outras tarefas. Quando você precisar do resultado real da tarefa, você pode recuperá-la usando a variável task. Vamos dar uma olhada mais de perto no que esta variável contém:
-
owner(proprietário) é o PID do processo que iniciou o processo Task. -
pidé o identificador do próprio processo Task. -
refé a referência do monitor de processo (process monitor reference).
Não vamos nos aprofundar no monitoramento de processos (Process monitoring) aqui. Entretanto, vale a pena saber que você pode monitorar um processo e receber notificações dele usando um valor de referência - por exemplo, quando e como um processo sai.
Para recuperar o resultado da tarefa, você pode usar tanto o Task.await/1 ou o Task.yield/1 que aceita uma estrutura Task como um argumento. Há uma diferença importante na forma de trabalho do await/1 e do yield/1, portanto, você tem que escolher sabiamente. Ambos param o programa e tentam recuperar o resultado da tarefa. A diferença vem da maneira como eles lidam com os tempos de espera do processo (process timeouts).
Os tempos de espera do processo garantem que os processos não fiquem presos para sempre. Para mostrar como eles funcionam, vamos aumentar o tempo que a função send_email/1 leva para 30 segundos, apenas temporariamente:
def send_email(email) do
Process.sleep(30_000)
IO.puts("Email to #{email} sent")
{:ok, "email_sent"}
end
Em seguida, recompile/0 no IEx . Agora vamos executar o código de forma assíncrona e canalizar o resultado da task em await/1:
iex> Task.async(fn -> Sender.send_email("hi@world.com") end) |> Task.await()
Após cinco segundos, você receberá uma exceção semelhante a esta:
** (exit) exited in: Task.await(%Task{owner: #PID<0.144.0>,
pid: #PID<0.151.0>,
ref: #Reference<0.2297312895.3696492546.156249>}, 5000)
** (EXIT) time out
(elixir) lib/task.ex:607: Task.await/2
Ao utilizar o await/1, esperamos que uma tarefa termine dentro de um certo período de tempo. Por padrão, este tempo é definido para 5000ms, que é de cinco segundos. Você pode mudar isso passando um inteiro com a quantidade de milissegundos como segundo argumento, por exemplo Task.await(task, 10_000). Você também pode desativar o tempo limite passando o átomo :infinity.
Em comparação, o Task.yield/1 simplesmente retorna nulo se a tarefa não tiver sido concluída. O timeout de yield/1 também é de 5000ms, mas não causa uma exceção e um travamento (crash). Você também pode fazer Task.yield(task) repetidamente para verificar um resultado, o que não é permitido por await/1. Uma tarefa concluída retornará {:ok, resultado} ou {:exit, reason}. Você pode ver isto em ação:
iex> task = Task.async(fn -> Sender.send_email("hi@world.com") end)
%Task{
owner: #PID<0.135.0>,
pid: #PID<0.147.0>,
ref: #Reference<0.3033103973.1551368196.24818>
}
iex> Task.yield(task)
nil
iex> Task.yield(task)
nil
Email to hi@world.com sent
iex> Task.yield(task)
{:ok, {:ok, "email_sent"}}
iex> Task.yield(task)
nil
Esta saída ilustra o que acontece quando você chama Task.yield/1 enquanto a tarefa ainda está em execução - você recebe nil (nulo). Assim que a mensagem de sucesso é impressa, recebemos {:ok, {:ok, "e-mail_sent"}} como esperado. Você também pode usar yield/2 e fornecer seu próprio timeout (tempo de espera), de modo semelhante a await/2, mas a opção :infinity não é permitida. Note que você só receberá o resultado de yield/1 uma vez, e as chamadas subsequentes retornarão nil.
Você pode estar se perguntando o que acontece se nossa tarefa está emperrada e nunca termina? Enquanto await/1 se encarrega de parar a tarefa, o yield/1 a deixará em funcionamento. É uma boa idéia parar a tarefa manualmente, chamando o Task.shutdown(task). A função shutdown/1 também aceita um timeout e dá ao processo uma última chance de ser concluído, antes de pará-lo. Se ele for concluído, você receberá o resultado normalmente. Você também pode parar um processo imediatamente (e de forma bastante violenta) usando o átomo :brutal_kill como segundo argumento.
Como você pode ver, usar yield/1 e shutdown/1 é um pouco mais trabalhoso do que await/1. Qual deles usar depende de seu caso de uso. Muitas vezes, o timeout (tempo de espera) da tarefa justifica uma exceção, caso em que o await/1 será mais conveniente de usar. Sempre que você precisar de mais controle sobre o timeout e o encerramento, você pode mudar para yield/1 e shutdown/1. A parte importante é que, após utilizar async/1, você deve sempre processar o resultado, usando ou await/1, ou yield/1 seguido de shutdown/1.
Para a nossa lógica notify_all/1, vamos usar o await/1 para simplificar. Lembre-se de reverter nossa mudança anterior no send_email/1 e definir Process.sleep/1 de volta para 3000ms:
def send_email(email) do
Process.sleep(3000)
Agora vamos trocar Task.start/1 para Task.async/1:
Em sender/lib/sender.change2.ex:
def notify_all(emails) do
emails
|> Enum.map(fn email ->
Task.async(fn ->
send_email(email)
end)
end)
|> Enum.map(&Task.await/1)
end
Observe que também estamos usando Enum.map/2 em vez de Enum.each/2, porque queremos mapear cada seqüência de e-mails para sua estrutura de tarefas correspondente. A estrutura é necessária para recuperar o resultado de cada processo.
Usamos aqui a sintaxe da função shorthand (abreviação) do Elixir - &Task.await/1. Se você não está familiarizado com ela, é simplesmente equivalente à escrita:
Enum.map(fn task ->
Task.await(task)
end)
Vamos experimentar as últimas mudanças no IEx:
iex> Sender.notify_all(emails)
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent
[ok: "email_sent", ok: "email_sent", ok: "email_sent", ok: "email_sent"]
A função retornou uma lista de resultados. Para cada tarefa, você tem uma tupla ok: "email_sent", que é o que a função send_email/1 retorna. Em sua lógica comercial, você pode retornar um {:erro, "mensagem de erro"} quando algo der errado. Então você será capaz de coletar o erro e potencialmente tentar novamente a operação ou fazer algo mais com o resultado.
A criação de tarefas a partir de listas de itens é na verdade muito comum em Elixir. Em seguida, vamos usar uma função especificamente projetada para fazer isso. Ela também oferece uma série de recursos adicionais, especialmente úteis quando se trabalha com grandes listas.
Keyword Lists em Elixir
Elixir tem uma notação especial para listas contendo tuplas de chave-valor, também conhecidas como keyword lists (listas de palavras-chave). Cada item da lista de palavras-chave deve ser uma tupla de dois elementos. O primeiro elemento é o nome da chave, que deve ser um átomo. O segundo é o valor e pode ser de qualquer tipo.
Elas são exibidas no IEx sem as chaves (curly braces), assim [ok: "email_sent"]. Isto é equivalente a [{:ok, "email_sent"}].
Gerenciamento de séries de tarefas
Imaginemos que temos um milhão de usuários, e queremos enviar um e-mail a todos eles. Podemos usar Enum.map/2 e Task.async/1 como fizemos antes, mas iniciar um milhão de processos colocará uma pressão repentina sobre os recursos de nosso sistema. Isso pode degradar o desempenho do sistema e potencialmente fazer com que outros serviços não respondam. Nosso provedor de serviços de e-mail também não ficará feliz, pois também colocamos muita pressão sobre a infra-estrutura de e-mail deles.
Por outro lado, não queremos enviar e-mails um a um, porque é lento e ineficiente. Parece que estamos em uma encruzilhada e, seja qual for o caminho que tomemos, acabamos em perigo. Não queremos escolher entre desempenho e confiabilidade - queremos ser capazes de executar processos do módulo Task para alavancar a concorrência, mas garantir que não sobrecarregamos nossos recursos de sistema à medida que dimensionamos nosso produto e aumentamos nossa base de usuários.
A solução para nossos problemas é async_stream/3. É outra função muito útil do módulo Task que é projetado para criar processos de tarefas a partir de uma lista de itens. Para cada item da lista, async_stream/3 iniciará um processo e executará a função que fornecemos para processar o item. Funciona como Enum.map/2 e Task.async/2 combinados, com uma grande diferença: você pode definir um limite no número de processos em execução ao mesmo tempo. A figura abaixo ilustra como isto funciona:
Neste exemplo, o limite de concorrência é fixado em quatro, portanto, mesmo que você tenha uma lista de cem itens, no máximo apenas quatro processos serão executados de forma concorrente em um determinado momento. Este é um exemplo de tratamento de contrapressão/retaguarda (back-pressure), que podemos discutir em profundidade em um próximo post, falando sobre pipelines de processamento de dados com GenStage. Por enquanto, tudo o que você precisa saber é que esta estratégia de manuseio de processos é ótima para evitar picos repentinos de uso do sistema. Em outras palavras, você obtém a concorrência e o desempenho sem sacrificar a confiabilidade.
Como o nome da função sugere, async_stream/3 retorna um Stream. Streams em Elixir são estruturas de dados que realizam uma ou mais operações que não funcionam imediatamente, somente quando explicitamente informado. É por isso que às vezes são chamados de listas preguiçosas. Vamos ver o que acontece quando executamos esta função a partir do IEx:
iex> Task.async_stream(emails, &Sender.send_email/1)
#Function<1.35903181/2 in Task.build_stream/3>
Em vez do resultado habitual, recebemos uma função, que vai criar um Stream. É importante entender como funcionam os fluxos, então vamos dar outro exemplo rápido no IEx:
iex> Stream.map([1, 2, 3], & &1 * 2)
#Stream<[
enum: [1, 2, 3],
funs: [#Function<49.33009823/1 in Stream.map/2>]
]>
Se tivéssemos usado Enum.map/2 a função teria retornado [2, 4, 6]. Em vez disso, o fluxo de resultados contém simplesmente a entrada inicial e uma lista de operações funs. Estas operações podem ser executadas posteriormente. Como tanto o Stream quanto o Enum implementam o protocolo Enumerable, muitas funções Enum têm uma alternativa preguiçosa no módulo Stream.
Uma maneira de executar um fluxo é usar a função Stream.run/1. Entretanto, o Stream.run/1 sempre retorna :ok, então só é útil quando você não está interessado no resultado final. Ao invés disso, você pode usar Enum.to_list/1, que tentará converter o fluxo para uma estrutura de dados List. Como resultado desta conversão, todas as operações no fluxo serão executadas, e o resultado será retornado como uma lista. Outras funções no módulo Enum também forçarão o fluxo a ser executado, como Enum.reduce/3. Você pode usá-las se pretender trabalhar mais com o resultado.
Agora, vamos atualizar o notify_all/1 para usar o async_stream/3. Desta vez, porém, vamos executar o stream usando Enum.to_list/1:
Em sender/lib/sender.change3.ex:
def notify_all(emails) do
emails
|> Task.async_stream(&send_email/1)
|> Enum.to_list()
end
E dê uma olhada no IEx, mas não se esqueça de rodar recompile/0 primeiro:
iex> Sender.notify_all(emails)
Email to hello@world.com sent
Email to hola@world.com sent
Email to nihao@world.com sent
Email to konnichiwa@world.com sent
[
ok: {:ok, "email_sent"},
ok: {:ok, "email_sent"},
ok: {:ok, "email_sent"},
ok: {:ok, "email_sent"}
]
Por que async_stream retorna um Stream?
Os Streams são projetados para emitir uma série de valores, um por um. Assim que o Task.async_stream/3 descobre que um processo de tarefa terminou, ele emitirá o resultado e tomará o próximo elemento da entrada, iniciando um novo processo. Isto significa que ele pode manter uma série de eventos concorrentes, o que é um dos benefícios do async_stream/3. Você também pode usar todas as outras funções do módulo Stream para compor fluxos complexos de processamento de dados.
Como você pode ver, a saída é semelhante à do exemplo Task.async/2. Entretanto, dependendo de quantos núcleos lógicos (threads) sua máquina tem, o tempo que leva para que a função seja concluída pode ser diferente.
Como mencionamos anteriormente, async_stream/3 mantém um limite em quantos processos podem ser executados ao mesmo tempo. Por padrão, este limite é definido para o número de núcleos lógicos disponíveis no sistema. Anteriormente utilizávamos o Task.async/2 para iniciar manualmente um processo para cada um dos quatro itens. Isto significa que se você tiver uma CPU com menos de quatro núcleos lógicos, o async_stream/3 parecerá mais lento. Você pode alterar facilmente este comportamento padrão através do parâmetro opcional max_concurrency. Vamos definir o max_concurrency para 1 temporariamente:
|> Task.async_stream(&send_email/1, max_concurrency: 1)
Quando você tentar as novas mudanças novamente, você verá que os e-mails são enviados um a um. Isto não é muito útil na prática, mas demonstra como funciona o max_concurrency. Você pode reverter a mudança ou defini-la para um número ainda maior. Em aplicações de produção, você pode comparar (benchmark) se o max_concurrency funciona melhor para seu caso de uso, levando em consideração os recursos de seu sistema e a necessidade de desempenho.
Outra opção que precisa ser mencionada é :ordered. Atualmente, async_stream/3 assume que queremos os resultados na mesma ordem em que foram originalmente ordenados. Esta preservação da ordem pode potencialmente retardar nosso processamento, pois async_stream/3 esperará que um processo lento seja concluído antes de passar para o próximo.
Em nosso caso, só precisamos dos resultados para verificar se um e-mail foi enviado com sucesso ou não. Não necessariamente precisamos deles exatamente na mesma ordem. Podemos potencialmente agilizar as coisas desativando a ordenação assim:
|> Task.async_stream(&send_email/1, ordered: false)
Agora o async_stream/3 não ficará parado se um processo estiver demorando mais do que outros.
Processos iniciados por async_stream/3 também estão sujeitos a timeouts (tempo de resposta), assim como aqueles iniciados por start/1 e async/2. O parâmetro opcional :timeout é suportado e o valor padrão é de 5000ms. Quando uma tarefa atinge o timeout, ela produzirá uma exceção, interrompendo o stream/fluxo e interrompendo o processo atual. Este comportamento pode ser alterado usando um argumento opcional :on_timeout, que você pode definir para :kill_task. Este argumento é similar ao :brutal_kill suportado pelo Task.shutdown/2. Segue um exemplo:
|> Task.async_stream(&send_email/1, on_timeout: :kill_task)
Como resultado do uso da opção :kill_task, quando um processo termina com um timeout, o async_stream/3 irá ignorá-lo e continuará normalmente.
Um processo pode falhar por muitas razões. Falamos de exceções de timeout, que acontecem quando os processos levam muito tempo para serem concluídos. Isto pode ser porque o trabalho que ele está fazendo consome muito tempo ou porque ele está esperando por uma API de terceiros lenta para responder. Às vezes é um erro inesperado que causa uma exceção. O importante é que quando um processo trava, ele também pode travar o processo que o iniciou, o que por sua vez trava seu processo pai, desencadeando uma reação em cadeia que pode, em última instância, travar toda a sua aplicação.
Isto parece um desastre de programação esperando para acontecer, mas a boa notícia é que em Elixir, graças a Erlang, tem um conjunto de ferramentas poderosas para gerenciar processos, pegar acidentes e se recuperar rapidamente. Na próxima seção, vamos explicar por que esta reação em cadeia é uma coisa boa, como isolá-la e como utilizá-la a nosso favor. Você também começará a aprender como construir as aplicações tolerantes a falhas pelas quais a linguagem Elixir é famosa.
Ligação de processos
Os processos em Elixir podem ser ligados entre si, e os processos do Task são normalmente ligados automaticamente ao processo que os iniciou. Este vínculo é simplesmente chamado de ligação de processos (process link). As ligações de processos têm um papel importante na construção de aplicações concorrentes e tolerantes a falhas. Elas nos ajudam a desligar imediatamente partes do sistema, ou mesmo todo o sistema quando necessário, impedindo que a aplicação funcione em um péssimo estado. Vamos ver por que as ligações de processos são úteis e como funcionam, observando alguns exemplos. A figura a seguir mostra três processos ligados entre si:
Como você pode ver, o Processo C acabou de falhar. Imagine que o Processo B continua funcionando, sem saber do acidente. Ele espera receber um resultado do Processo C e se reportar ao Processo A. Entretanto, isto nunca acontecerá, portanto este sistema é mantido em um péssimo estado.
Isto pode ser evitado graças à ligação de processos. Quando dois processos estão ligados, eles formam uma relação especial - assim que um sai, o outro será notificado. Quando se tem uma cadeia de processos interligados, todos eles serão eventualmente notificados quando ocorrer um acidente. Por padrão, os processos interligados terminarão e limparão a memória que eles utilizavam, evitando outros problemas potenciais no decorrer da linha. Isto significa que o acidente do Processo C desencadeará uma reação em cadeia e o Processo B terminará, seguido pelo Processo A.
Entretanto, uma reação em cadeia como esta só faz sentido quando ocorre um erro grave. Dê uma olhada nesta figura, que mostra os processos em execução para uma aplicação web bancária:
A maioria das aplicações web dependem muito de um banco de dados, e o acesso ao banco de dados é geralmente gerenciado por um processo. Neste exemplo, o processo de Repo tem esta responsabilidade. Se o Repo falhar, porque o banco de dados se torna indisponível, esta aplicação não poderá funcionar de forma alguma. Se isso acontecer, você pode deixar todos os processos terminarem, o que deixará o site offline.
Agora, vamos considerar o processo Sender, que é responsável pelo envio de e-mails. Ao contrário do processo de Repo , não é essencial conceder aos usuários acesso à sua conta bancária. Se o processo do Sender falhar por qualquer razão, não queremos encerrar a aplicação inteira. Queremos conter esta falha e deixar o resto do sistema continuar.
Você pode isolar os acidentes configurando um processo para capturar as saídas. Capturar uma saída significa reconhecer a mensagem de saída de um processo vinculado, mas continuar a funcionar em vez de terminar. Isto também significa que a mensagem de saída não será propagada mais para outros processos. Voltando ao nosso exemplo original - se o Processo B estiver configurado para capturar as saídas, ele continuará funcionando após a queda do Processo C, mantendo o Processo A seguro também. Você pode configurar um processo para capturar saídas manualmente, mas normalmente você quer usar um tipo especial de processo chamado supervisor. Vamos falar sobre supervisors a seguir.
Como você pode ver, as ligações de processo são um mecanismo essencial para detectar saídas e lidar com falhas inesperadas de processo. A maioria das funções de utilidade no módulo Task cria links de processo automaticamente, ligando-se ao processo atual, mas há algumas que não o fazem. Você tem a opção de ligar ou não ao processo atual, mas você tem que escolher a função correta.
Por exemplo, quando usamos async/1 e async_stream/3, foi criado um link de processo para cada novo processo. Task.start/1, por outro lado, não cria um link de processo, mas existe o Task.start_link/1 que faz exatamente isso.
A maioria das funções do módulo Task que se ligam ao processo atual por padrão têm uma função alternativa, geralmente terminando com _nolink. Task.Supervisor.async_nolink/3 é a alternativa ao Task.async/1. Task.async_stream/3 pode ser substituído pelo Task.Supervisor.async_stream_nolink/4. Todas as funções no módulo Task.Supervisor são projetados para serem ligados a um supervisor.
A seguir, vamos aprender sobre supervisors e como podemos usar um quando iniciarmos processos de tarefas.
Conhecendo o Supervisor
Assim como no local de trabalho, onde os supervisores são responsáveis por grupos de funcionários, os supervisors (supervisores) de Elixir são responsáveis pelos processos a eles atribuídos. Os processos subordinados também são obrigados a se reportar a seu supervisor, que tem que garantir que tudo esteja funcionando sem problemas. Para isso, os supervisors vêm com um conjunto de características que lhes permitem gerenciar de forma eficaz outros processos. Eles podem iniciar e interromper processos e reiniciá-los em caso de erros imprevistos no sistema. Eles são configurados para capturar as saídas, assim, quando um processo supervisionado sai com um erro, esse erro será isolado e não se propagará mais. Tudo isso faz dos supervisors um importante bloco de construção para aplicações tolerantes a falhas.
Os processos supervisionados são chamados de child processes (processos filhos). Qualquer processo OTP pode ser supervisionado, e você também pode adicionar um supervisor como filho de outro supervisor. Tudo o que você tem que fazer é pedir ao supervisor que inicie o processo que você quer que seja gerenciado. Isto nos permite construir facilmente uma hierarquia de processos também chamada de supervision tree (árvore de supervisão).
Nossa aplicação sender já tem um supervisor no local, que você pode ver em application.ex:
Em sender/lib/sender/application.ex:
def start(_type, _args) do
children = [
# Starts a worker by calling: Sender.Worker.start_link(arg)
# {Sender.Worker, arg}
]
# See https://hexdocs.pm/elixir/Supervisor.html
# for other strategies and supported options
opts = [strategy: :one_for_one, name: Sender.Supervisor]
Supervisor.start_link(children, opts)
end
A função start/2 tem algum código padrão (boilerplate code) e algumas instruções já em vigor. No final da função temos Supervisor.start_link(children, opts) que inicia o supervisor principal da aplicação. Como a variável children é apenas uma lista vazia, na verdade não há processos filhos para supervisionar. Esta configuração é o resultado de usarmos o argumento --sup quando chamamos o comando mix new para criar nosso projeto.
Agora que você sabe como os supervisores são úteis, vamos ver como podemos usar um para nossos processos do módulo Task. Não queremos que nosso processo atual falhe se uma única tarefa falhar, então isolaremos erros potenciais iniciando os processos das tarefas sob um supervisor.
Adicionando um Supervisor
Elixir fornece um Supervisor behaviour para criar processos de supervisão, que você mesmo tem que implementar. Entretanto, há também alguns supervisores embutidos que podemos usar sem escrever (quase) qualquer código, e um deles é o Task.Supervisor. Ele é feito especificamente para trabalhar com processos de tarefas, por isso é uma excelente escolha para nós. Abra o application.ex e atualize a lista de filhos:
Em sender/lib/sender/application.change1.ex:
def start(_type, _args) do
children = [
{Task.Supervisor, name: Sender.EmailTaskSupervisor}
]
opts = [strategy: :one_for_one, name: Sender.Supervisor]
Supervisor.start_link(children, opts)
end
O elemento que acrescentamos é referido como um child specification (especificação de filho). Existem diferentes tipos de formatos para escrever __child specifications, mas na maioria das vezes é um tuple, contendo informações sobre o processo do filho. Esta informação é usada pelo supervisor para identificar, iniciar e ligar o processo.
Como o Task.Supervisor é um módulo incorporado, nosso child specification é simplesmente o nome do módulo, seguido por uma lista de opções. Usamos a opção :name para dar-lhe um nome único de nossa escolha, que usaremos mais tarde. Nomear um processo é conhecido como name registration (registro de nome). É comum anexar "Supervisor" ao nomear supervisores, por isso o chamamos Sender.EmailTaskSupervisor.
O child specification também poderia ser um map. Podemos reescrever a última mudança e usar um map como este:
def start(_type, _args) do
children = [
%{
id: Sender.EmailTaskSupervisor,
start: {
Task.Supervisor,
:start_link,
[[name: Sender.EmailTaskSupervisor]]
}
}
]
opts = [strategy: :one_for_one, name: Sender.Supervisor]
Supervisor.start_link(children, opts)
end
Usar um map é mais verboso, mas permite definir outras opções de configuração no map, além dos valores :id e :start. Há também uma função de ajuda Supervisor.child_spec/1² que retorna um map e permite que você sobreponha apenas as chaves que você precisa.
Vamos manter o formato tuple por enquanto, mas os outros formatos também são abordados no livro Concurrent Data Processing in Elixir.
Nosso supervisor está agora pronto para uso. É tão simples quanto isso, e não há necessidade de criar nenhum arquivo ou escrever mais código. Esta simples mudança nos permite usar toda uma série de funções encontradas no módulo Task.Supervisor, algumas das quais introduzimos anteriormente.
Usando Task.Supervisor
Antes de passarmos a usar o EmailTaskSupervisor, vamos ver o que acontece quando ocorre um erro e não há supervisor no local. Vamos simular um erro, levantando uma exceção ao enviar um de nossos e-mails falsos. Edite sender.ex e adicione a seguinte cláusula de função pouco antes de enviar_email/1:
Em sender/lib/sender.change4.ex:
def send_email("konnichiwa@world.com" = email), do:
raise "Oops, couldn't send email to #{email}!"
def send_email(email) do
Process.sleep(3000)
IO.puts("Email to #{email} sent")
{:ok, "email_sent"}
end
Usando pattern matching, vamos abrir uma exceção somente quando o endereço de e-mail for konnichiwa@world.com. Vamos ver o impacto deste erro na prática. Reinicie seu IEx shell e execute a função embutida self/0:
iex(1)> self()
#PID<0.136.0>
Você obtém um identificador de processo para o processo atual, que é o própria shell do IEx. Seu número pode ser diferente do meu, mas tudo bem. Tome nota disso.
Agora, vamos executar nossa função notify_all/1 e ver o que acontece:
iex(2)> Sender.notify_all(emails)
[error] Task #PID<0.154.0> started from #PID<0.136.0> terminating
** (RuntimeError) Oops, couldn't send email to konnichiwa@world.com!
(sender) lib/sender.ex:8: Sender.send_email/1
(elixir) lib/task/supervised.ex:90: Task.Supervised.invoke_mfa/2
(elixir) lib/task/supervised.ex:35: Task.Supervised.reply/5
(stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
Function: &:erlang.apply/2
Args: [#Function<0.104978293/1 in Sender.notify_all/1>,
["konnichiwa@world.com"]]
** (EXIT from #PID<0.136.0>) shell process exited with reason:
an exception was raised:
** (RuntimeError) Oops, couldn't send email to konnichiwa@world.com!
(sender) lib/sender.ex:8: Sender.send_email/1
(elixir) lib/task/supervised.ex:90: Task.Supervised.invoke_mfa/2
(elixir) lib/task/supervised.ex:35: Task.Supervised.reply/5
(stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
iex(1)>
Essa é uma grande mensagem de erro. Na verdade, há duas mensagens de erro que parecem ser quase idênticas. Lembra-se do que dissemos sobre as ligações de processos e como as falhas se propagam entre processos interligados? Neste caso, um processo de tarefa vinculado ao IEx levanta uma exceção. Devido a este link, criado pelo async_stream/3 , o IEx também trava com a mesma mensagem de exceção. Você pode verificar isto executando self() novamente:
iex(1)> self()
#PID<0.155.0>
Temos um novo identificador de processo, o que significa que um novo processo IEx foi iniciado e o antigo se quebrou.
Podemos evitar isto usando o novo EmailTaskSupervisor. Alterar notify_all/1 para usar Task.Supervisor.async_stream_nolink/4 em vez de Task.async_stream/3:
Em sender/lib/sender.change4.ex:
def notify_all(emails) do
Sender.EmailTaskSupervisor
|> Task.Supervisor.async_stream_nolink(emails, &send_email/1)
|> Enum.to_list()
end
O novo código é muito parecido com o antigo. O primeiro argumento de async_stream_nolink/4 é um módulo supervisor para os processos de tarefa - é onde você pode usar Sender.EmailTaskSupervisor. O resto dos argumentos são idênticos ao Task.async_stream/3. Vamos executar recompile() e notificar_tudo/1 novamente:
iex> Sender.notify_all(emails)
[error] Task #PID<0.173.0> started from #PID<0.155.0> terminating
** (RuntimeError) Oops, couldn't send email to konnichiwa@world.com!
(sender) lib/sender.ex:8: Sender.send_email/1
(elixir) lib/task/supervised.ex:90: Task.Supervised.invoke_mfa/2
(elixir) lib/task/supervised.ex:35: Task.Supervised.reply/5
(stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
Function: &:erlang.apply/2
Args: [#Function<0.32410907/1 in Sender.notify_all/1>,
["konnichiwa@world.com"]]
[info] Email to hello@world.com sent
[info] Email to hola@world.com sent
[info] Email to nihao@world.com sent
[
ok: {:ok, "email_sent"},
ok: {:ok, "email_sent"},
ok: {:ok, "email_sent"},
exit: {%RuntimeError{
message: "Oops, couldn't send email to konnichiwa@world.com!"
},
[
{Sender, :send_email, 1, [file: 'lib/sender.ex', line: 8]},
{Task.Supervised, :invoke_mfa, 2,
[file: 'lib/task/supervised.ex', line: 90]},
{Task.Supervised, :reply, 5, [file: 'lib/task/supervised.ex', line: 35]},
{:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 249]}
]}
]
A exceção familiar ainda está impressa, mas desta vez obtivemos um resultado significativo. Podemos ver que todas as tarefas foram concluídas com sucesso, exceto uma. A tarefa que se quebrou retornou um tuple :exit contendo a mensagem de erro e até mesmo um traço de pilha. O mais importante é que nosso processo atual foi isolado do acidente - você pode verificar isso executando o self() novamente.
É fácil ver que esta é uma enorme melhoria em relação a nossa versão anterior. O supervisor que introduzimos trata de processos e erros simultâneos, enquanto ainda estamos colhendo os benefícios de desempenho do async_stream.
Você já deve ter ouvido falar de uma filosofia associada a Erlang chamada Let It Crash (deixe falhar). As aplicações Erlang podem se recuperar rapidamente de erros reiniciando partes de seu sistema. Elixir também adota esta filosofia. Infelizmente, esta abordagem também é freqüentemente mal compreendida. Na próxima seção vamos explicar o que realmente significa Let It Crash e como os supervisores podem reiniciar processos para tornar nossos sistemas ainda mais resilientes.
Entendendo o Let it Crash
Usamos o Task.Supervisor para isolar uma falha de processo, mas pode parecer estranho que não tenhamos evitado a falha simplesmente adicionando o tratamento de erros na função send_email/1. Neste caso específico, fizemos isso de propósito, apenas para simular uma exceção inesperada. Na prática, você deve providenciar o tratamento de erros quando espera que um erro ocorra, e deixar o resto para o supervisor como último recurso.
Ao discutir o tratamento de erros para o Elixir, a frase let it crash é freqüentemente usada. Como resultado, algumas pessoas assumem que let it crash significa que Erlang e os desenvolvedores Elixir não fazem nenhuma manipulação de erros, o que não é o caso. A correspondência de padrões e as macro em Elixir facilitam o trabalho com {:ok, resultado} e {:erro, msg} tuples, e esta abordagem é amplamente utilizada na comunidade. Elixir também tem try e rescue para capturar exceções, semelhante à try e catch em outras linguagens.
Entretanto, por mais que tentemos como engenheiros, sabemos que erros podem acontecer. Isto muitas vezes leva a algo chamado Defensive programming (programação defensiva). Ela descreve a prática de tentar incessantemente cobrir todos os cenários possíveis de falha, mesmo quando alguns cenários são muito improváveis de acontecer, e não vale a pena lidar com eles.
Erlang e Elixir adotam uma abordagem diferente para a programação defensiva. Como todo código é executado em processos e processos que são leves, eles se concentram em como o sistema pode se recuperar de falhas versus como evitar todas as falhas. Você pode optar por permitir que uma parte (ou mesmo o todo) da aplicação se quebre e reinicie, mas você mesmo tratará dos outros erros. Esta mudança no pensamento e no projeto do software é a razão pela qual Erlang ficou famoso por sua confiabilidade e escalabilidade.
É aí que entram em jogo os supervisores. Vimos que os supervisores podem isolar as colisões, mas também podem reiniciar os processos filhos. Há três valores diferentes de reinicialização disponíveis para nós:
- :temporary nunca reiniciará os processos filhos.
- :transient reiniciará os processos filhos, mas somente quando eles terminam com um erro.
- :permanent reinicia sempre os filhos, mantendo-os em funcionamento, mesmo quando elas tentam desligar-se sem um erro.
ℹ️ Esclarecimento sobre a terminologia utilizada
A palavra reinício/restart é um pouco enganosa no contexto dos processos em Elixir. Uma vez que um processo termina, ele não pode ser trazido de volta à vida. Portanto, reiniciar um processo resulta em iniciar um novo processo para tomar o lugar do antigo, usando a mesma child especification.
Dependendo da freqüência do acidente, o próprio supervisor também pode terminar quando um processo filho não puder ser recuperado. Lembre-se de que a responsabilidade do supervisor é de zelar por seus processos. Se for usado um valor de restart :transient ou :permanent e um processo continuar a falhar, o supervisor irá sair, porque falhou em reiniciar esse processo. Isso é discutido com mais detalhes no livro Concurrent Data Processing in Elixir no Capítulo 2, Processos de Longa Duração Usando o GenServer, na página 25.
O Task.Supervisor incorporado que usamos até agora já está usando o valor de reinício :temporary para processos filhos. Este é um padrão sensato, pois evita que todos os processos de tarefas se quebrem se outra tarefa sair com um erro. No entanto, estas opções serão úteis no próximo capítulo do livro quando é construido árvores de supervisão mais complexas.
Conclusão
Cobrimos os processos do Elixir e usamos o módulo Task para fazer o trabalho concorrentemente e em paralelo, evitando facilmente erros de timeout e exceções inesperadas. Você também aprendeu sobre supervisores, que são a base da construção de aplicações Elixir tolerantes a falhas.
Embora o módulo Task seja muito poderoso e versátil, ele só é útil para executar funcões concorrentes ocasionais. Como resultado, os processos do módulo Task tem vida curta, vivem e terminam assim que são concluídos. No próximo capítulo do livro, é introduzido outro tipo de processo, que é capaz de funcionar e manter o estado durante o tempo que for necessário. Vai mudar completamente o jeito como você constrói aplicações para sempre, e isto não é um exagero.
O próximo capítulo do livro fala sobre GenServer e você pode continuar lendo nele, ou quem sabe em um próximo post por aqui mesmo...
até logo 🙂
Top comments (2)
como sempre, excelente artigos :D
valeu mestre :)
Some comments may only be visible to logged-in visitors. Sign in to view all comments.