Orchid.Scheduler (Orchid v0.6.1)

Scheduler is responsible for managing and scheduling the execution order of steps in the Recipe.

Summary

Types

initial_params()

Initial parameters can be a list of Params or a map of Params.

Functions

build(recipe, initial_params, workflow_context)

Initialize scheduler context.

done?(context)

get_results(context)

get_results(context, key)

inject_opts(ctx, selector, new_opts)

Batch update configuration (at runtime).

mark_running_steps(ctx, step_indices, mode \\ :running)

Mark those steps that have started running(or remove when failed).

merge_result(ctx, step_idx, output_params)

After the Step is executed, merge the result back into the Context.

next_ready_steps(ctx)

Core scheduling function: Identify all the steps that are "inputs params ready" and "not executed".

Types

initial_params()

@type initial_params() :: [Orchid.Param.t()] | Orchid.Scheduler.Context.param_map()

Initial parameters can be a list of Params or a map of Params.

Functions

build(recipe, initial_params, workflow_context)

@spec build(
  Orchid.Recipe.t() | [Orchid.Step.t()],
  initial_params(),
  Orchid.WorkflowCtx.t()
) ::
  {:ok, Orchid.Scheduler.Context.t()} | {:error, term()}

Initialize scheduler context.

done?(context)

@spec done?(Orchid.Scheduler.Context.t()) :: boolean()

get_results(context)

@spec get_results(Orchid.Scheduler.Context.t()) ::
  Orchid.Scheduler.Context.param_map()

get_results(context, key)

inject_opts(ctx, selector, new_opts)

@spec inject_opts(
  Orchid.Scheduler.Context.t(),
  (Orchid.Step.t() -> boolean()),
  keyword()
) ::
  Orchid.Scheduler.Context.t()

Batch update configuration (at runtime).

This is used in scenarios where after an external service crashes and re-assigned, but several steps' options still use the old references.

mark_running_steps(ctx, step_indices, mode \\ :running)

@spec mark_running_steps(
  Orchid.Scheduler.Context.t(),
  Orchid.Scheduler.Context.step_index()
  | [Orchid.Scheduler.Context.step_index()],
  mode :: :running | :reattempt
) :: Orchid.Scheduler.Context.t()

Mark those steps that have started running(or remove when failed).

merge_result(ctx, step_idx, output_params)

@spec merge_result(
  Orchid.Scheduler.Context.t(),
  non_neg_integer(),
  [Orchid.Param.t()] | Orchid.Param.t()
) :: Orchid.Scheduler.Context.t()

After the Step is executed, merge the result back into the Context.

next_ready_steps(ctx)

@spec next_ready_steps(Orchid.Scheduler.Context.t()) :: [
  {Orchid.Step.t(), Orchid.Scheduler.Context.step_index()}
]

Core scheduling function: Identify all the steps that are "inputs params ready" and "not executed".