Tasks em

Sobre Tasks

Tasks são processos destinados a executar uma operação específica. Normalmente elas não se comunicam com outros processos, mas podem retornar um resultado ao processo que iniciou a task.

É comum usar tasks para paralelizar o trabalho.

async/await

Para iniciar uma task, use Task.async/1. Ela recebe uma função anônima como argumento e a executa em um novo processo vinculado ao processo que fez a chamada. Ela retorna uma struct %Task{}.

Para obter o resultado da execução, passe a struct %Task{} para Task.await/2. Ela vai esperar a task terminar e retornar o resultado dela. O segundo argumento é um timeout em milissegundos, com valor padrão de 5000. O valor :infinity pode ser passado como timeout para permitir que a task execute sem timeout.

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

Repare que, entre iniciar a task e aguardá-la, o processo que iniciou a task não fica bloqueado e pode fazer outras operações.

Toda task iniciada com Task.async/1 deve ser aguardada, porque ela vai enviar uma mensagem ao processo que fez a chamada. Task.await/2 pode ser chamada para cada task apenas uma vez.

start/start_link

Se você quer iniciar uma task apenas por efeitos colaterais, use Task.start/1 ou Task.start_link/1. Task.start/1 inicia uma task não vinculada ao processo que fez a chamada, e Task.start_link/1 inicia uma task vinculada ao processo que fez a chamada. As duas funções retornam uma tupla {:ok, pid}.

Timeouts

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

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

É importante ter em mente que Task.await é síncrona, e cada vez que ela é chamada, o relógio do timeout começa do zero. Isso significa que, quando ela é usada dessa forma, sequencialmente, os timeouts se somam.

A primeira task aguardada vai rodar no máximo até o timeout, mas a segunda task aguardada poderá rodar por todo o tempo que a primeira rodou, mais o timeout.

Considere uma situação em que cada task demora um pouco mais para rodar que a anterior, mas a diferença de duração entre as tasks é menor que o timeout do await.

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

Podemos usar :timer.tc/1, uma função do Erlang, para medir quanto tempo essas tasks vão rodar, 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 que o timeout dado de 1 segundo (1000 milissegundos).

Se é esse o comportamento que você quer, ótimo, vá em frente! Se não, a alternativa é usar Task.async_stream/3 ou Task.await_many/2.

Task.async_stream/3

Task.async_stream/3 permite rodar a mesma função para várias entradas diferentes, cada uma na sua própria task. As tasks são vinculadas ao processo atual, assim como quando rodadas com Task.async/1. Ela retorna um stream.

Ela aceita um timeout que se aplica a cada task separadamente. Considere as mesmas entradas e a mesma função usadas 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

Task.async_stream/3 é ótima quando você tem uma única função para rodar com entradas diferentes. Ela oferece bastante controle ao aceitar outras opções úteis, como quantas tasks rodar ao mesmo tempo e o que fazer quando uma task dá timeout.

Se você precisa rodar tasks que realizam operações diferentes, talvez precise de Task.await_many/2.

Task.await_many/2

Task.await_many/2 foi introduzida no Elixir 1.11. Ela permite aguardar várias tasks com um timeout compartilhado. Considere as mesmas entradas e a mesma função usadas 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