José Valim
3b77086073
One common concern about using Elixir libraries is that "use SomeModule" does not make clear how it impacts the caller. Most of the time those changes are minor but it is impossible to know without traversing the source code. This pull request suggests defining a summary alongside each module that succintly explains the impact of using it. This is documented as best practice in our guides and we added the note to all of Elixir modules.
1164 lines
36 KiB
Elixir
1164 lines
36 KiB
Elixir
defmodule DynamicSupervisor do
|
|
@moduledoc ~S"""
|
|
A supervisor optimized to only start children dynamically.
|
|
|
|
The `Supervisor` module was designed to handle mostly static children
|
|
that are started in the given order when the supervisor starts. A
|
|
`DynamicSupervisor` starts with no children. Instead, children are
|
|
started on demand via `start_child/2` and there is no ordering between
|
|
children. This allows the `DynamicSupervisor` to hold millions of
|
|
children by using efficient data structures and to execute certain
|
|
operations, such as shutting down, concurrently.
|
|
|
|
## Examples
|
|
|
|
A dynamic supervisor is started with no children and often a name:
|
|
|
|
children = [
|
|
{DynamicSupervisor, name: MyApp.DynamicSupervisor, strategy: :one_for_one}
|
|
]
|
|
|
|
Supervisor.start_link(children, strategy: :one_for_one)
|
|
|
|
The options given in the child specification are documented in `start_link/1`.
|
|
|
|
Once the dynamic supervisor is running, we can use it to start children
|
|
on demand. Given this sample `GenServer`:
|
|
|
|
defmodule Counter do
|
|
use GenServer
|
|
|
|
def start_link(initial) do
|
|
GenServer.start_link(__MODULE__, initial)
|
|
end
|
|
|
|
def inc(pid) do
|
|
GenServer.call(pid, :inc)
|
|
end
|
|
|
|
def init(initial) do
|
|
{:ok, initial}
|
|
end
|
|
|
|
def handle_call(:inc, _, count) do
|
|
{:reply, count, count + 1}
|
|
end
|
|
end
|
|
|
|
We can use `start_child/2` with a child specification to start a `Counter`
|
|
server:
|
|
|
|
{:ok, counter1} = DynamicSupervisor.start_child(MyApp.DynamicSupervisor, {Counter, 0})
|
|
Counter.inc(counter1)
|
|
#=> 0
|
|
|
|
{:ok, counter2} = DynamicSupervisor.start_child(MyApp.DynamicSupervisor, {Counter, 10})
|
|
Counter.inc(counter2)
|
|
#=> 10
|
|
|
|
DynamicSupervisor.count_children(MyApp.DynamicSupervisor)
|
|
#=> %{active: 2, specs: 2, supervisors: 0, workers: 2}
|
|
|
|
## Scalability and partitioning
|
|
|
|
The `DynamicSupervisor` is a single process responsible for starting
|
|
other processes. In some applications, the `DynamicSupervisor` may
|
|
become a bottleneck. To address this, you can start multiple instances
|
|
of the `DynamicSupervisor` and then pick a "random" instance to start
|
|
the child on.
|
|
|
|
Instead of:
|
|
|
|
children = [
|
|
{DynamicSupervisor, name: MyApp.DynamicSupervisor}
|
|
]
|
|
|
|
and:
|
|
|
|
DynamicSupervisor.start_child(MyApp.DynamicSupervisor, {Counter, 0})
|
|
|
|
You can do this:
|
|
|
|
children = [
|
|
{PartitionSupervisor,
|
|
child_spec: DynamicSupervisor,
|
|
name: MyApp.DynamicSupervisors}
|
|
]
|
|
|
|
and then:
|
|
|
|
DynamicSupervisor.start_child(
|
|
{:via, PartitionSupervisor, {MyApp.DynamicSupervisors, self()}},
|
|
{Counter, 0}
|
|
)
|
|
|
|
In the code above, we start a partition supervisor that will by default
|
|
start a dynamic supervisor for each core in your machine. Then, instead
|
|
of calling the `DynamicSupervisor` by name, you call it through the
|
|
partition supervisor, using `self()` as the routing key. This means each
|
|
process will be assigned one of the existing dynamic supervisors.
|
|
Read the `PartitionSupervisor` docs for more information.
|
|
|
|
## Module-based supervisors
|
|
|
|
Similar to `Supervisor`, dynamic supervisors also support module-based
|
|
supervisors.
|
|
|
|
defmodule MyApp.DynamicSupervisor do
|
|
# Automatically defines child_spec/1
|
|
use DynamicSupervisor
|
|
|
|
def start_link(init_arg) do
|
|
DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
|
|
end
|
|
|
|
@impl true
|
|
def init(_init_arg) do
|
|
DynamicSupervisor.init(strategy: :one_for_one)
|
|
end
|
|
end
|
|
|
|
See the `Supervisor` docs for a discussion of when you may want to use
|
|
module-based supervisors. A `@doc` annotation immediately preceding
|
|
`use DynamicSupervisor` will be attached to the generated `child_spec/1`
|
|
function.
|
|
|
|
> #### `use DynamicSupervisor` {: .info}
|
|
>
|
|
> When you `use DynamicSupervisor`, the `DynamicSupervisor` module will
|
|
> set `@behaviour DynamicSupervisor` and define a `child_spec/1`
|
|
> function, so your module can be used as a child in a supervision tree.
|
|
|
|
## Name registration
|
|
|
|
A supervisor is bound to the same name registration rules as a `GenServer`.
|
|
Read more about these rules in the documentation for `GenServer`.
|
|
|
|
## Migrating from Supervisor's :simple_one_for_one
|
|
|
|
In case you were using the deprecated `:simple_one_for_one` strategy from
|
|
the `Supervisor` module, you can migrate to the `DynamicSupervisor` in
|
|
few steps.
|
|
|
|
Imagine the given "old" code:
|
|
|
|
defmodule MySupervisor do
|
|
use Supervisor
|
|
|
|
def start_link(init_arg) do
|
|
Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
|
|
end
|
|
|
|
def start_child(foo, bar, baz) do
|
|
# This will start child by calling MyWorker.start_link(init_arg, foo, bar, baz)
|
|
Supervisor.start_child(__MODULE__, [foo, bar, baz])
|
|
end
|
|
|
|
@impl true
|
|
def init(init_arg) do
|
|
children = [
|
|
# Or the deprecated: worker(MyWorker, [init_arg])
|
|
%{id: MyWorker, start: {MyWorker, :start_link, [init_arg]}}
|
|
]
|
|
|
|
Supervisor.init(children, strategy: :simple_one_for_one)
|
|
end
|
|
end
|
|
|
|
It can be upgraded to the DynamicSupervisor like this:
|
|
|
|
defmodule MySupervisor do
|
|
use DynamicSupervisor
|
|
|
|
def start_link(init_arg) do
|
|
DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
|
|
end
|
|
|
|
def start_child(foo, bar, baz) do
|
|
# If MyWorker is not using the new child specs, we need to pass a map:
|
|
# spec = %{id: MyWorker, start: {MyWorker, :start_link, [foo, bar, baz]}}
|
|
spec = {MyWorker, foo: foo, bar: bar, baz: baz}
|
|
DynamicSupervisor.start_child(__MODULE__, spec)
|
|
end
|
|
|
|
@impl true
|
|
def init(init_arg) do
|
|
DynamicSupervisor.init(
|
|
strategy: :one_for_one,
|
|
extra_arguments: [init_arg]
|
|
)
|
|
end
|
|
end
|
|
|
|
The difference is that the `DynamicSupervisor` expects the child specification
|
|
at the moment `start_child/2` is called, and no longer on the init callback.
|
|
If there are any initial arguments given on initialization, such as `[initial_arg]`,
|
|
it can be given in the `:extra_arguments` flag on `DynamicSupervisor.init/1`.
|
|
"""
|
|
|
|
@behaviour GenServer
|
|
|
|
@doc """
|
|
Callback invoked to start the supervisor and during hot code upgrades.
|
|
|
|
Developers typically invoke `DynamicSupervisor.init/1` at the end of
|
|
their init callback to return the proper supervision flags.
|
|
"""
|
|
@callback init(init_arg :: term) :: {:ok, sup_flags()} | :ignore
|
|
|
|
@typedoc "The supervisor flags returned on init"
|
|
@type sup_flags() :: %{
|
|
strategy: strategy(),
|
|
intensity: non_neg_integer(),
|
|
period: pos_integer(),
|
|
max_children: non_neg_integer() | :infinity,
|
|
extra_arguments: [term()]
|
|
}
|
|
|
|
@typedoc "Options given to `start_link` functions"
|
|
@type option :: GenServer.option()
|
|
|
|
@typedoc "Options given to `start_link` and `init/1` functions"
|
|
@type init_option ::
|
|
{:strategy, strategy()}
|
|
| {:max_restarts, non_neg_integer()}
|
|
| {:max_seconds, pos_integer()}
|
|
| {:max_children, non_neg_integer() | :infinity}
|
|
| {:extra_arguments, [term()]}
|
|
|
|
@typedoc "Supported strategies"
|
|
@type strategy :: :one_for_one
|
|
|
|
@typedoc "Return values of `start_child` functions"
|
|
@type on_start_child ::
|
|
{:ok, pid}
|
|
| {:ok, pid, info :: term}
|
|
| :ignore
|
|
| {:error, {:already_started, pid} | :max_children | term}
|
|
|
|
# In this struct, `args` refers to the arguments passed to init/1 (the `init_arg`).
|
|
defstruct [
|
|
:args,
|
|
:extra_arguments,
|
|
:mod,
|
|
:name,
|
|
:strategy,
|
|
:max_children,
|
|
:max_restarts,
|
|
:max_seconds,
|
|
children: %{},
|
|
restarts: []
|
|
]
|
|
|
|
@doc """
|
|
Returns a specification to start a dynamic supervisor under a supervisor.
|
|
|
|
It accepts the same options as `start_link/1`.
|
|
|
|
See `Supervisor` for more information about child specifications.
|
|
"""
|
|
@doc since: "1.6.1"
|
|
def child_spec(options) when is_list(options) do
|
|
id =
|
|
case Keyword.get(options, :name, DynamicSupervisor) do
|
|
name when is_atom(name) -> name
|
|
{:global, name} -> name
|
|
{:via, _module, name} -> name
|
|
end
|
|
|
|
%{
|
|
id: id,
|
|
start: {DynamicSupervisor, :start_link, [options]},
|
|
type: :supervisor
|
|
}
|
|
end
|
|
|
|
@doc false
|
|
defmacro __using__(opts) do
|
|
quote location: :keep, bind_quoted: [opts: opts] do
|
|
@behaviour DynamicSupervisor
|
|
unless Module.has_attribute?(__MODULE__, :doc) do
|
|
@doc """
|
|
Returns a specification to start this module under a supervisor.
|
|
|
|
See `Supervisor`.
|
|
"""
|
|
end
|
|
|
|
def child_spec(arg) do
|
|
default = %{
|
|
id: __MODULE__,
|
|
start: {__MODULE__, :start_link, [arg]},
|
|
type: :supervisor
|
|
}
|
|
|
|
Supervisor.child_spec(default, unquote(Macro.escape(opts)))
|
|
end
|
|
|
|
defoverridable child_spec: 1
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Starts a supervisor with the given options.
|
|
|
|
This function is typically not invoked directly, instead it is invoked
|
|
when using a `DynamicSupervisor` as a child of another supervisor:
|
|
|
|
children = [
|
|
{DynamicSupervisor, name: MySupervisor}
|
|
]
|
|
|
|
If the supervisor is successfully spawned, this function returns
|
|
`{:ok, pid}`, where `pid` is the PID of the supervisor. If the supervisor
|
|
is given a name and a process with the specified name already exists,
|
|
the function returns `{:error, {:already_started, pid}}`, where `pid`
|
|
is the PID of that process.
|
|
|
|
Note that a supervisor started with this function is linked to the parent
|
|
process and exits not only on crashes but also if the parent process exits
|
|
with `:normal` reason.
|
|
|
|
## Options
|
|
|
|
* `:name` - registers the supervisor under the given name.
|
|
The supported values are described under the "Name registration"
|
|
section in the `GenServer` module docs.
|
|
|
|
* `:strategy` - the restart strategy option. The only supported
|
|
value is `:one_for_one` which means that no other child is
|
|
terminated if a child process terminates. You can learn more
|
|
about strategies in the `Supervisor` module docs.
|
|
|
|
* `:max_restarts` - the maximum number of restarts allowed in
|
|
a time frame. Defaults to `3`.
|
|
|
|
* `:max_seconds` - the time frame in which `:max_restarts` applies.
|
|
Defaults to `5`.
|
|
|
|
* `:max_children` - the maximum amount of children to be running
|
|
under this supervisor at the same time. When `:max_children` is
|
|
exceeded, `start_child/2` returns `{:error, :max_children}`. Defaults
|
|
to `:infinity`.
|
|
|
|
* `:extra_arguments` - arguments that are prepended to the arguments
|
|
specified in the child spec given to `start_child/2`. Defaults to
|
|
an empty list.
|
|
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec start_link([option | init_option]) :: Supervisor.on_start()
|
|
def start_link(options) when is_list(options) do
|
|
keys = [:extra_arguments, :max_children, :max_seconds, :max_restarts, :strategy]
|
|
{sup_opts, start_opts} = Keyword.split(options, keys)
|
|
start_link(Supervisor.Default, init(sup_opts), start_opts)
|
|
end
|
|
|
|
@doc """
|
|
Starts a module-based supervisor process with the given `module` and `init_arg`.
|
|
|
|
To start the supervisor, the `c:init/1` callback will be invoked in the given
|
|
`module`, with `init_arg` as its argument. The `c:init/1` callback must return a
|
|
supervisor specification which can be created with the help of the `init/1`
|
|
function.
|
|
|
|
If the `c:init/1` callback returns `:ignore`, this function returns
|
|
`:ignore` as well and the supervisor terminates with reason `:normal`.
|
|
If it fails or returns an incorrect value, this function returns
|
|
`{:error, term}` where `term` is a term with information about the
|
|
error, and the supervisor terminates with reason `term`.
|
|
|
|
The `:name` option can also be given in order to register a supervisor
|
|
name, the supported values are described in the "Name registration"
|
|
section in the `GenServer` module docs.
|
|
|
|
If the supervisor is successfully spawned, this function returns
|
|
`{:ok, pid}`, where `pid` is the PID of the supervisor. If the supervisor
|
|
is given a name and a process with the specified name already exists,
|
|
the function returns `{:error, {:already_started, pid}}`, where `pid`
|
|
is the PID of that process.
|
|
|
|
Note that a supervisor started with this function is linked to the parent
|
|
process and exits not only on crashes but also if the parent process exits
|
|
with `:normal` reason.
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec start_link(module, term, [option]) :: Supervisor.on_start()
|
|
def start_link(module, init_arg, opts \\ []) do
|
|
GenServer.start_link(__MODULE__, {module, init_arg, opts[:name]}, opts)
|
|
end
|
|
|
|
@doc """
|
|
Dynamically adds a child specification to `supervisor` and starts that child.
|
|
|
|
`child_spec` should be a valid child specification as detailed in the
|
|
"Child specification" section of the documentation for `Supervisor`. The child
|
|
process will be started as defined in the child specification. Note that while
|
|
the `:id` field is still required in the spec, the value is ignored and
|
|
therefore does not need to be unique.
|
|
|
|
If the child process start function returns `{:ok, child}` or `{:ok, child,
|
|
info}`, then child specification and PID are added to the supervisor and
|
|
this function returns the same value.
|
|
|
|
If the child process start function returns `:ignore`, then no child is added
|
|
to the supervision tree and this function returns `:ignore` too.
|
|
|
|
If the child process start function returns an error tuple or an erroneous
|
|
value, or if it fails, the child specification is discarded and this function
|
|
returns `{:error, error}` where `error` is the error or erroneous value
|
|
returned from child process start function, or failure reason if it fails.
|
|
|
|
If the supervisor already has N children in a way that N exceeds the amount
|
|
of `:max_children` set on the supervisor initialization (see `init/1`), then
|
|
this function returns `{:error, :max_children}`.
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec start_child(
|
|
Supervisor.supervisor(),
|
|
Supervisor.child_spec()
|
|
| {module, term}
|
|
| module
|
|
| (old_erlang_child_spec :: :supervisor.child_spec())
|
|
) ::
|
|
on_start_child()
|
|
def start_child(supervisor, {_, _, _, _, _, _} = child_spec) do
|
|
validate_and_start_child(supervisor, child_spec)
|
|
end
|
|
|
|
def start_child(supervisor, child_spec) do
|
|
validate_and_start_child(supervisor, Supervisor.child_spec(child_spec, []))
|
|
end
|
|
|
|
defp validate_and_start_child(supervisor, child_spec) do
|
|
case validate_child(child_spec) do
|
|
{:ok, child} -> call(supervisor, {:start_child, child})
|
|
error -> {:error, error}
|
|
end
|
|
end
|
|
|
|
defp validate_child(%{id: _, start: {mod, _, _} = start} = child) do
|
|
restart = Map.get(child, :restart, :permanent)
|
|
type = Map.get(child, :type, :worker)
|
|
modules = Map.get(child, :modules, [mod])
|
|
significant = Map.get(child, :significant, false)
|
|
|
|
shutdown =
|
|
case type do
|
|
:worker -> Map.get(child, :shutdown, 5_000)
|
|
:supervisor -> Map.get(child, :shutdown, :infinity)
|
|
end
|
|
|
|
validate_child(start, restart, shutdown, type, modules, significant)
|
|
end
|
|
|
|
defp validate_child({_, start, restart, shutdown, type, modules}) do
|
|
validate_child(start, restart, shutdown, type, modules, false)
|
|
end
|
|
|
|
defp validate_child(other) do
|
|
{:invalid_child_spec, other}
|
|
end
|
|
|
|
defp validate_child(start, restart, shutdown, type, modules, significant) do
|
|
with :ok <- validate_start(start),
|
|
:ok <- validate_restart(restart),
|
|
:ok <- validate_shutdown(shutdown),
|
|
:ok <- validate_type(type),
|
|
:ok <- validate_modules(modules),
|
|
:ok <- validate_significant(significant) do
|
|
{:ok, {start, restart, shutdown, type, modules}}
|
|
end
|
|
end
|
|
|
|
defp validate_start({m, f, args}) when is_atom(m) and is_atom(f) and is_list(args), do: :ok
|
|
defp validate_start(mfa), do: {:invalid_mfa, mfa}
|
|
|
|
defp validate_type(type) when type in [:supervisor, :worker], do: :ok
|
|
defp validate_type(type), do: {:invalid_child_type, type}
|
|
|
|
defp validate_restart(restart) when restart in [:permanent, :temporary, :transient], do: :ok
|
|
defp validate_restart(restart), do: {:invalid_restart_type, restart}
|
|
|
|
defp validate_shutdown(shutdown) when is_integer(shutdown) and shutdown >= 0, do: :ok
|
|
defp validate_shutdown(shutdown) when shutdown in [:infinity, :brutal_kill], do: :ok
|
|
defp validate_shutdown(shutdown), do: {:invalid_shutdown, shutdown}
|
|
|
|
defp validate_significant(false), do: :ok
|
|
defp validate_significant(significant), do: {:invalid_significant, significant}
|
|
|
|
defp validate_modules(:dynamic), do: :ok
|
|
|
|
defp validate_modules(mods) do
|
|
if is_list(mods) and Enum.all?(mods, &is_atom/1) do
|
|
:ok
|
|
else
|
|
{:invalid_modules, mods}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Terminates the given child identified by `pid`.
|
|
|
|
If successful, this function returns `:ok`. If there is no process with
|
|
the given PID, this function returns `{:error, :not_found}`.
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec terminate_child(Supervisor.supervisor(), pid) :: :ok | {:error, :not_found}
|
|
def terminate_child(supervisor, pid) when is_pid(pid) do
|
|
call(supervisor, {:terminate_child, pid})
|
|
end
|
|
|
|
@doc """
|
|
Returns a list with information about all children.
|
|
|
|
Note that calling this function when supervising a large number
|
|
of children under low memory conditions can cause an out of memory
|
|
exception.
|
|
|
|
This function returns a list of tuples containing:
|
|
|
|
* `id` - it is always `:undefined` for dynamic supervisors
|
|
|
|
* `child` - the PID of the corresponding child process or the
|
|
atom `:restarting` if the process is about to be restarted
|
|
|
|
* `type` - `:worker` or `:supervisor` as defined in the child
|
|
specification
|
|
|
|
* `modules` - as defined in the child specification
|
|
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec which_children(Supervisor.supervisor()) :: [
|
|
# module() | :dynamic here because :supervisor.modules() is not exported
|
|
{:undefined, pid | :restarting, :worker | :supervisor, [module()] | :dynamic}
|
|
]
|
|
def which_children(supervisor) do
|
|
call(supervisor, :which_children)
|
|
end
|
|
|
|
@doc """
|
|
Returns a map containing count values for the supervisor.
|
|
|
|
The map contains the following keys:
|
|
|
|
* `:specs` - the number of children processes
|
|
|
|
* `:active` - the count of all actively running child processes managed by
|
|
this supervisor
|
|
|
|
* `:supervisors` - the count of all supervisors whether or not the child
|
|
process is still alive
|
|
|
|
* `:workers` - the count of all workers, whether or not the child process
|
|
is still alive
|
|
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec count_children(Supervisor.supervisor()) :: %{
|
|
specs: non_neg_integer,
|
|
active: non_neg_integer,
|
|
supervisors: non_neg_integer,
|
|
workers: non_neg_integer
|
|
}
|
|
def count_children(supervisor) do
|
|
call(supervisor, :count_children) |> :maps.from_list()
|
|
end
|
|
|
|
@doc """
|
|
Synchronously stops the given supervisor with the given `reason`.
|
|
|
|
It returns `:ok` if the supervisor terminates with the given
|
|
reason. If it terminates with another reason, the call exits.
|
|
|
|
This function keeps OTP semantics regarding error reporting.
|
|
If the reason is any other than `:normal`, `:shutdown` or
|
|
`{:shutdown, _}`, an error report is logged.
|
|
"""
|
|
@doc since: "1.7.0"
|
|
@spec stop(Supervisor.supervisor(), reason :: term, timeout) :: :ok
|
|
def stop(supervisor, reason \\ :normal, timeout \\ :infinity) do
|
|
GenServer.stop(supervisor, reason, timeout)
|
|
end
|
|
|
|
@doc """
|
|
Receives a set of `options` that initializes a dynamic supervisor.
|
|
|
|
This is typically invoked at the end of the `c:init/1` callback of
|
|
module-based supervisors. See the "Module-based supervisors" section
|
|
in the module documentation for more information.
|
|
|
|
It accepts the same `options` as `start_link/1` (except for `:name`)
|
|
and it returns a tuple containing the supervisor options.
|
|
|
|
## Examples
|
|
|
|
def init(_arg) do
|
|
DynamicSupervisor.init(max_children: 1000)
|
|
end
|
|
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec init([init_option]) :: {:ok, sup_flags()}
|
|
def init(options) when is_list(options) do
|
|
strategy = Keyword.get(options, :strategy, :one_for_one)
|
|
intensity = Keyword.get(options, :max_restarts, 3)
|
|
period = Keyword.get(options, :max_seconds, 5)
|
|
max_children = Keyword.get(options, :max_children, :infinity)
|
|
extra_arguments = Keyword.get(options, :extra_arguments, [])
|
|
|
|
flags = %{
|
|
strategy: strategy,
|
|
intensity: intensity,
|
|
period: period,
|
|
max_children: max_children,
|
|
extra_arguments: extra_arguments
|
|
}
|
|
|
|
{:ok, flags}
|
|
end
|
|
|
|
## Callbacks
|
|
|
|
@impl true
|
|
def init({mod, init_arg, name}) do
|
|
Process.put(:"$initial_call", {:supervisor, mod, 1})
|
|
Process.flag(:trap_exit, true)
|
|
|
|
case mod.init(init_arg) do
|
|
{:ok, flags} when is_map(flags) ->
|
|
name =
|
|
cond do
|
|
is_nil(name) -> {self(), mod}
|
|
is_atom(name) -> {:local, name}
|
|
is_tuple(name) -> name
|
|
end
|
|
|
|
state = %DynamicSupervisor{mod: mod, args: init_arg, name: name}
|
|
|
|
case init(state, flags) do
|
|
{:ok, state} -> {:ok, state}
|
|
{:error, reason} -> {:stop, {:supervisor_data, reason}}
|
|
end
|
|
|
|
:ignore ->
|
|
:ignore
|
|
|
|
other ->
|
|
{:stop, {:bad_return, {mod, :init, other}}}
|
|
end
|
|
end
|
|
|
|
defp init(state, flags) do
|
|
extra_arguments = Map.get(flags, :extra_arguments, [])
|
|
max_children = Map.get(flags, :max_children, :infinity)
|
|
max_restarts = Map.get(flags, :intensity, 1)
|
|
max_seconds = Map.get(flags, :period, 5)
|
|
strategy = Map.get(flags, :strategy, :one_for_one)
|
|
auto_shutdown = Map.get(flags, :auto_shutdown, :never)
|
|
|
|
with :ok <- validate_strategy(strategy),
|
|
:ok <- validate_restarts(max_restarts),
|
|
:ok <- validate_seconds(max_seconds),
|
|
:ok <- validate_dynamic(max_children),
|
|
:ok <- validate_extra_arguments(extra_arguments),
|
|
:ok <- validate_auto_shutdown(auto_shutdown) do
|
|
{:ok,
|
|
%{
|
|
state
|
|
| extra_arguments: extra_arguments,
|
|
max_children: max_children,
|
|
max_restarts: max_restarts,
|
|
max_seconds: max_seconds,
|
|
strategy: strategy
|
|
}}
|
|
end
|
|
end
|
|
|
|
defp validate_strategy(strategy) when strategy in [:one_for_one], do: :ok
|
|
defp validate_strategy(strategy), do: {:error, {:invalid_strategy, strategy}}
|
|
|
|
defp validate_restarts(restart) when is_integer(restart) and restart >= 0, do: :ok
|
|
defp validate_restarts(restart), do: {:error, {:invalid_intensity, restart}}
|
|
|
|
defp validate_seconds(seconds) when is_integer(seconds) and seconds > 0, do: :ok
|
|
defp validate_seconds(seconds), do: {:error, {:invalid_period, seconds}}
|
|
|
|
defp validate_dynamic(:infinity), do: :ok
|
|
defp validate_dynamic(dynamic) when is_integer(dynamic) and dynamic >= 0, do: :ok
|
|
defp validate_dynamic(dynamic), do: {:error, {:invalid_max_children, dynamic}}
|
|
|
|
defp validate_extra_arguments(list) when is_list(list), do: :ok
|
|
defp validate_extra_arguments(extra), do: {:error, {:invalid_extra_arguments, extra}}
|
|
|
|
defp validate_auto_shutdown(auto_shutdown) when auto_shutdown in [:never], do: :ok
|
|
|
|
defp validate_auto_shutdown(auto_shutdown),
|
|
do: {:error, {:invalid_auto_shutdown, auto_shutdown}}
|
|
|
|
@impl true
|
|
def handle_call(:which_children, _from, state) do
|
|
%{children: children} = state
|
|
|
|
reply =
|
|
for {pid, args} <- children do
|
|
case args do
|
|
{:restarting, {_, _, _, type, modules}} ->
|
|
{:undefined, :restarting, type, modules}
|
|
|
|
{_, _, _, type, modules} ->
|
|
{:undefined, pid, type, modules}
|
|
end
|
|
end
|
|
|
|
{:reply, reply, state}
|
|
end
|
|
|
|
def handle_call(:count_children, _from, state) do
|
|
%{children: children} = state
|
|
specs = map_size(children)
|
|
|
|
{active, workers, supervisors} =
|
|
Enum.reduce(children, {0, 0, 0}, fn
|
|
{_pid, {:restarting, {_, _, _, :worker, _}}}, {active, worker, supervisor} ->
|
|
{active, worker + 1, supervisor}
|
|
|
|
{_pid, {:restarting, {_, _, _, :supervisor, _}}}, {active, worker, supervisor} ->
|
|
{active, worker, supervisor + 1}
|
|
|
|
{_pid, {_, _, _, :worker, _}}, {active, worker, supervisor} ->
|
|
{active + 1, worker + 1, supervisor}
|
|
|
|
{_pid, {_, _, _, :supervisor, _}}, {active, worker, supervisor} ->
|
|
{active + 1, worker, supervisor + 1}
|
|
end)
|
|
|
|
reply = [specs: specs, active: active, supervisors: supervisors, workers: workers]
|
|
{:reply, reply, state}
|
|
end
|
|
|
|
def handle_call({:terminate_child, pid}, _from, %{children: children} = state) do
|
|
case children do
|
|
%{^pid => info} ->
|
|
:ok = terminate_children(%{pid => info}, state)
|
|
{:reply, :ok, delete_child(pid, state)}
|
|
|
|
%{} ->
|
|
{:reply, {:error, :not_found}, state}
|
|
end
|
|
end
|
|
|
|
def handle_call({:start_task, args, restart, shutdown}, from, state) do
|
|
{init_restart, init_shutdown} = Process.get(Task.Supervisor)
|
|
|
|
if restart == nil and init_restart != :temporary do
|
|
{:reply, {:restart, init_restart}, state}
|
|
else
|
|
restart = restart || init_restart
|
|
shutdown = shutdown || init_shutdown
|
|
|
|
child =
|
|
{{Task.Supervised, :start_link, args}, restart, shutdown, :worker, [Task.Supervised]}
|
|
|
|
handle_call({:start_child, child}, from, state)
|
|
end
|
|
end
|
|
|
|
def handle_call({:start_child, child}, _from, state) do
|
|
%{children: children, max_children: max_children} = state
|
|
|
|
if map_size(children) < max_children do
|
|
handle_start_child(child, state)
|
|
else
|
|
{:reply, {:error, :max_children}, state}
|
|
end
|
|
end
|
|
|
|
defp handle_start_child({{m, f, args} = mfa, restart, shutdown, type, modules}, state) do
|
|
%{extra_arguments: extra} = state
|
|
|
|
case reply = start_child(m, f, extra ++ args) do
|
|
{:ok, pid, _} ->
|
|
{:reply, reply, save_child(pid, mfa, restart, shutdown, type, modules, state)}
|
|
|
|
{:ok, pid} ->
|
|
{:reply, reply, save_child(pid, mfa, restart, shutdown, type, modules, state)}
|
|
|
|
_ ->
|
|
{:reply, reply, state}
|
|
end
|
|
end
|
|
|
|
defp start_child(m, f, a) do
|
|
try do
|
|
apply(m, f, a)
|
|
catch
|
|
kind, reason ->
|
|
{:error, exit_reason(kind, reason, __STACKTRACE__)}
|
|
else
|
|
{:ok, pid, extra} when is_pid(pid) -> {:ok, pid, extra}
|
|
{:ok, pid} when is_pid(pid) -> {:ok, pid}
|
|
:ignore -> :ignore
|
|
{:error, _} = error -> error
|
|
other -> {:error, other}
|
|
end
|
|
end
|
|
|
|
defp save_child(pid, mfa, restart, shutdown, type, modules, state) do
|
|
mfa = mfa_for_restart(mfa, restart)
|
|
put_in(state.children[pid], {mfa, restart, shutdown, type, modules})
|
|
end
|
|
|
|
defp mfa_for_restart({m, f, _}, :temporary), do: {m, f, :undefined}
|
|
defp mfa_for_restart(mfa, _), do: mfa
|
|
|
|
defp exit_reason(:exit, reason, _), do: reason
|
|
defp exit_reason(:error, reason, stack), do: {reason, stack}
|
|
defp exit_reason(:throw, value, stack), do: {{:nocatch, value}, stack}
|
|
|
|
@impl true
|
|
def handle_cast(_msg, state) do
|
|
{:noreply, state}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:EXIT, pid, reason}, state) do
|
|
case maybe_restart_child(pid, reason, state) do
|
|
{:ok, state} -> {:noreply, state}
|
|
{:shutdown, state} -> {:stop, :shutdown, state}
|
|
end
|
|
end
|
|
|
|
def handle_info({:"$gen_restart", pid}, state) do
|
|
%{children: children} = state
|
|
|
|
case children do
|
|
%{^pid => restarting_args} ->
|
|
{:restarting, child} = restarting_args
|
|
|
|
case restart_child(pid, child, state) do
|
|
{:ok, state} -> {:noreply, state}
|
|
{:shutdown, state} -> {:stop, :shutdown, state}
|
|
end
|
|
|
|
# We may hit clause if we send $gen_restart and then
|
|
# someone calls terminate_child, removing the child.
|
|
%{} ->
|
|
{:noreply, state}
|
|
end
|
|
end
|
|
|
|
def handle_info(msg, state) do
|
|
:logger.error(
|
|
%{
|
|
label: {DynamicSupervisor, :unexpected_msg},
|
|
report: %{
|
|
msg: msg
|
|
}
|
|
},
|
|
%{
|
|
domain: [:otp, :elixir],
|
|
error_logger: %{tag: :error_msg},
|
|
report_cb: &__MODULE__.format_report/1
|
|
}
|
|
)
|
|
|
|
{:noreply, state}
|
|
end
|
|
|
|
@impl true
|
|
def code_change(_, %{mod: mod, args: init_arg} = state, _) do
|
|
case mod.init(init_arg) do
|
|
{:ok, flags} when is_map(flags) ->
|
|
case init(state, flags) do
|
|
{:ok, state} -> {:ok, state}
|
|
{:error, reason} -> {:error, {:supervisor_data, reason}}
|
|
end
|
|
|
|
:ignore ->
|
|
{:ok, state}
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
@impl true
|
|
def terminate(_, %{children: children} = state) do
|
|
:ok = terminate_children(children, state)
|
|
end
|
|
|
|
defp terminate_children(children, state) do
|
|
{pids, times, stacks} = monitor_children(children)
|
|
size = map_size(pids)
|
|
|
|
timers =
|
|
Enum.reduce(times, %{}, fn {time, pids}, acc ->
|
|
Map.put(acc, :erlang.start_timer(time, self(), :kill), pids)
|
|
end)
|
|
|
|
stacks = wait_children(pids, size, timers, stacks)
|
|
|
|
for {pid, {child, reason}} <- stacks do
|
|
report_error(:shutdown_error, reason, pid, child, state)
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
defp monitor_children(children) do
|
|
Enum.reduce(children, {%{}, %{}, %{}}, fn
|
|
{_, {:restarting, _}}, acc ->
|
|
acc
|
|
|
|
{pid, {_, restart, _, _, _} = child}, {pids, times, stacks} ->
|
|
case monitor_child(pid) do
|
|
:ok ->
|
|
times = exit_child(pid, child, times)
|
|
{Map.put(pids, pid, child), times, stacks}
|
|
|
|
{:error, :normal} when restart != :permanent ->
|
|
{pids, times, stacks}
|
|
|
|
{:error, reason} ->
|
|
{pids, times, Map.put(stacks, pid, {child, reason})}
|
|
end
|
|
end)
|
|
end
|
|
|
|
defp monitor_child(pid) do
|
|
ref = Process.monitor(pid)
|
|
Process.unlink(pid)
|
|
|
|
receive do
|
|
{:EXIT, ^pid, reason} ->
|
|
receive do
|
|
{:DOWN, ^ref, :process, ^pid, _} -> {:error, reason}
|
|
end
|
|
after
|
|
0 -> :ok
|
|
end
|
|
end
|
|
|
|
defp exit_child(pid, {_, _, shutdown, _, _}, times) do
|
|
case shutdown do
|
|
:brutal_kill ->
|
|
Process.exit(pid, :kill)
|
|
times
|
|
|
|
:infinity ->
|
|
Process.exit(pid, :shutdown)
|
|
times
|
|
|
|
time ->
|
|
Process.exit(pid, :shutdown)
|
|
Map.update(times, time, [pid], &[pid | &1])
|
|
end
|
|
end
|
|
|
|
defp wait_children(_pids, 0, timers, stacks) do
|
|
for {timer, _} <- timers do
|
|
_ = :erlang.cancel_timer(timer)
|
|
|
|
receive do
|
|
{:timeout, ^timer, :kill} -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
end
|
|
|
|
stacks
|
|
end
|
|
|
|
defp wait_children(pids, size, timers, stacks) do
|
|
receive do
|
|
{:DOWN, _ref, :process, pid, reason} ->
|
|
case pids do
|
|
%{^pid => child} ->
|
|
stacks = wait_child(pid, child, reason, stacks)
|
|
wait_children(pids, size - 1, timers, stacks)
|
|
|
|
%{} ->
|
|
wait_children(pids, size, timers, stacks)
|
|
end
|
|
|
|
{:timeout, timer, :kill} ->
|
|
for pid <- Map.fetch!(timers, timer), do: Process.exit(pid, :kill)
|
|
wait_children(pids, size, Map.delete(timers, timer), stacks)
|
|
end
|
|
end
|
|
|
|
defp wait_child(pid, {_, _, :brutal_kill, _, _} = child, reason, stacks) do
|
|
case reason do
|
|
:killed -> stacks
|
|
_ -> Map.put(stacks, pid, {child, reason})
|
|
end
|
|
end
|
|
|
|
defp wait_child(pid, {_, restart, _, _, _} = child, reason, stacks) do
|
|
case reason do
|
|
{:shutdown, _} -> stacks
|
|
:shutdown -> stacks
|
|
:normal when restart != :permanent -> stacks
|
|
reason -> Map.put(stacks, pid, {child, reason})
|
|
end
|
|
end
|
|
|
|
defp maybe_restart_child(pid, reason, %{children: children} = state) do
|
|
case children do
|
|
%{^pid => {_, restart, _, _, _} = child} ->
|
|
maybe_restart_child(restart, reason, pid, child, state)
|
|
|
|
%{} ->
|
|
{:ok, state}
|
|
end
|
|
end
|
|
|
|
defp maybe_restart_child(:permanent, reason, pid, child, state) do
|
|
report_error(:child_terminated, reason, pid, child, state)
|
|
restart_child(pid, child, state)
|
|
end
|
|
|
|
defp maybe_restart_child(_, :normal, pid, _child, state) do
|
|
{:ok, delete_child(pid, state)}
|
|
end
|
|
|
|
defp maybe_restart_child(_, :shutdown, pid, _child, state) do
|
|
{:ok, delete_child(pid, state)}
|
|
end
|
|
|
|
defp maybe_restart_child(_, {:shutdown, _}, pid, _child, state) do
|
|
{:ok, delete_child(pid, state)}
|
|
end
|
|
|
|
defp maybe_restart_child(:transient, reason, pid, child, state) do
|
|
report_error(:child_terminated, reason, pid, child, state)
|
|
restart_child(pid, child, state)
|
|
end
|
|
|
|
defp maybe_restart_child(:temporary, reason, pid, child, state) do
|
|
report_error(:child_terminated, reason, pid, child, state)
|
|
{:ok, delete_child(pid, state)}
|
|
end
|
|
|
|
defp delete_child(pid, %{children: children} = state) do
|
|
%{state | children: Map.delete(children, pid)}
|
|
end
|
|
|
|
defp restart_child(pid, child, state) do
|
|
case add_restart(state) do
|
|
{:ok, %{strategy: strategy} = state} ->
|
|
case restart_child(strategy, pid, child, state) do
|
|
{:ok, state} ->
|
|
{:ok, state}
|
|
|
|
{:try_again, state} ->
|
|
send(self(), {:"$gen_restart", pid})
|
|
{:ok, state}
|
|
end
|
|
|
|
{:shutdown, state} ->
|
|
report_error(:shutdown, :reached_max_restart_intensity, pid, child, state)
|
|
{:shutdown, delete_child(pid, state)}
|
|
end
|
|
end
|
|
|
|
defp add_restart(state) do
|
|
%{max_seconds: max_seconds, max_restarts: max_restarts, restarts: restarts} = state
|
|
|
|
now = :erlang.monotonic_time(1)
|
|
restarts = add_restart([now | restarts], now, max_seconds)
|
|
state = %{state | restarts: restarts}
|
|
|
|
if length(restarts) <= max_restarts do
|
|
{:ok, state}
|
|
else
|
|
{:shutdown, state}
|
|
end
|
|
end
|
|
|
|
defp add_restart(restarts, now, period) do
|
|
for then <- restarts, now <= then + period, do: then
|
|
end
|
|
|
|
defp restart_child(:one_for_one, current_pid, child, state) do
|
|
{{m, f, args} = mfa, restart, shutdown, type, modules} = child
|
|
%{extra_arguments: extra} = state
|
|
|
|
case start_child(m, f, extra ++ args) do
|
|
{:ok, pid, _} ->
|
|
state = delete_child(current_pid, state)
|
|
{:ok, save_child(pid, mfa, restart, shutdown, type, modules, state)}
|
|
|
|
{:ok, pid} ->
|
|
state = delete_child(current_pid, state)
|
|
{:ok, save_child(pid, mfa, restart, shutdown, type, modules, state)}
|
|
|
|
:ignore ->
|
|
{:ok, delete_child(current_pid, state)}
|
|
|
|
{:error, reason} ->
|
|
report_error(:start_error, reason, {:restarting, current_pid}, child, state)
|
|
state = put_in(state.children[current_pid], {:restarting, child})
|
|
{:try_again, state}
|
|
end
|
|
end
|
|
|
|
defp report_error(error, reason, pid, child, %{name: name, extra_arguments: extra}) do
|
|
:logger.error(
|
|
%{
|
|
label: {:supervisor, error},
|
|
report: [
|
|
{:supervisor, name},
|
|
{:errorContext, error},
|
|
{:reason, reason},
|
|
{:offender, extract_child(pid, child, extra)}
|
|
]
|
|
},
|
|
%{
|
|
domain: [:otp, :sasl],
|
|
report_cb: &:logger.format_otp_report/1,
|
|
logger_formatter: %{title: "SUPERVISOR REPORT"},
|
|
error_logger: %{tag: :error_report, type: :supervisor_report}
|
|
}
|
|
)
|
|
end
|
|
|
|
defp extract_child(pid, {{m, f, args}, restart, shutdown, type, _modules}, extra) do
|
|
[
|
|
pid: pid,
|
|
id: :undefined,
|
|
mfargs: {m, f, extra ++ args},
|
|
restart_type: restart,
|
|
shutdown: shutdown,
|
|
child_type: type
|
|
]
|
|
end
|
|
|
|
@impl true
|
|
def format_status(:terminate, [_pdict, state]) do
|
|
state
|
|
end
|
|
|
|
def format_status(_, [_pdict, %{mod: mod} = state]) do
|
|
[data: [{~c"State", state}], supervisor: [{~c"Callback", mod}]]
|
|
end
|
|
|
|
## Helpers
|
|
|
|
@compile {:inline, call: 2}
|
|
|
|
defp call(supervisor, req) do
|
|
GenServer.call(supervisor, req, :infinity)
|
|
end
|
|
|
|
@doc false
|
|
def format_report(%{
|
|
label: {__MODULE__, :unexpected_msg},
|
|
report: %{msg: msg}
|
|
}) do
|
|
{~c"DynamicSupervisor received unexpected message: ~p~n", [msg]}
|
|
end
|
|
end
|