Tareas en

Acerca de Tareas

Las tareas son procesos destinados a ejecutar una operación concreta. Normalmente no se comunican con otros procesos, pero pueden devolver un resultado al proceso que inició la tarea.

Las tareas se usan habitualmente para paralelizar el trabajo.

async/await

Para iniciar una tarea, usa Task.async/1. Recibe una función anónima como argumento y la ejecuta en un proceso nuevo que está vinculado al proceso que la llama. Devuelve una estructura %Task{}.

Para obtener el resultado de la ejecución, pasa la estructura %Task{} a Task.await/2. Esperará a que la tarea termine y devolverá su resultado. El segundo argumento es un tiempo de espera en milisegundos, que por defecto es 5000. El valor :infinity se puede pasar como tiempo de espera para permitir que la tarea se ejecute sin límite de tiempo.

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

Ten en cuenta que, entre que se inicia la tarea y se espera a que termine, el proceso que la inició no se bloquea y puede hacer otras operaciones.

Toda tarea iniciada con Task.async/1 se debe esperar, porque enviará un mensaje al proceso que la llamó. Task.await/2 se puede llamar solo una vez por cada tarea.

start/start_link

Si quieres iniciar una tarea solo por sus efectos secundarios, usa Task.start/1 o Task.start_link/1. Task.start/1 inicia una tarea que no está vinculada al proceso que la llama, y Task.start_link/1 inicia una tarea que sí está vinculada al proceso que la llama. Ambas funciones devuelven una tupla {:ok, pid}.

Tiempos de espera

Es un patrón habitual iniciar tareas asíncronas y luego esperarlas con Enum.map/2, así:

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

Es importante tener en cuenta que Task.await es síncrono y que, cada vez que se llama, su reloj del tiempo de espera empieza de nuevo en 0. Eso significa que, cuando se usa de esta forma, de manera secuencial, los tiempos de espera se suman.

La primera tarea que se espera no se ejecutará más tiempo que el tiempo de espera, pero la segunda tarea que se espera podrá ejecutarse durante todo el tiempo que se ejecutó la primera, más el tiempo de espera.

Imagina una situación en la que cada tarea tarda un poco más en ejecutarse que la anterior, pero la diferencia de duración entre tareas es menor que el tiempo de espera.

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

Podemos usar :timer.tc/1, una función de Erlang, para medir cuánto tardarán en ejecutarse esas tareas, en microsegundos.

: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 microsegundos son 9 segundos, mucho más que el tiempo de espera de 1 segundo (1000 milisegundos).

Si este es el comportamiento que quieres, ¡perfecto, adelante! Si no, la alternativa es usar Task.async_stream/3 o Task.await_many/2.

Task.async_stream/3

Task.async_stream/3 te permite ejecutar la misma función para muchas entradas distintas, cada una en su propia tarea. Las tareas están vinculadas al proceso actual, igual que cuando se ejecutan con Task.async/1. Devuelve un stream.

Acepta un tiempo de espera que se aplica a cada tarea por separado. Imagina las mismas entradas y la misma función usadas con 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 es ideal si tienes una sola función que ejecutar con entradas distintas. Ofrece mucho control, ya que acepta otras opciones útiles, como cuántas tareas ejecutar al mismo tiempo y qué hacer cuando una tarea agota su tiempo de espera.

Si necesitas ejecutar tareas que realizan operaciones distintas, quizá necesites Task.await_many/2 en su lugar.

Task.await_many/2

Task.await_many/2 se introdujo en Elixir 1.11. Te permite esperar muchas tareas con un tiempo de espera compartido. Imagina las mismas entradas y la misma función usadas con 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