stefanchrobot
Supervising async tasks
I need to:
- do some asynchronous work concurrently
- kill the work when the parent
GenServerexits - be able to specify the timeout
- do some bookkeeping when the work succeeds
- do some other bookkeeping when the work fails
Is the Task.Supervisor the right tool for the job?
I’m thinking of writing a GenServer that would use Task.Supervisor.async_nolink to start Tasks, specifying the :shutdown option and handling the {ref, result} and :DOWN messages.
But how do I make sure that the tasks are immediately terminated when the GenServer is shut down? I guess Task.Supervisor.async is what I should use. But how will I get the results/crashes? Am I going to receive the same messages as with async_nolink?
Marked As Solved
josevalim
By parts:
- do some asynchronous work concurrently
Use a Task.Supervisor
- kill the work when the parent GenServer exits
Use Task.Supervisor.async but notice you can’t use await, you will receive the done or the down message in your handle_info
- be able to specify the timeout
When you call Task.Supervisor.async, also call Process.send_after(self, {:kill, task}, timeout). You will receive that message In handle_info, kill the task if timeout has passed
- do some bookkeeping when the work succeeds
Match the {ref, result} message in handle_info and cancel the timeout started above
- do some other bookkeeping when the work fails
Match the DOWN message in handle_info and cancel the timeout started above. Also make sure you call Process.flag(:trap_exit, true) so your genserver doesn’t die when the task crashes
Also Liked
joaquinalcerro
I was following this thread due to how common this use case is as expressed by @stefanchrobot and and the solution @josevalim gave seemed interesting and simple.
So I went and code it myself as I understood the use case and Task documentation (https://hexdocs.pm/elixir/1.5/Task.html#content).
It took me a while to grasp but this is the code. Please correct me if there is anything I misunderstood:
-
Crete the project
mix new stask --sup cd stask mix deps.get -
The supervision tree
defmodule Stask.Application do @moduledoc false use Application def start(_type, _args) do children = [ {Task.Supervisor, name: Stask.TaskSupervisor, restart: :temporary } ] opts = [strategy: :one_for_one, name: Stask.Supervisor] Supervisor.start_link(children, opts) end end -
The basic task
defmodule Stask do def some_task(text) do # Some hard work Process.sleep(15000) # Final result {:ok, "Hello #{text}"} end end -
The GenServer
defmodule Stask.Server do use GenServer ## Client API def start() do GenServer.start(__MODULE__, nil, name: __MODULE__) end def execute_task(pid, name, task_timeout) do GenServer.cast(pid, {:execute, name, task_timeout}) end ## Server Callbacks def init(_) do Process.flag(:trap_exit, true) {:ok, %{msg: "", timer: "", timeout: ""}} end def handle_cast({:execute, name, task_timeout}, state) do task = Task.Supervisor.async(Stask.TaskSupervisor, Stask, :some_task, [name]) timer_ref = Process.send_after(self(), {:kill, task}, task_timeout) {:noreply, %{state | timer: timer_ref}} end ## Handle info functions # This handle function executes when the task has timed out def handle_info({:kill, task}, state) do # Do some book keeping here IO.puts("Task has been canceled due to time out") # You can specify how much time you can wait a task to exit. If it # does not exit in this treshold time it will be killed # The default time is 5000ms. You can have Task.shutdown(task, 2000). # You can also kill it immediatly. Task.shutdown(task, :brutal_kill) # Check documentation: https://hexdocs.pm/elixir/1.5/Task.html#shutdown/2 case Task.shutdown(task) do {:ok, _reply} -> # Do some book keeping, the task responded while the task was # been shutdown IO.puts("The task responded while it was shutting down") {:exit, _reason} -> # Do seme book keeping, if task dies while it was waiting for # shutdown IO.puts("The task died before just it was shutdown") nil -> # Do some book kepping, the task was shutdown IO.puts("The task was shutdown") end {:noreply, state} end # This handle_info functions receives the message when the task # finished successfully in time def handle_info({_ref, {:ok, msg}}, state) do # Do some book keeping, the task finish successfully IO.puts("Task finished successfully") Process.cancel_timer(state.timer) {:noreply, %{state | msg: msg}} end # Once the task finised successfully, it exits normally. # This handle_info function responds to this message. def handle_info({:EXIT, _pid, :normal}, state) do # Do some book keeping IO.puts("The task exited and finished normally") {:noreply, state} end # Finally, when the task, the caller receives a DOWN message. # In this case the caller was the GenServer. def handle_info({:DOWN, _ref, :process, _pid, :normal}, state) do # Do some book keeping one the task goes down IO.puts(state.msg) {:noreply, state} end end
csangonzo
I might be a few years late
but here’s what I did today:
defmodule Utils.Worker do
@moduledoc """
- Needs:
- timeout
- function to run
"""
@callback run_job(params :: any()) :: any()
@callback job_ok(params :: any(), result :: any()) :: any()
@callback job_error(params :: any(), result :: any()) :: any()
defmacro __using__(opts) do
quote location: :keep, bind_quoted: [opts: opts] do
@behaviour Utils.Worker
@task_timeout Keyword.get(opts, :task_timeout) || 10_000
@task_supervisor Keyword.get(opts, :task_supervisor)
@test_jobs [
%{id: 1, config_id: 1},
%{id: 2, config_id: 1},
%{id: 3, config_id: 3},
%{id: 4, config_id: 1},
%{id: 5, config_id: 2},
%{id: 6, config_id: 4},
%{id: 7, config_id: 4},
%{id: 8, config_id: 5},
%{id: 9, config_id: 5},
%{id: 10, config_id: 5},
%{id: 11, config_id: 5},
%{id: 12, config_id: 5}
]
use GenServer
# ************************************************************
# API (Client)
# ************************************************************
def start_link(_) do
GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
end
@impl true
def init(_) do
{:ok, %{}}
end
def start_jobs(jobs \\ @test_jobs) do
GenServer.cast(__MODULE__, {:sync_data, jobs})
end
# ************************************************************
# Callbacks (Server)
# ************************************************************
@impl true
def handle_cast({:sync_data, test_jobs}, state) do
# This is kinda sketchy
task_map =
Enum.map(
test_jobs,
fn x ->
task =
Task.Supervisor.async_nolink(@task_supervisor, fn -> run_job(x) end)
timer_ref = Process.send_after(self(), {:timeout, task}, @task_timeout)
{task.ref, %{params: x, timer_ref: timer_ref}}
end
)
|> Map.new()
{:noreply, Map.merge(state, task_map)}
end
# If the task times out
@impl true
def handle_info({:timeout, task}, state) do
Process.exit(task.pid, :timeout)
{:noreply, state}
end
# If the task fails
@impl true
def handle_info({:DOWN, task_ref, _type, task_pid, reason}, state) do
task_state = Map.get(state, task_ref)
new_state = Map.delete(state, task_ref)
Process.cancel_timer(task_state.timer_ref)
# Process.demonitor(task_ref, [:flush])
job_error(task_state.params, reason)
{:noreply, new_state}
end
# If the task succeeds
@impl true
def handle_info({task_ref, result}, state) do
# The task succeed so we can cancel the monitoring and discard the DOWN message
task_state = Map.get(state, task_ref)
new_state = Map.delete(state, task_ref)
Process.cancel_timer(task_state.timer_ref)
Process.demonitor(task_ref, [:flush])
job_ok(task_state.params, result)
{:noreply, new_state}
end
end
end
end
stefanchrobot
Thanks for the response, that’s really helpful!
Things really clicked when I realized that tasks run under Task.Supervisor are always linked to that supervisor. Task.Supervisor.async and Task.Supervisor.async_nolink are used to decide whether to link (or not) to the caller of those functions, which should be a GenServer that should probably do something about the messages from the tasks.
Then I also realized that if I want to do any extra bookkeeping after the tasks complete, it has to be done in some GenServer, not in a supervisor.
Once I’m done with my stuff, I’d like to work on a PR to improve the docs for the Task.Supervisor.
Popular in Questions
Other popular topics
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #advent-of-code
- #elixirconf-us
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #performance









