scheduler

class Scheduler(nodes: 'dict[str, NodeSpecification]', edges: 'list[Edge]', source_nodes: 'list[NodeID]' = <factory>, event_maker: 'EventMaker' = <factory>, _logger: 'logging.Logger' = <factory>)

nodes: dict[str, NodeSpecification]

edges: list[Edge]

source_nodes: list[Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)]]

event_maker: EventMaker

property epoch_log: set[int]

classmethod from_specification(nodes: dict[str, NodeSpecification], edges: list[Edge]) → Self

Create an instance of a Scheduler from NodeSpecification and Edge

property subepochs: dict[Epoch, set[Epoch]]

property epoch: Epoch

The current epoch that would run next/is running

iter_epoch(epoch: Epoch | None = None) → Iterator[list[MetaEvent]]

Iter batches of ready events from an epoch until it’s completed.

Parameters:

epoch (Epoch) – Epoch to iterate over. If None , adds an epoch to iterate over.

Raises:

EpochCompletedError if the requested epoch has already been completed –

iter_ready() → Iterator[list[MetaEvent]]

Iter batches of ready events from all epochs, infinitely, until no more nodes can be run in any epoch.

TODO: document stateful/stateless behavior, need for callers to handle breaks in iteration

add_epoch(epoch: Epoch | None = None) → Epoch

Add another epoch with a prepared graph to the scheduler.

add_subepoch(epoch: Epoch) → Epoch

Add subepoch!

Creates a topo sorter with all the nodes downstream of the node that created the epoch.

is_active(epoch: Epoch | None = None) → bool

Graph remains active while it holds at least one epoch that is active.

get_ready(epoch: Epoch | None = None) → list[MetaEvent]

Output the set of nodes that are ready across different epochs.

Parameters:

epoch (Epoch | None) – if an Epoch, get ready events for that epoch, if None , get ready events for all epochs.

node_is_ready(node: Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)], epoch: Epoch, subepochs: bool = False) → bool

Check if a single node is ready in a single or any epoch

Parameters:

• node (NodeID) – the node to check

• epoch (int | None) – the epoch to check

• subepochs (bool) – If True, return True if ready in any subepoch too

node_is_done(node: Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)], epoch: Epoch) → bool

Node is expired or done in specified epoch

sources_finished(epoch: Epoch) → bool

Check the source nodes of the given epoch have been processed. If epoch is None, check the source nodes of the latest epoch.

update(events: MutableSequence[Event | MetaEvent] | MutableSequence[Event]) → MutableSequence[Event] | MutableSequence[Event | MetaEvent]

When a set of events are received, update the graphs within the scheduler. Currently only has TopoSorter.done() implemented.

done(epoch: Epoch, node_id: str, signal: Annotated[str, AfterValidator(func=_is_identifier)] | None = None, with_signals: bool = True) → list[MetaEvent]

Mark a node in a given epoch as done.

Parameters:

with_signals (bool) – When marking this node as done, also mark all its signals as done.

expire(epoch: Epoch, node_id: str, signal: Annotated[str, AfterValidator(func=_is_identifier)] | None = None, with_signals: bool = True, unlock_optionals: bool = True) → list[MetaEvent]

Mark a node as having been completed without making its dependent nodes ready. i.e. when the node emitted NoEvent

epoch_completed(epoch: Epoch) → bool

Check if the epoch has been completed - as in it is fully completed and should no longer be acted upon. Distinct from is_active, which is used to check if an epoch should be marked as complete

end_epoch(epoch: Epoch) → list[MetaEvent]

enable_node(node_id: str) → None

Enable edges attached to the node and the NodeSpecification enable switches to True. Enabling a node clears any existing epochs, so it should only be done between process calls.

disable_node(node_id: str) → None

Disable edges attached to the node and the NodeSpecification enable switches to False

clear() → None

Remove epoch records, restarting the scheduler

has_cycle() → bool

Checks that the graph is acyclic.

generations() → list[list[Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)]]]

Get the topological generations of the graph: tuples for each set of nodes that can be run at the same time.

Order within a generation is not guaranteed to be stable.

asset_generations() → dict[Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)], list[tuple[str, ...]]]

generations() except only including nodes with direct dependencies on assets, to determine when the asset should be initialized vs. received in the ZMQ Runner.

Packed in a dictionary with the asset ID as the key, and the value as the generations for that asset.

upstream_nodes(node: Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)]) → set[Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)]]

All the nodes that have an effect on the given node

From: * Dependencies * If the node has optional dependencies, nodes whose NoEvents it should listen to

get_epoch_state(epoch: Epoch) → SorterState