Elixir

v1.5.1

Elixir

Elixir v1.5.1 Task.Supervisor View Source

A task supervisor.

This module defines a supervisor which can be used to dynamically supervise tasks.

start_link/1 can be used to start the supervisor. See the Task module for more examples.

Name registration

A Task.Supervisor is bound to the same name registration rules as a GenServer. Read more about them in the GenServer docs.

Link to this section Summary

Types

option()

Option values used by start_link

Functions

async(supervisor, fun)

Starts a task that can be awaited on

async(supervisor, module, fun, args)

Starts a task that can be awaited on

async_nolink(supervisor, fun)

Starts a task that can be awaited on

async_nolink(supervisor, module, fun, args)

Starts a task that can be awaited on

async_stream(supervisor, enumerable, fun, options \\ [])

Returns a stream that runs the given function fun concurrently on each item in enumerable

async_stream(supervisor, enumerable, module, function, args, options \\ [])

Returns a stream that runs the given module, function, and args concurrently on each item in enumerable

async_stream_nolink(supervisor, enumerable, fun, options \\ [])

Returns a stream that runs the given function concurrently on each item in enumerable

async_stream_nolink(supervisor, enumerable, module, function, args, options \\ [])

Returns a stream that runs the given module, function, and args concurrently on each item in enumerable

children(supervisor)

Returns all children PIDs

start_child(supervisor, fun)

Starts a task as a child of the given supervisor

start_child(supervisor, module, fun, args)

Starts a task as a child of the given supervisor

start_link(opts \\ [])

Starts a new supervisor

terminate_child(supervisor, pid)

Terminates the child with the given pid

Link to this section Types

Link to this type option() View Source 
option() ::
  Supervisor.option() |
  {:restart, :supervisor.restart()} |
  {:shutdown, :supervisor.shutdown()}

Option values used by start_link

Link to this section Functions

Link to this function async(supervisor, fun) View Source 
async(Supervisor.supervisor(), (() -> any())) :: Task.t()

Starts a task that can be awaited on.

The supervisor must be a reference as defined in Task.Supervisor. The task will still be linked to the caller, see Task.async/3 for more information and async_nolink/2 for a non-linked variant.

Note this function requires the task supervisor to have :temporary as the :restart option (the default), as async/2 keeps a direct reference to the task which is lost if the task is restarted.

Link to this function async(supervisor, module, fun, args) View Source 
async(Supervisor.supervisor(), module(), atom(), [term()]) :: Task.t()

Starts a task that can be awaited on.

The supervisor must be a reference as defined in Task.Supervisor. The task will still be linked to the caller, see Task.async/3 for more information and async_nolink/2 for a non-linked variant.

Note this function requires the task supervisor to have :temporary as the :restart option (the default), as async/4 keeps a direct reference to the task which is lost if the task is restarted.

Link to this function async_nolink(supervisor, fun) View Source 
async_nolink(Supervisor.supervisor(), (() -> any())) :: Task.t()

Starts a task that can be awaited on.

The supervisor must be a reference as defined in Task.Supervisor. The task won’t be linked to the caller, see Task.async/3 for more information.

Note this function requires the task supervisor to have :temporary as the :restart option (the default), as async_nolink/2 keeps a direct reference to the task which is lost if the task is restarted.

Compatibility with OTP behaviours

If you create a task using async_nolink inside an OTP behaviour like GenServer, you should match on the message coming from the task inside your GenServer.handle_info/2 callback.

The reply sent by the task will be in the format {ref, result}, where ref is the monitor reference held by the task struct and result is the return value of the task function.

Keep in mind that, regardless of how the task created with async_nolink terminates, the caller’s process will always receive a :DOWN message with the same ref value that is held by the task struct. If the task terminates normally, the reason in the :DOWN message will be :normal.

Link to this function async_nolink(supervisor, module, fun, args) View Source 
async_nolink(Supervisor.supervisor(), module(), atom(), [term()]) :: Task.t()

Starts a task that can be awaited on.

The supervisor must be a reference as defined in Task.Supervisor. The task won’t be linked to the caller, see Task.async/3 for more information.

Note this function requires the task supervisor to have :temporary as the :restart option (the default), as async_nolink/4 keeps a direct reference to the task which is lost if the task is restarted.

Link to this function async_stream(supervisor, enumerable, fun, options \\ []) View Source 
async_stream(Supervisor.supervisor(), Enumerable.t(), (term() -> term()), keyword()) :: Enumerable.t()

Returns a stream that runs the given function fun concurrently on each item in enumerable.

Each item in enumerable is passed as argument to the given function fun and processed by its own task. The tasks will be spawned under the given supervisor and linked to the current process, similarly to async/2.

