SuperWorker.Supervisor (SuperWorker v0.6.0)

Copy Markdown View Source

Documentation for SuperWorker.Supervisor. This module is new model supervisor. Better support for modern applications. That is an all-in-one supervisor for Elixir application. Easy to identify the worker for directly communicating with it.

New supervisor supports the following features:

  • Group processes
  • Chain processes
  • Standalone processes

Group processes

Group processes are a set of processes that are started together. If one of the processes is crashed, depending on the restart strategy of group, only that process or all the processes will be restarted. Each group has a separated restart strategy.

Chain processes

Chain processes are a set of processes that support for chain processing. Each process in a chain has order to process data. The output of the previous process is passed to the next process.

Standalone processes

Standalone processes are independent processes that are started separately. Each process has its own restart strategy.

All type of processes can be started in parallel & can be stopped individually or in a group.

The restart strategy only works when the process is crashed, for normal exit, shutdown, or terminated it will be ignored.

Fault tolerance

Workers are spread over one or more partitions. Each partition is a plain process monitored by the supervisor master:

  • a crashing worker is restarted by its partition according to the restart strategy of its parent (standalone/group/chain) — one bad worker never affects workers on other partitions;
  • a crashing partition is detected via a monitor and restarted individually; the rest of the supervisor keeps serving requests;
  • partitions monitor the master, so nothing survives the supervisor itself;
  • link semantics with the process that started the supervisor are standard: an abnormal exit of that process stops the supervisor, a :normal exit does not.

Introspection

Running supervisors can be discovered and inspected at runtime:

Sup.running_supervisors()               # [{sup_id, pid}]
Sup.supervisor_info(:sup1)              # partitions health + counts
Sup.list_groups(:sup1)                  # groups with strategies & ids
Sup.list_chains(:sup1)                  # chains with config & ids
Sup.list_standalone_workers(:sup1)

Utilities

The library also ships small focused helpers:

  • SuperWorker.CircuitBreaker — protect external calls with a closed/open/half-open circuit (the protected function runs in the caller, never inside the breaker process);
  • SuperWorker.TermStorage — thin :persistent_term wrapper for rarely changing node-wide data;
  • SuperWorker.Supervisor.Utils.safe_call/1,3 — invoke user functions without letting exceptions escape ({:ok, result} / {:error, {kind, reason}}).

Examples

# Start a supervisor with 2 partitions & 2 groups:
alias SuperWorker.Supervisor, as: Sup

# Config for supervisor
opts = [id: :sup1, num_partitions: 2, link: false]

# Start supervisor
Sup.start_with_config(opts)

# Add group in runtime, you also can add group in config.
Sup.add_group(:sup1, [id: :group1, restart_strategy: :one_for_all])
Sup.add_group_worker(:sup1, :group1, {Dev, :task, [15]}, [id: :g1_1])

Sup.add_group(:sup1, [id: :group2, restart_strategy: :one_for_one])
Sup.add_group_worker(:sup1, :group2, fn ->
  receive do
    {:ping, ref, from} ->
      send(from, {:pong, ref})
    msg ->
      :ok
  end
end, [id: :g2_2])

ref = make_ref()
Sup.send_to_group_worker(:sup1, :group2, :g2_2, {:ping, ref, self()})

receive do
  {:pong, ^ref} ->
    :ok
end

Supervisor can add directly to other supervisor or by add config to config file or self start with start/start_link/start_with_config function.

Summary

Functions

Add a chain to the supervisor. Chain's options follow docs in Chain module.

Add a group to the supervisor. Group's options follow docs in Group module.

Add a worker to a group in the supervisor. Function's options follow Worker module. Support worker is function (mfa, anonymous function) or GenServer

Add a standalone worker process to the supervisor. function for start worker can be a function or a {module, function, arguments} or a GenServer. Standalone worker is run independently from other workers follow :one_to_one strategy. If worker crashes, it will check the restart strategy of worker then act accordingly.

Send data to all workers in current group of worker. Using for communite between workers in the same group.

Returns a specification to start this module under a supervisor.

get current group id of worker.

