Feladatok ebben a kurzusban:

A(z) Feladatok fogalomról

A feladatok olyan folyamatok, amelyek egyetlen konkrét művelet végrehajtására szolgálnak. Általában nem kommunikálnak más folyamatokkal, de visszaadhatnak egy eredményt annak a folyamatnak, amelyik elindította őket.

A feladatokat jellemzően a munka párhuzamosítására használjuk.

async/await

Feladat indításához használd a Task.async/1 függvényt. Ez egy névtelen függvényt vár argumentumként, és egy új folyamatban hajtja végre, amely össze van kapcsolva a hívó folyamattal. A visszatérési értéke egy %Task{} struktúra.

A végrehajtás eredményének lekéréséhez add át a %Task{} struktúrát a Task.await/2 függvénynek. Ez megvárja, amíg a feladat befejeződik, és visszaadja az eredményét. A második argumentum egy ezredmásodpercben megadott időtúllépés, amelynek az alapértéke 5000. Időtúllépésként átadhatod az :infinity értéket is, ekkor a feladat időtúllépés nélkül futhat.

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

Ne feledd, hogy a feladat elindítása és megvárása között az elindító folyamat nincs blokkolva, és közben más műveleteket is végezhet.

Minden Task.async/1-gyel indított feladatot meg kell várni, mert üzenetet küld a hívó folyamatnak. A Task.await/2 minden feladatnál csak egyszer hívható meg.

start/start_link

Ha egy feladatot csak a mellékhatások miatt akarsz elindítani, használd a Task.start/1 vagy a Task.start_link/1 függvényt. A Task.start/1 olyan feladatot indít, amely nincs összekapcsolva a hívó folyamattal, a Task.start_link/1 pedig olyat, amely össze van kapcsolva vele. Mindkét függvény egy {:ok, pid} tuple-t ad vissza.

Időtúllépések

Gyakori minta, hogy aszinkron feladatokat indítunk, majd az Enum.map/2 segítségével megvárjuk őket, így:

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

Fontos tudni, hogy a Task.await szinkron, és minden híváskor nulláról indítja az időtúllépés számlálóját. Ez azt jelenti, hogy ha így, egymás után használjuk, az időtúllépések összeadódnak.

Az első megvárt feladat legfeljebb az időtúllépés idejéig fut, a második viszont annyi ideig futhat, amennyi ideig az első futott, plusz az időtúllépés.

Vegyünk egy olyan helyzetet, amikor minden feladat egy kicsit tovább fut, mint az előző, de a futási idő különbsége kisebb, mint a várakozásra beállított időtúllépés.

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

Használhatjuk az Erlang :timer.tc/1 függvényét, hogy lemérjük, mennyi ideig futnak ezek a feladatok, mikromásodpercben.

: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]}

A 9004517 mikromásodperc 9 másodperc, jóval több, mint a megadott 1 másodperces (1000 ezredmásodperces) időtúllépés.

Ha ez az a viselkedés, amit szeretnél, akkor remek, vágj bele! Ha nem, akkor választhatsz a Task.async_stream/3 vagy a Task.await_many/2 közül.

Task.async_stream/3

A Task.async_stream/3 lehetővé teszi, hogy ugyanazt a függvényt sok különböző bemenetre futtasd, mindegyiket a saját feladatában. A feladatok össze vannak kapcsolva az aktuális folyamattal, akárcsak a Task.async/1 esetében. A visszatérési értéke egy stream.

Egy időtúllépést fogad el, amely minden feladatra külön vonatkozik. Nézzük ugyanazokat a bemeneteket és ugyanazt a függvényt a Task.async_stream/3-mal:

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

A Task.async_stream/3 remek választás, ha egyetlen függvényt szeretnél különböző bemenetekkel futtatni. Sok vezérlést ad a kezedbe más hasznos opciókkal, például hogy egyszerre hány feladat fusson, és mi történjen a feladat időtúllépésekor.

Ha olyan feladatokat kell futtatnod, amelyek különböző műveleteket végeznek, akkor inkább a Task.await_many/2 lehet a megoldás.

Task.await_many/2

A Task.await_many/2 az Elixir 1.11-ben jelent meg. Lehetővé teszi, hogy sok feladatot közös időtúllépéssel várj meg. Nézzük ugyanazokat a bemeneteket és ugyanazt a függvényt a Task.await_many/2-vel:

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