Tarefas em

Sobre Tarefas

As tarefas são processos destinados a executar uma operação específica. Normalmente não comunicam com outros processos, mas podem devolver um resultado ao processo que iniciou a tarefa.

As tarefas são habitualmente usadas para paralelizar trabalho.

async/await

Para iniciar uma tarefa, usa Task.async/1. Recebe uma função anónima como argumento e executa-a num novo processo ligado ao processo que a chamou. Devolve uma struct %Task{}.

Para obter o resultado da execução, passa a struct %Task{} a Task.await/2. Este espera que a tarefa termine e devolve o seu resultado. O segundo argumento é um tempo limite em milissegundos, com o valor predefinido de 5000. Podes passar o valor :infinity como tempo limite para permitir que a tarefa seja executada sem limite de tempo.

task = Task.async(fn -> Enum.sum(0..999_999) end)
# => %Task{
#      owner: #PID<0.110.0>,
#      pid: #PID<0.119.0>,
#      ref: #Reference<0.3715954046.815529991.83334>
#    }

Task.await(task)
# => 499999500000

Repara que, entre iniciar a tarefa e esperar por ela, o processo que a iniciou não fica bloqueado e pode realizar outras operações.

Qualquer tarefa iniciada com Task.async/1 deve ser aguardada, porque vai enviar uma mensagem ao processo que a iniciou. O Task.await/2 só pode ser chamado uma vez por cada tarefa.

start/start_link

Se quiseres iniciar uma tarefa apenas para produzir efeitos secundários, usa Task.start/1 ou Task.start_link/1. O Task.start/1 inicia uma tarefa que não está ligada ao processo que a chamou, e o Task.start_link/1 inicia uma tarefa ligada ao processo que a chamou. Ambas as funções devolvem um tuplo {:ok, pid}.

Tempos limite

É um padrão comum iniciar tarefas assíncronas e depois aguardá-las com Enum.map/2, assim:

inputs
|> Enum.map(&Task.async(fn -> function.(&1) end))
|> Enum.map(&Task.await(&1, 1_000))

É importante ter em conta que Task.await é síncrono e que, sempre que é chamado, reinicia a contagem do seu tempo limite a partir de 0. Isto significa que, quando é usado desta forma, em sequência, os tempos limite vão-se somando.

A primeira tarefa a ser aguardada não será executada durante mais tempo do que o tempo limite, mas a segunda tarefa a ser aguardada poderá ser executada durante o tempo que a primeira tarefa demorou, mais o tempo limite.

Considera uma situação em que cada tarefa demora um pouco mais a executar do que a anterior, mas a diferença de duração entre as tarefas é menor do que o tempo limite de espera.

inputs = 1..10
function = fn n -> :timer.sleep(n * 900) && n end

Podemos usar :timer.tc/1, uma função de Erlang, para medir quanto tempo essas tarefas demoram a executar, em microssegundos.

:timer.tc(fn ->
  inputs
  |> Enum.map(&Task.async(fn -> function.(&1) end))
  |> Enum.map(&Task.await(&1, 1_000))
end)

# => {9004517, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]}

9004517 microssegundos são 9 segundos, muito mais do que o tempo limite indicado de 1 segundo (1000 milissegundos).

Se este é o comportamento que queres, ótimo, força! Se não, a alternativa é usar Task.async_stream/3 ou Task.await_many/2.

Task.async_stream/3

O Task.async_stream/3 permite-te executar a mesma função para vários valores de entrada diferentes, cada um na sua própria tarefa. As tarefas ficam ligadas ao processo atual, tal como quando são executadas com Task.async/1. Devolve um stream.

Aceita um tempo limite que se aplica a cada tarefa separadamente. Considera os mesmos valores de entrada e a mesma função usados com Task.async_stream/3:

inputs
|> Task.async_stream(function, timeout: 1_000)
|> Enum.to_list()

# => ** (exit) exited in: Task.Supervised.stream(1000)
#       ** (EXIT) time out
#       (elixir 1.11.3) lib/task/supervised.ex:304: Task.Supervised.stream_reduce/7
#       (elixir 1.11.3) lib/enum.ex:3473: Enum.reverse/1
#       (elixir 1.11.3) lib/enum.ex:3066: Enum.to_list/1

O Task.async_stream/3 é ótimo quando tens uma única função para executar com valores de entrada diferentes. Oferece muito controlo, pois aceita outras opções úteis, como o número de tarefas a executar em simultâneo e o que fazer quando uma tarefa excede o tempo limite.

Se precisares de executar tarefas que realizam operações diferentes, talvez precises antes do Task.await_many/2.

Task.await_many/2

O Task.await_many/2 foi introduzido no Elixir 1.11. Permite-te aguardar muitas tarefas com um tempo limite partilhado. Considera os mesmos valores de entrada e a mesma função usados com Task.await_many/2:

inputs
|> Enum.map(&Task.async(fn -> function.(&1) end))
|> Task.await_many(1_000)

# => ** (exit) exited in: Task.await_many([...], 1000)
#        ** (EXIT) time out
#        (elixir 1.11.3) lib/task.ex:725: Task.await_many/5
#        (elixir 1.11.3) lib/task.ex:709: Task.await_many/2