Get supervisor id in current process (not support for GenServer worker).

Lists the chains of a supervisor with their configuration and worker ids.

Lists the groups of a supervisor with their restart strategies and worker ids.

Lists the standalone workers of a supervisor.

Remove standalone worker from supervisor.

Check if supervisor is running.

Lists all supervisors started through this library on the current node.

Send data to the entry worker in the chain. If chain doesn't has any worker, it will be dropped.

Send data to a random worker in the group.

Send data to other worker in the same group.

Send data to a random worker in the same group.

Send data directly to the worker standalone in the supervisor.

Start supervisor run as independent process with default options.

Start supervisor as independent process (no link process).

Work like start_link/1 with default options.

start_link for using supervisor as child in other supervisor or link to current process.

Start supervisor with configurations (Keyword). For run standalone, please set option :link to false. For link to other process (not current process), please set link to pid of that process. result format: {:ok, pid} or {:error, reason}

Returns an overview of a running supervisor: partition health and counts of groups, chains and workers.

Types

t()

@type t() :: %SuperWorker.Supervisor{
  id: atom(),
  link: boolean(),
  master_pid: term(),
  num_partitions: pos_integer(),
  partitions: %{required(pos_integer()) => pid()},
  report_to: [pid() | {atom(), pid()}],
  table: nil
}

Functions

add_chain(sup_id, options, timeout \\ 5000)

@spec add_chain(atom(), list(), non_neg_integer()) :: {:ok, atom()} | {:error, any()}

Add a chain to the supervisor. Chain's options follow docs in Chain module.

add_chain_worker(sup_id, chain_id, mfa_or_fun, opts, timeout \\ 3000)

@spec add_chain_worker(
  atom(),
  atom(),
  {module(), atom(), list()} | fun() | module() | {module(), list()},
  list(),
  non_neg_integer()
) :: {:ok, atom()} | {:error, any()}

Add a worker to the chain in supervisor.

add_group(sup_id, options, timeout \\ 3000)

@spec add_group(atom(), list(), non_neg_integer()) :: {:ok, atom()} | {:error, any()}

Add a group to the supervisor. Group's options follow docs in Group module.

add_group_worker(sup_id, group_id, mfa_or_fun, opts, timeout \\ 3000)

@spec add_group_worker(
  atom(),
  atom(),
  {module(), atom(), list()} | fun() | module() | {module(), list()},
  list(),
  non_neg_integer()
) :: {:ok, atom()} | {:error, any()}

Add a worker to a group in the supervisor. Function's options follow Worker module. Support worker is function (mfa, anonymous function) or GenServer

add_standalone_worker(sup_id, mfa_or_fun, options \\ [], timeout \\ 3000)

@spec add_standalone_worker(
  atom(),
  {module(), atom(), list()} | fun() | module() | {module(), list()},
  list(),
  non_neg_integer()
) :: {:ok, atom()} | {:error, any()}

Add a standalone worker process to the supervisor. function for start worker can be a function or a {module, function, arguments} or a GenServer. Standalone worker is run independently from other workers follow :one_to_one strategy. If worker crashes, it will check the restart strategy of worker then act accordingly.

broadcast_to_group(sup_id, group_id, data, timeout \\ 3000)

@spec broadcast_to_group(atom(), atom(), any(), non_neg_integer()) ::
  :ok | {:error, any()}

Send data to all workers in a group.

broadcast_to_my_group(data)

@spec broadcast_to_my_group(any()) :: :ok | {:error, any()}

Send data to all workers in current group of worker. Using for communite between workers in the same group.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

count_workers_in_chain(sup_id, chain_id, timeout \\ 3000)

@spec count_workers_in_chain(atom(), any(), non_neg_integer()) ::
  {:ok, pos_integer()} | {:error, any()}

count workers in chain

count_workers_in_group(sup_id, group_id, timeout \\ 3000)

@spec count_workers_in_group(atom(), any(), non_neg_integer()) ::
  {:ok, non_neg_integer()} | {:error, term()}

count workers in group

get_my_group()

@spec get_my_group() :: atom() | nil

get current group id of worker.

