Tasks in

Über Tasks

Tasks sind Prozesse, die eine bestimmte Operation ausführen sollen. Normalerweise kommunizieren sie nicht mit anderen Prozessen, aber sie können ein Ergebnis an den Prozess zurückgeben, der den Task gestartet hat.

Tasks werden häufig verwendet, um Arbeit zu parallelisieren.

async/await

Um einen Task zu starten, verwendest du Task.async/1. Sie nimmt eine anonyme Funktion als Argument entgegen und führt sie in einem neuen Prozess aus, der mit dem aufrufenden Prozess verlinkt ist. Sie gibt ein %Task{}-Struct zurück.

Um das Ergebnis der Ausführung zu erhalten, übergibst du das %Task{}-Struct an Task.await/2. Es wartet, bis der Task fertig ist, und gibt sein Ergebnis zurück. Das zweite Argument ist ein Timeout in Millisekunden, standardmäßig 5000. Der Wert :infinity kann als Timeout übergeben werden, damit der Task ohne Timeout ausgeführt wird.

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

Beachte, dass der Prozess, der den Task gestartet hat, zwischen dem Starten und dem Warten auf den Task nicht blockiert ist und andere Operationen ausführen kann.

Jeder Task, der mit Task.async/1 gestartet wird, sollte abgewartet werden, weil er eine Nachricht an den aufrufenden Prozess sendet. Task.await/2 kann für jeden Task nur einmal aufgerufen werden.

start/start_link

Wenn du einen Task nur wegen seiner Seiteneffekte starten möchtest, verwende Task.start/1 oder Task.start_link/1. Task.start/1 startet einen Task, der nicht mit dem aufrufenden Prozess verlinkt ist, und Task.start_link/1 startet einen Task, der mit dem aufrufenden Prozess verlinkt ist. Beide Funktionen geben ein {:ok, pid}-Tupel zurück.

Timeouts

Es ist ein gängiges Muster, asynchrone Tasks zu starten und sie dann mit Enum.map/2 abzuwarten, und zwar so:

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

Es ist wichtig zu wissen, dass Task.await synchron ist und bei jedem Aufruf seine Timeout-Uhr bei 0 startet. Das bedeutet, dass sich die Timeouts addieren, wenn es auf diese Weise sequenziell verwendet wird.

Der erste Task, der abgewartet wird, läuft nicht länger als das Timeout, aber der zweite Task, der abgewartet wird, darf so lange laufen, wie der erste Task lief, plus das Timeout.

Stell dir eine Situation vor, in der jeder Task etwas länger braucht als der vorherige, aber der Unterschied in der Laufzeit der Tasks kleiner ist als das Timeout beim Abwarten.

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

Wir können :timer.tc/1, eine Erlang-Funktion, verwenden, um zu messen, wie lange diese Tasks laufen, und zwar in Mikrosekunden.

: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 Mikrosekunden sind 9 Sekunden, viel mehr als das angegebene Timeout von 1 Sekunde (1000 Millisekunden).

Wenn das das Verhalten ist, das du willst, prima, dann leg los! Wenn nicht, ist die Alternative, entweder Task.async_stream/3 oder Task.await_many/2 zu verwenden.

Task.async_stream/3

Mit Task.async_stream/3 kannst du dieselbe Funktion für viele verschiedene Eingabewerte ausführen, jeden in seinem eigenen Task. Die Tasks sind mit dem aktuellen Prozess verlinkt, genau wie bei der Ausführung mit Task.async/1. Es gibt einen Stream zurück.

Es akzeptiert ein Timeout, das für jeden Task separat gilt. Betrachte dieselben Eingabewerte und dieselbe Funktion, verwendet mit 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 ist ideal, wenn du eine einzige Funktion mit verschiedenen Eingabewerten ausführen möchtest. Es bietet viel Kontrolle, weil es weitere nützliche Optionen akzeptiert, etwa wie viele Tasks gleichzeitig laufen sollen und was bei einem Task-Timeout passieren soll.

Wenn du Tasks ausführen musst, die unterschiedliche Operationen ausführen, brauchst du stattdessen vielleicht Task.await_many/2.

Task.await_many/2

Task.await_many/2 wurde in Elixir 1.11 eingeführt. Es ermöglicht dir, viele Tasks mit einem gemeinsamen Timeout abzuwarten. Betrachte dieselben Eingabewerte und dieselbe Funktion, verwendet mit 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