Tâches en

À propos de Tâches

Les tâches sont des processus destinés à exécuter une opération précise. Ils ne communiquent généralement pas avec les autres processus, mais ils peuvent renvoyer un résultat au processus qui a lancé la tâche.

Les tâches sont couramment utilisées pour paralléliser le travail.

async/await

Pour lancer une tâche, utilise Task.async/1. Elle prend une fonction anonyme en argument et l'exécute dans un nouveau processus lié au processus appelant. Elle renvoie une structure %Task{}.

Pour récupérer le résultat de l'exécution, passe la structure %Task{} à Task.await/2. Elle va attendre que la tâche se termine et renvoyer son résultat. Le deuxième argument est un délai d'attente en millisecondes, dont la valeur par défaut est 5000. La valeur :infinity peut être passée comme délai d'attente pour permettre à la tâche de s'exécuter sans limite de temps.

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

Note qu'entre le lancement de la tâche et l'attente de son résultat, le processus qui a lancé la tâche n'est pas bloqué et peut effectuer d'autres opérations.

Toute tâche lancée avec Task.async/1 doit être attendue, car elle envoie un message au processus appelant. Task.await/2 ne peut être appelée qu'une seule fois par tâche.

start/start_link

Si tu veux lancer une tâche uniquement pour ses effets de bord, utilise Task.start/1 ou Task.start_link/1. Task.start/1 lance une tâche qui n'est pas liée au processus appelant, et Task.start_link/1 lance une tâche qui est liée au processus appelant. Les deux fonctions renvoient un tuple {:ok, pid}.

Délais d'attente

C'est un schéma courant de lancer des tâches asynchrones puis de les attendre avec Enum.map/2, comme ceci :

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

Il est important de savoir que Task.await est synchrone et qu'à chaque appel, son compteur de délai d'attente repart de 0. Cela signifie que lorsqu'elle est utilisée de cette façon, séquentiellement, les délais d'attente s'additionnent.

La première tâche attendue ne s'exécutera pas plus longtemps que le délai d'attente, mais la deuxième tâche attendue pourra s'exécuter aussi longtemps que la première, plus le délai d'attente.

Imagine une situation où chaque tâche prend un peu plus de temps à s'exécuter que la précédente, mais où la différence de durée d'exécution entre les tâches est inférieure au délai d'attente.

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

On peut utiliser :timer.tc/1, une fonction Erlang, pour mesurer combien de temps ces tâches vont s'exécuter, en microsecondes.

: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 microsecondes représentent 9 secondes, bien plus que le délai d'attente donné de 1 seconde (1000 millisecondes).

Si c'est le comportement que tu veux, tant mieux, fonce ! Sinon, l'alternative consiste à utiliser soit Task.async_stream/3, soit Task.await_many/2.

Task.async_stream/3

Task.async_stream/3 permet d'exécuter la même fonction pour de nombreuses entrées différentes, chacune dans sa propre tâche. Les tâches sont liées au processus courant, exactement comme lorsqu'elles sont lancées avec Task.async/1. Elle renvoie un flux.

Elle accepte un délai d'attente qui s'applique séparément à chaque tâche. Prenons les mêmes entrées et la même fonction utilisées avec 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 est idéale si tu as une seule fonction à exécuter avec différentes entrées. Elle offre beaucoup de contrôle grâce à d'autres options utiles, comme le nombre de tâches à exécuter en même temps et le comportement à adopter en cas de dépassement du délai d'attente d'une tâche.

Si tu as besoin de lancer des tâches qui effectuent des opérations différentes, c'est plutôt Task.await_many/2 qu'il te faut.

Task.await_many/2

Task.await_many/2 a été introduite dans Elixir 1.11. Elle permet d'attendre de nombreuses tâches avec un délai d'attente partagé. Prenons les mêmes entrées et la même fonction utilisées avec 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