`mod scheduler`¶

module scheduler¶

Functions

fn downstream_nodes<'a>(edges: &'a [EdgeRec], node: &'a str, exclude: &FxIndexSet<&str>) -> FxIndexSet<&'a str>¶: Compute all the nodes downstream (that depend on) a given node.

Structs and Unions

struct EpochIter<'a>¶

Iterator for ready events within an epoch and its subepochs. Since the scheduler itself can’t be mutably borrowed across iterations, creates a proxy iterator object with methods for updating the scheduler.

Note that since iterating an epoch also iterates events from its subepochs, the epoch returned during iteration must be used when fetching the relevant inputs to run a node, and used when calling the done and expire methods - don’t just recycle the epoch passed when starting iteration.

Examples

# use noob_core::exceptions::CoreError;
# use noob_core::scheduler::Scheduler;
# use noob_core::FxIndexMap;
# use noob_core::sorter::EdgeRec;
let mut scheduler = Scheduler::from_graph(
    FxIndexMap::default(),
    vec![
        EdgeRec::from(("a", "a1", "b")),
        EdgeRec::from(("b", "b1", "c")),
    ]
)?;

let ep = scheduler.add_epoch();
let mut batches = scheduler.iter_epoch_at(ep)?;
while let Some(ready) = batches.next() {
    for (epoch, node) in ready {
        // in reality, actually do something
        // events = call_the_node()
        batches.done(&epoch, node, true)?;
    }
}
# Ok::<(), CoreError>(())

Implementations

impl EpochIter<'_>¶

Functions

fn done(&mut self, epoch: &Epoch, item: ItemID, with_signals: bool) -> CoreResult<Vec<Epoch>>¶: Call Scheduler.done

fn expire(&mut self, epoch: &Epoch, item: ItemID, with_signals: bool, unlock_optionals: bool) -> CoreResult<Vec<Epoch>>¶: Call Scheduler.expire

Traits implemented

impl Iterator for EpochIter<'_>¶

struct ReadyIter<'a>¶

Iterator over all nodes in any epoch that are ready. Ends iteration when no remaining nodes are available in any epoch, but since usually between calls to next the graph is advanced, callers have to implement breaking iteration themselves.

Implementations

impl ReadyIter<'_>¶

Functions

fn done(&mut self, epoch: &Epoch, item: ItemID, with_signals: bool) -> CoreResult<Vec<Epoch>>¶: Call Scheduler.done

fn expire(&mut self, epoch: &Epoch, item: ItemID, with_signals: bool, unlock_optionals: bool) -> CoreResult<Vec<Epoch>>¶: Call Scheduler.expire

Traits implemented

impl Iterator for ReadyIter<'_>¶

struct Scheduler¶

The thing that says what nodes run, when. Coordinates a set of Sorters, keyed by noob_core::epoch::Epochs.

Typical use is to receive one from an instantiated (python) Tube, and then to use one of its two iteration modes in a loop with the update method.

In general, for actions that can either take place in a specific epoch or any epoch, the naming convention is something like

{do_something} - in any epoch
{do_something}_at - in a specific, given epoch.

the _at form is usually fallible since it’s possible that the epoch has already been completed or is otherwise invalid for the operation, and the “any epoch” form is infallible and adds new epochs if necessary.

Examples

# use noob_core::exceptions::CoreError;
# use noob_core::scheduler::Scheduler;
# use noob_core::FxIndexMap;
# use noob_core::sorter::EdgeRec;
let mut scheduler = Scheduler::from_graph(
    FxIndexMap::default(),
    vec![
        EdgeRec::from(("a", "a1", "b")),
        EdgeRec::from(("b", "b1", "c")),
    ]
)?;

let ep = scheduler.add_epoch();
let mut batches = scheduler.iter_epoch_at(ep)?;
while let Some(ready) = batches.next() {
    for (epoch, node) in ready {
        // in reality, call the node somehow and have events
        // events = node_calling_thing()
        // scheduler.update(events)
        // for docs, just trivially mark done
        batches.done(&epoch, node, true)?;
    }
}
# Ok::<(), CoreError>(())

graph_items: FxHashSet<ItemID>¶: All the nodes and signals that we care about for updates - i.e. all those that are present in our template sorter

Implementations

impl Scheduler¶

Functions

fn add_epoch(&mut self) -> Epoch¶: Add the next epoch, after the last highest created root epoch

fn add_epoch_at(&mut self, epoch: impl Into<Epoch>) -> CoreResult<Epoch>¶: Add a specific epoch, can fail if the epoch has already been created or completed If epoch is not a root epoch, dispatches to Scheduler.add_subepoch

fn add_subepoch(&mut self, epoch: impl Into<Epoch>) -> CoreResult<Epoch>¶: Add a subepoch consisting of a subgraph of the nodes that are downstream from the node that induced the epoch (the node id in the Epoch.leaf).

The state of the graph items in the parent epoch is transferred to any nodes that also exist in the subepoch, and the node that induced the subepoch is expired in the parent.