See async_stream/6 for discussion, options, and examples.

Link to this function async_stream(supervisor, enumerable, module, function, args, options \\ []) View Source 
async_stream(Supervisor.supervisor(), Enumerable.t(), module(), atom(), [term()], keyword()) :: Enumerable.t()

Returns a stream that runs the given module, function, and args concurrently on each item in enumerable.

Each item will be prepended to the given args and processed by its own task. The tasks will be spawned under the given supervisor and linked to the current process, similarly to async/4.

When streamed, each task will emit {:ok, value} upon successful completion or {:exit, reason} if the caller is trapping exits. Results are emitted in the same order as the original enumerable.

The level of concurrency can be controlled via the :max_concurrency option and defaults to System.schedulers_online/0. A timeout can also be given as an option representing the maximum amount of time to wait without a task reply.

Note this function requires the task supervisor to have :temporary as the :restart option (the default), as async_stream/6 keeps a direct reference to the task which is lost if the task is restarted.

Finally, if you find yourself trapping exits to handle exits inside the async stream, consider using async_stream_nolink/6 to start tasks that are not linked to the current process.

Options

  • :max_concurrency - sets the maximum number of tasks to run at the same time. Defaults to System.schedulers_online/0.
  • :ordered - whether the results should be returned in the same order as the input stream. This option is useful when you have large streams and don’t want to buffer results before they are delivered. Defaults to true.
  • :timeout - the maximum amount of time to wait (in milliseconds) without receiving a task reply (across all running tasks). Defaults to 5000.

  • :on_timeout - what do to when a task times out. The possible values are:

    • :exit (default) - the process that spawned the tasks exits.
    • :kill_task - the task that timed out is killed. The value emitted for that task is {:exit, :timeout}.

Examples

Let’s build a stream and then enumerate it:

stream = Task.Supervisor.async_stream(MySupervisor, collection, Mod, :expensive_fun, [])
Enum.to_list(stream)
Link to this function async_stream_nolink(supervisor, enumerable, fun, options \\ []) View Source 
async_stream_nolink(Supervisor.supervisor(), Enumerable.t(), (term() -> term()), keyword()) :: Enumerable.t()

Returns a stream that runs the given function concurrently on each item in enumerable.

Each item in enumerable is passed as argument to the given function fun and processed by its own task. The tasks will be spawned under the given supervisor and linked to the current process, similarly to async_nolink/2.

See async_stream/6 for discussion and examples.

Link to this function async_stream_nolink(supervisor, enumerable, module, function, args, options \\ []) View Source 
async_stream_nolink(Supervisor.supervisor(), Enumerable.t(), module(), atom(), [term()], keyword()) :: Enumerable.t()

Returns a stream that runs the given module, function, and args concurrently on each item in enumerable.

Each item in enumerable will be prepended to the given args and processed by its own task. The tasks will be spawned under the given supervisor and will not be linked to the current process, similarly to async_nolink/4.

See async_stream/6 for discussion, options, and examples.

Link to this function children(supervisor) View Source 
children(Supervisor.supervisor()) :: [pid()]

Returns all children PIDs.

Link to this function start_child(supervisor, fun) View Source 
start_child(Supervisor.supervisor(), (() -> any())) :: {:ok, pid()}

Starts a task as a child of the given supervisor.

Note that the spawned process is not linked to the caller, but only to the supervisor. This command is useful in case the task needs to perform side-effects (like I/O) and does not need to report back to the caller.

Link to this function start_child(supervisor, module, fun, args) View Source 
start_child(Supervisor.supervisor(), module(), atom(), [term()]) :: {:ok, pid()}

Starts a task as a child of the given supervisor.

Similar to start_child/2 except the task is specified by the given module, fun and args.

Link to this function start_link(opts \\ []) View Source 
start_link([option()]) :: Supervisor.on_start()

Starts a new supervisor.

The supported options are:

  • :name - used to register a supervisor name, the supported values are described under the Name Registration section in the GenServer module docs;

  • :restart - the restart strategy, may be :temporary (the default), :transient or :permanent. :temporary means the task is never restarted, :transient means it is restarted if the exit is not :normal, :shutdown or {:shutdown, reason}. A :permanent restart strategy means it is always restarted. It defaults to :temporary so tasks aren’t automatically restarted when they complete nor in case of crashes. Note the :async functions in this module support only :temporary restarts;

  • :shutdown - :brutal_kill if the tasks must be killed directly on shutdown or an integer indicating the timeout value, defaults to 5000 milliseconds;

  • :max_restarts and :max_seconds - as specified in Supervisor;

Link to this function terminate_child(supervisor, pid) View Source 
terminate_child(Supervisor.supervisor(), pid()) :: :ok

Terminates the child with the given pid.