get_my_supervisor()

@spec get_my_supervisor() :: atom() | nil

Get supervisor id in current process (not support for GenServer worker).

get_pid_chain_worker(sup_id, chain_id, worker_id, timeout \\ 3000)

@spec get_pid_chain_worker(atom(), any(), any(), non_neg_integer()) ::
  {:ok, any()} | {:error, any()}

get pid of chain worker

get_pid_group_worker(sup_id, group_id, worker_id, timeout \\ 3000)

@spec get_pid_group_worker(atom(), any(), any(), non_neg_integer()) ::
  pid() | {:error, any()}

get pid of group worker.

get_pid_standalone_worker(sup_id, worker_id, timeout \\ 3000)

@spec get_pid_standalone_worker(atom(), any(), non_neg_integer()) ::
  {:ok, pid()} | {:error, term()}

get pid of standalone worker

group_exists?(sup_id, group_id, timeout \\ 3000)

@spec group_exists?(atom(), any(), non_neg_integer()) :: boolean() | {:error, any()}

group is existed

list_chains(sup_id, timeout \\ 3000)

@spec list_chains(atom(), non_neg_integer()) ::
  {:ok,
   [
     %{
       id: atom(),
       restart_strategy: atom(),
       send_type: atom(),
       queue_length: pos_integer(),
       worker_ids: [term()]
     }
   ]}
  | {:error, term()}

Lists the chains of a supervisor with their configuration and worker ids.

Examples

Sup.add_chain(:sup1, id: :chain1)
{:ok,
 [
   %{
     id: :chain1,
     restart_strategy: :one_for_one,
     send_type: :random,
     queue_length: 50,
     worker_ids: []
   }
 ]} = Sup.list_chains(:sup1)

list_groups(sup_id, timeout \\ 3000)

@spec list_groups(atom(), non_neg_integer()) ::
  {:ok, [%{id: atom(), restart_strategy: atom(), worker_ids: [term()]}]}
  | {:error, term()}

Lists the groups of a supervisor with their restart strategies and worker ids.

Examples

Sup.add_group(:sup1, id: :group1, restart_strategy: :one_for_all)
{:ok,
 [
   %{id: :group1, restart_strategy: :one_for_all, worker_ids: []}
 ]} = Sup.list_groups(:sup1)

list_standalone_workers(sup_id, timeout \\ 3000)

@spec list_standalone_workers(atom(), non_neg_integer()) ::
  {:ok, [%{id: term(), name: atom() | nil, restart_strategy: atom()}]}
  | {:error, term()}

Lists the standalone workers of a supervisor.

Examples

Sup.add_standalone_worker(:sup1, {MyGenServer, []}, id: :sa1)
{:ok, [%{id: :sa1, name: nil, restart_strategy: :transient}]} =
  Sup.list_standalone_workers(:sup1)

remove_chain(sup_id, chain_id, timeout \\ 3000)

@spec remove_chain(atom(), any(), non_neg_integer()) :: :ok | {:error, term()}

remove chain.

remove_chain_worker(sup_id, chain_id, worker_id, timeout \\ 3000)

@spec remove_chain_worker(atom(), any(), any(), non_neg_integer()) ::
  :ok | {:error, term()}

remove a worker from chain.

remove_group(sup_id, group_id, timeout \\ 3000)

@spec remove_group(atom(), any(), non_neg_integer()) :: :ok | {:error, any()}

remove group

remove_group_worker(sup_id, group_id, worker_id, timeout \\ 3000)

@spec remove_group_worker(atom(), any(), any(), non_neg_integer()) ::
  :ok | {:error, any()}

remove a worker out of group

remove_standalone_worker(sup_id, worker_id, timeout \\ 3000)

@spec remove_standalone_worker(atom(), any(), non_neg_integer()) ::
  :ok | {:error, term()}

Remove standalone worker from supervisor.

restart_group(sup_id, group_id, timeout \\ 3000)

@spec restart_group(atom(), any(), non_neg_integer()) ::
  {:ok, atom()} | {:error, any()}

Restart all workers in group

restart_group_worker(sup_id, group_id, worker_id, timeout \\ 3000)