fn done(&mut self, epoch: &Epoch, item: ItemID, with_signals: bool) -> CoreResult<Vec<Epoch>>¶: Mark a node or signal as having been run to completion in a given epoch. If with_signals is passed and item is a node, and signals belonging to the node will also be marked done.

Special care has to be taken when handling subepochs - creating subepochs by marking a node as done in a subep does not necessarily handle expiring the signals of the node in the parent epoch. Handle the parent epoch state explicitly. (The preferred interface for downstream use is to update the scheduler from events, rather than directly marking nodes done)

fn end_epoch(&mut self, epoch: impl Into<Epoch>) -> CoreResult<Vec<Epoch>>¶: Declare that an epoch has been completed, performing bookkeeping, cleanup, and emitting the epoch completion meta event in python. Only root epochs cause epoch sorters to be dropped - a parent epoch is not completed until all of its subepochs are, and the completion of a parent epoch means that all its subepochs are completed.

fn epoch_completed(&self, epoch: &Epoch) -> bool¶: Whether the given epoch and all of its subepochs have been completed.

fn epoch_log(&self) -> BTreeSet<u32>¶: Clone of the internal log of epochs used to track completed epochs, since epochs can be completed out of order.

fn expire(&mut self, epoch: &Epoch, item: ItemID, with_signals: bool, unlock_optionals: bool) -> CoreResult<Vec<Epoch>>¶: Mark a node as having been completed without any result, without making its successors ready, effectively canceling the tree of obligate nodes beneath it.

fn first_active_epoch(&self) -> Option<Epoch>¶: The lowest root epoch that is active, directly or via subepochs - python’s iter_epoch no-arg resolution (scheduler.py:111-120)

fn from_graph(nodes: FxIndexMap<String, NodeFlags>, edges: Vec<EdgeRec>) -> CoreResult<Scheduler>¶

fn generations(&self) -> Vec<Vec<ItemID>>¶: Topological generations of the sorter: sets of graph items that would be yielded if all of the items in the preceding generation were marked done.

fn get_epoch_state(&self, epoch: &Epoch) -> CoreResult<SorterState>¶: Get the state of a sorter for a given epoch Extremely expensive, clones everything only to be used when debugging

fn get_ready(&mut self) -> Vec<(Epoch, ItemID)>¶: Get nodes that are ready in any epoch This and get_ready_at are not preferred calling patterns, as they require a continual check of epoch completion rather than just iterating, which also naturally closes. They are public only within the crate to expose them to python, but prefer the iterators.

fn get_ready_at(&mut self, epoch: &Epoch) -> Vec<(Epoch, ItemID)>¶: Get nodes that are ready in a specific epoch. See the note in Scheduler.get_ready

fn has_cycle(&self) -> bool¶

fn is_active(&self) -> bool¶: Is the scheduler active in any epoch?

fn is_active_at(&self, epoch: &Epoch) -> bool¶: Is the scheduler active in a specific epoch?

fn iter_epoch(&mut self) -> EpochIter<'_>¶: Create an iterator for ready events in the active epoch and its subepochs

fn iter_epoch_at(&mut self, epoch: impl Into<Epoch>) -> CoreResult<EpochIter<'_>>¶: Create an iterator for ready events in the given epoch and its subepochs, creating it if it doesn’t exist.

fn iter_ready(&mut self) -> ReadyIter<'_>¶: Create and iterator that iterates over all ready events until there are no more left, in any epoch.

fn node_is_done(&self, node: &ItemID, epoch: &Epoch) -> bool¶: Whether the node is done (expired or ran) in an epoch and all of its subepochs

fn node_is_ready(&self, node: &ItemID, epoch: &Epoch, subepochs: bool) -> bool¶: Check if a node is ready without marking it as out

fn source_nodes(&self) -> FxIndexSet<ItemID>¶: Any nodes that have no dependencies within the graph.

fn sources_finished(&self, epoch: &Epoch) -> bool¶: Whether all nodes that have no dependencies, or “source” nodes, have been finished (either expired or done) in a given epoch. When all the sources are completed in a (root) epoch, then the successor epoch is created, as that would be the earliest that any nodes within it could run, if stateless.

fn subepochs(&self) -> FxHashMap<Epoch, FxIndexSet<Epoch>>¶: Clone of the internal mapping between an epoch and any subepochs it has (at any depth)

fn update(&mut self, mut events: Vec<UpdateEvent>) -> CoreResult<Vec<Epoch>>¶

After a node(s) emit a set of events, update the state of the scheduler, returning any epochs that were completed by the updates.

This is the primary way that the scheduler should consume the results of processing: in a loop:

get available nodes
get their inputs from previously stored events
run them
store their values -> creating events
update the scheduler with events.

fn upstream_nodes(&self, node: ItemID) -> CoreResult<FxHashSet<ItemID>>¶: The collection of nodes and signals that influence a node’s readiness: both direct predecessors, and any predecessors that might reach us through a chain of optional dependencies.

mod scheduler¶

`mod scheduler`¶