Packages
reactor
0.15.6
1.0.2
1.0.1
1.0.0
0.17.0
0.16.0
0.15.6
0.15.5
0.15.4
0.15.3
0.15.2
0.15.1
0.15.0
0.14.0
0.13.3
0.13.2
0.13.1
0.13.0
0.12.1
0.12.0
0.11.0
0.10.3
0.10.2
0.10.1
0.10.0
0.9.1
0.9.0
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.0
0.6.0
0.5.2
0.5.1
0.5.0
0.4.1
0.4.0
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.3.0
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.0
An asynchronous, graph-based execution engine
Current section
Files
Jump to
Current section
Files
lib/reactor/executor/state.ex
defmodule Reactor.Executor.State do
@moduledoc """
Contains the reactor execution state.
This is run-time only information.
"""
@defaults %{
async?: true,
halt_timeout: 5000,
max_iterations: :infinity,
timeout: :infinity
}
defstruct async?: @defaults.async?,
concurrency_key: nil,
current_tasks: %{},
errors: [],
halt_timeout: @defaults.halt_timeout,
max_concurrency: nil,
max_iterations: @defaults.max_iterations,
pool_owner: false,
retries: %{},
run_id: nil,
skipped: MapSet.new(),
started_at: nil,
timeout: @defaults.timeout
alias Reactor.{Executor.ConcurrencyTracker, Step}
@type t :: %__MODULE__{
async?: boolean,
concurrency_key: ConcurrencyTracker.pool_key(),
current_tasks: %{Task.t() => Step.t()},
errors: [any],
halt_timeout: pos_integer() | :infinity,
max_concurrency: pos_integer(),
max_iterations: pos_integer() | :infinity,
pool_owner: boolean,
retries: %{reference() => pos_integer()},
run_id: any,
skipped: MapSet.t(),
started_at: DateTime.t(),
timeout: pos_integer() | :infinity
}
@doc false
@spec init(map) :: t
def init(attrs \\ %{}) do
@defaults
|> Map.merge(attrs)
|> do_init()
end
defp do_init(attrs) do
attrs
|> maybe_set_max_concurrency()
|> maybe_allocate_concurrency_pool()
|> maybe_set_run_id()
|> Map.put(:started_at, DateTime.utc_now())
|> then(&struct!(__MODULE__, &1))
end
defp maybe_set_max_concurrency(attrs)
when is_integer(attrs.max_concurrency) and attrs.max_concurrency > 0,
do: attrs
defp maybe_set_max_concurrency(attrs) when attrs.async? == false,
do: Map.put(attrs, :max_concurrency, 0)
defp maybe_set_max_concurrency(attrs),
do: Map.put(attrs, :max_concurrency, System.schedulers_online())
defp maybe_allocate_concurrency_pool(attrs) when is_reference(attrs.concurrency_key) do
attrs
|> Map.put(:pool_owner, false)
end
defp maybe_allocate_concurrency_pool(attrs) do
attrs
|> Map.put(:concurrency_key, ConcurrencyTracker.allocate_pool(attrs.max_concurrency))
|> Map.put(:pool_owner, true)
end
defp maybe_set_run_id(attrs) do
attrs
|> Map.update(:run_id, make_ref(), fn
nil -> make_ref()
ref -> ref
end)
end
end