View Source Reactor.Executor.Async (reactor v0.10.1)

Handle the asynchronous execution of a batch of steps, along with any mutations to the reactor or execution state.

Summary

Functions

collect_remaining_tasks_for_shutdown(reactor, state)

When the Reactor needs to shut down for any reason, we need to await all the currently running asynchronous steps and delete any task vertices.

handle_completed_steps(reactor, state)

Handle zero or one completed async steps and then decide what to do.

start_steps(reactor, state, steps, supervisor \\ {:via, PartitionSupervisor, {Reactor.TaskSupervisor, self()}})

Start as many of the provided steps as possible.

Functions

collect_remaining_tasks_for_shutdown(reactor, state)

@spec collect_remaining_tasks_for_shutdown(Reactor.t(), Reactor.Executor.State.t()) ::
  {Reactor.t(), Reactor.Executor.State.t()}

When the Reactor needs to shut down for any reason, we need to await all the currently running asynchronous steps and delete any task vertices.

handle_completed_steps(reactor, state)

@spec handle_completed_steps(Reactor.t(), Reactor.Executor.State.t()) ::
  {:recurse | :continue | :undo | :halt, Reactor.t(),
   Reactor.Executor.State.t()}

Handle zero or one completed async steps and then decide what to do.

start_steps(reactor, state, steps, supervisor \\ {:via, PartitionSupervisor, {Reactor.TaskSupervisor, self()}})

@spec start_steps(
  Reactor.t(),
  Reactor.Executor.State.t(),
  [Reactor.Step.t()],
  Supervisor.supervisor()
) ::
  {:continue | :recurse, Reactor.t(), Reactor.Executor.State.t()}
  | {:error, any()}

Start as many of the provided steps as possible.

Takes into account he maximum concurrency and available work slots.