@spec restart_group_worker(atom(), any(), any(), non_neg_integer()) ::
  {:ok, atom()} | {:error, any()}

Restart a worker in group

running?(id)

@spec running?(atom()) :: boolean()

Check if supervisor is running.

running_supervisors()

@spec running_supervisors() :: [{atom(), pid()}]

Lists all supervisors started through this library on the current node.

Entries whose process died are filtered out (and cleaned up lazily).

Examples

iex> Sup.start_with_config(id: :doctest_sup)
iex> Sup.running_supervisors() |> Enum.any?(fn {id, pid} -> id == :doctest_sup and is_pid(pid) end)
true

send_to_chain(sup_id, chain_id, data, timeout \\ 3000)

@spec send_to_chain(atom(), any(), any(), non_neg_integer()) ::
  {:ok, any()} | {:error, any()}

Send data to the entry worker in the chain. If chain doesn't has any worker, it will be dropped.

send_to_group_random(sup_id, group_id, data, timeout \\ 3000)

@spec send_to_group_random(atom(), any(), any(), non_neg_integer()) ::
  :ok | {:error, any()}

Send data to a random worker in the group.

send_to_group_worker(sup_id, group_id, worker_id, data, timeout \\ 3000)

@spec send_to_group_worker(atom(), any(), any(), any(), non_neg_integer()) ::
  :ok | {:error, any()}

Send data to a worker in the group.

send_to_my_group(worker_id, data)

@spec send_to_my_group(any(), any()) :: :ok | {:error, any()}

Send data to other worker in the same group.

send_to_my_group_random(data)

@spec send_to_my_group_random(any()) :: :ok | {:error, any()}

Send data to a random worker in the same group.

send_to_standalone_worker(sup_id, worker_id, data, timeout \\ 3000)

@spec send_to_standalone_worker(atom(), any(), any(), non_neg_integer()) ::
  :ok | {:error, term()}

Send data directly to the worker standalone in the supervisor.

start()

@spec start() :: {:ok, pid()} | {:error, any()}

Start supervisor run as independent process with default options.

start(options)

@spec start(t()) :: {:ok, pid()} | {:error, any()}

Start supervisor as independent process (no link process).

start_link()

@spec start_link() :: {:ok, pid()} | {:error, any()}

Work like start_link/1 with default options.

start_link(options)

@spec start_link(t()) :: {:ok, pid()} | {:error, any()}

start_link for using supervisor as child in other supervisor or link to current process.

start_with_config(config)

@spec start_with_config(
  id: atom(),
  link: boolean() | pid(),
  num_partitions: integer(),
  report_to: list()
) :: {:ok, pid()} | {:error, any()}

Start supervisor with configurations (Keyword). For run standalone, please set option :link to false. For link to other process (not current process), please set link to pid of that process. result format: {:ok, pid} or {:error, reason}

stop(sup_id, shutdown_type \\ :kill, timeout \\ 3000)

@spec stop(atom(), shutdown_type :: atom(), timeout :: non_neg_integer()) ::
  {:ok, atom()} | {:error, :not_running}

Stop supervisor.

Shutdown types:

  • :kill (default) worker processes are killed with Process.exit(pid, :kill).
  • any other type (e.g. :normal) is not implemented yet; it currently falls back to the same brutal kill, so workers cannot run cleanup code.

Returns {:ok, sup_id} when the stop signal was delivered (the supervisor shuts down asynchronously shortly after) or {:error, :not_running}.

supervisor_info(sup_id, timeout \\ 3000)

@spec supervisor_info(atom(), non_neg_integer()) :: {:ok, map()} | {:error, term()}

Returns an overview of a running supervisor: partition health and counts of groups, chains and workers.

Examples

{:ok,
 %{
   id: :my_sup,
   num_partitions: 4,
   partitions: [%{id: 1, pid: pid, alive?: true, message_queue_len: 0}, ...],
   num_groups: 2,
   num_chains: 1,
   num_standalone_workers: 3,
   total_worker_processes: 9
 }} = Sup.supervisor_info(:my_sup)