ExResilience.Bulkhead (ex_resilience v0.4.0)

Copy Markdown View Source

Concurrency-limiting bulkhead implemented as a GenServer.

Limits the number of concurrent executions of a function. Callers beyond the limit are queued and served in order. If the queue wait exceeds max_wait, the caller receives {:error, :bulkhead_full}.

Uses :atomics for the permit counter (lock-free CAS) and the GenServer for queue management.

Options

  • :name -- required. Registered name for this bulkhead instance.
  • :max_concurrent -- maximum concurrent executions. Default 10.
  • :max_wait -- maximum time in ms a caller waits in the queue. Default 5_000. Set to 0 to reject immediately when full.

Examples

iex> {:ok, _} = ExResilience.Bulkhead.start_link(name: :test_bh, max_concurrent: 2)
iex> ExResilience.Bulkhead.call(:test_bh, fn -> :done end)
{:ok, :done}

Summary

Types

option()
result()

Functions

active_count(name)

Returns the current number of active (in-use) permits.

call(name, fun, timeout \\ :infinity)

Executes fun within the bulkhead's concurrency limit.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

queue_length(name)

Returns the current queue length.

start_link(opts)

Starts a bulkhead process.

Types

option()

@type option() ::
  {:name, atom()}
  | {:max_concurrent, pos_integer()}
  | {:max_wait, non_neg_integer()}

result()

@type result() :: {:ok, term()} | {:error, :bulkhead_full} | {:error, term()}

Functions

active_count(name)

@spec active_count(atom()) :: non_neg_integer()

Returns the current number of active (in-use) permits.

call(name, fun, timeout \\ :infinity)

@spec call(atom(), (-> term()), non_neg_integer() | :infinity) :: result()

Executes fun within the bulkhead's concurrency limit.

Returns {:ok, result} on success, {:error, :bulkhead_full} if the bulkhead is at capacity and the wait queue times out, or {:error, reason} if the function returns an error tuple.

Raises if the function raises.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

queue_length(name)

@spec queue_length(atom()) :: non_neg_integer()

Returns the current queue length.

start_link(opts)

@spec start_link([option()]) :: GenServer.on_start()

Starts a bulkhead process.

See module docs for available options.