Εργασίες σε

Σχετικά με την έννοια Εργασίες

Τα tasks είναι διεργασίες που προορίζονται να εκτελέσουν μία συγκεκριμένη λειτουργία. Συνήθως δεν επικοινωνούν με άλλες διεργασίες, αλλά μπορούν να επιστρέψουν ένα αποτέλεσμα στη διεργασία που ξεκίνησε το task.

Τα tasks χρησιμοποιούνται συχνά για την παραλληλοποίηση της εργασίας.

async/await

Για να ξεκινήσεις ένα task, χρησιμοποίησε το Task.async/1. Δέχεται μια ανώνυμη συνάρτηση ως όρισμα και την εκτελεί σε μια νέα διεργασία που είναι συνδεδεμένη με τη διεργασία που την κάλεσε. Επιστρέφει ένα %Task{} struct.

Για να πάρεις το αποτέλεσμα της εκτέλεσης, πέρασε το struct %Task{} στο Task.await/2. Θα περιμένει να τελειώσει το task και θα επιστρέψει το αποτέλεσμά του. Το δεύτερο όρισμα είναι ένα χρονικό όριο σε χιλιοστά του δευτερολέπτου, με προεπιλεγμένη τιμή το 5000. Μπορείς να περάσεις την τιμή :infinity ως χρονικό όριο, ώστε το task να εκτελεστεί χωρίς χρονικό όριο.

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

Σημείωσε ότι, από τη στιγμή που ξεκινάς το task μέχρι να το περιμένεις, η διεργασία που ξεκίνησε το task δε μπλοκάρεται και μπορεί να κάνει άλλες λειτουργίες.

Κάθε task που ξεκινά με Task.async/1 πρέπει να το περιμένεις, γιατί θα στείλει ένα μήνυμα στη διεργασία που το κάλεσε. Το Task.await/2 μπορεί να κληθεί για κάθε task μόνο μία φορά.

start/start_link

Αν θέλεις να ξεκινήσεις ένα task μόνο για τις παρενέργειές του, χρησιμοποίησε το Task.start/1 ή το Task.start_link/1. Το Task.start/1 ξεκινά ένα task που δεν είναι συνδεδεμένο με τη διεργασία που το κάλεσε, ενώ το Task.start_link/1 ξεκινά ένα task που είναι συνδεδεμένο με τη διεργασία που το κάλεσε. Και οι δύο συναρτήσεις επιστρέφουν μια πλειάδα {:ok, pid}.

Χρονικά όρια

Είναι συνηθισμένο μοτίβο να ξεκινάς ασύγχρονα tasks και μετά να τα περιμένεις χρησιμοποιώντας το Enum.map/2, όπως εδώ:

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

Είναι σημαντικό να έχεις υπόψη σου ότι το Task.await είναι σύγχρονο και ότι κάθε φορά που καλείται, το ρολόι του χρονικού ορίου ξεκινά από το 0. Αυτό σημαίνει ότι, όταν χρησιμοποιείται με αυτόν τον τρόπο, διαδοχικά, τα χρονικά όρια αθροίζονται.

Το πρώτο task που περιμένουμε δε θα εκτελεστεί περισσότερο από το χρονικό όριο, αλλά το δεύτερο task που περιμένουμε θα εκτελεστεί για όσο χρόνο εκτελέστηκε το πρώτο task, συν το χρονικό όριο.

Σκέψου μια κατάσταση όπου κάθε task αργεί λίγο περισσότερο να εκτελεστεί από το προηγούμενο, αλλά η διαφορά στη διάρκεια εκτέλεσης των tasks είναι μικρότερη από το χρονικό όριο της αναμονής.

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

Μπορούμε να χρησιμοποιήσουμε το :timer.tc/1, μια συνάρτηση της Erlang, για να μετρήσουμε πόσο χρόνο θα εκτελεστούν αυτά τα tasks, σε μικροδευτερόλεπτα.

: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 μικροδευτερόλεπτα είναι 9 δευτερόλεπτα, πολύ περισσότερα από το χρονικό όριο του 1 δευτερολέπτου (1000 χιλιοστά του δευτερολέπτου) που δόθηκε.

Αν αυτή είναι η συμπεριφορά που θέλεις, υπέροχα, προχώρα! Αν όχι, η εναλλακτική είναι να χρησιμοποιήσεις είτε το Task.async_stream/3 είτε το Task.await_many/2.

Task.async_stream/3

Το Task.async_stream/3 σου επιτρέπει να εκτελέσεις την ίδια συνάρτηση για πολλά διαφορετικά δεδομένα εισόδου, το καθένα στο δικό του task. Τα tasks είναι συνδεδεμένα με την τρέχουσα διεργασία, ακριβώς όπως όταν εκτελούνται με το Task.async/1. Επιστρέφει μια ροή.

Δέχεται ένα χρονικό όριο που ισχύει για κάθε task ξεχωριστά. Σκέψου τα ίδια δεδομένα εισόδου και την ίδια συνάρτηση με το 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 είναι ιδανικό αν έχεις μία μόνο συνάρτηση να εκτελέσεις με διαφορετικά δεδομένα εισόδου. Παρέχει πολύ έλεγχο, καθώς δέχεται και άλλες χρήσιμες επιλογές, όπως πόσα tasks να εκτελεστούν ταυτόχρονα και τι να γίνει σε περίπτωση χρονικού ορίου σε κάποιο task.

Αν χρειάζεται να εκτελέσεις tasks που κάνουν διαφορετικές λειτουργίες, ίσως χρειαστείς το Task.await_many/2.

Task.await_many/2

Το Task.await_many/2 εισήχθη στην Elixir 1.11. Σου επιτρέπει να περιμένεις πολλά tasks με ένα κοινό χρονικό όριο. Σκέψου τα ίδια δεδομένα εισόδου και την ίδια συνάρτηση με το 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