Scheduler

Note

This documentation is mirrored from the graph-scheduler package and often refers to nodes, edges , and graphs. In PsyNeuLink terms, nodes are Mechanisms or Compositions, edges are Projections, and graphs are Compositions. The one exception is during learning, in which Projections may be assigned for execution as nodes to ensure that MappingProjections are updated in the proper order.

Note

This documentation was modified from the original due to environment-specific TimeScale renamings. If there is any confusion, please see the original documentation at https://www.github.com/kmantel/graph-scheduler

Overview

A Scheduler is used to generate the order in which the nodes of a graph are executed. By default, a Scheduler executes nodes in an order determined by the pattern of edges among the nodes in the graph, with each node executed once per PASS through the graph. For example, in a graph in which a node A projects to a node B that projects to a node C, A will execute first followed by B, and then C in each PASS through the graph. However, a Scheduler can be used to implement more complex patterns of execution, by specifying Conditions that determine when and how many times individual nodes execute, and whether and how this depends on the execution of other nodes. Any executable node in a graph can be assigned a Condition, and Conditions can be combined in arbitrary ways to generate any pattern of execution of the nodes in a graph that is logically possible.

Creating a Scheduler

A Scheduler can be created explicitly using its constructor. However, more commonly it is created automatically for a Composition when it is created.When creating a Scheduler explicitly, the set of nodes to be executed and their order must be specified in the Scheduler’s constructor using one the following:

• a Composition in the composition argument - if a Composition is specified, the Scheduler is created using the nodes and edges in the Composition’s graph, with any feedback Projections pruned as needed to produce an acyclic graph. If there is a cycle comprised of all non-feedback projections, this cycle is reduced to a single consideration set

• a graph specification dictionary in the graph argument - each entry of the dictionary must be a node of a graph, and the value of each entry must be a set of zero or more nodes that project directly to the key. The graph must be acyclic; an error is generated if any cycles (e.g., recurrent dependencies) are detected. The Scheduler computes a toposort from the graph that is used as the default order of executions, subject to any Conditions that have been specified (see below).

Conditions can be added to a Scheduler when it is created by specifying a ConditionSet (a set of Conditions) in the conditions argument of its constructor. Individual Conditions and/or ConditionSets can also be added after the Scheduler has been created, using its add_condition and add_condition_set methods, respectively.

Graph structure Conditions are applied to the Scheduler’s graph in the order in which they are added.

Algorithm

When a Scheduler is created, it constructs a consideration_queue: a list of consideration sets that defines the order in which nodes are eligible to be executed. This is determined by the topological ordering of the graph provided to the Scheduler's constructor, which is then modified by any graph structure conditions that are added to the Scheduler. Each consideration_set is a set of nodes that are eligible to execute at the same time/TIME_STEP (i.e., that appear at the same “depth” in a sequence of dependencies, and among which there are no dependencies). The first consideration_set consists of only ORIGIN nodes. The second consists of all nodes that receive edges from the nodes in the first consideration_set. The third consists of nodes that receive edges from nodes in the first two consideration sets, and so forth. When the Scheduler is run, it uses the consideration_queue to determine which nodes are eligible to execute in each TIME_STEP of a PASS, and then evaluates the Condition associated with each node in the current consideration_set to determine which should actually be assigned for execution.

Pseudocode:

consideration_queue <- list(toposort(graph))

reset TimeScale.TRIAL counters
while TimeScale.TRIAL are not satisfied
and TimeScale.RUN termination conditions are not satisfied:
    reset TimeScale.PASS counters
    cur_index <- 0

    while TimeScale.TRIAL termination conditions are not satisfied
    and TimeScale.RUN termination conditions are not satisfied
    and cur_index < len(consideration_queue):

        cur_consideration_set <- consideration_queue[cur_index]
        do:
            cur_consideration_set_has_changed <- False
            for cur_node in cur_consideration_set:
                if  cur_node not in cur_consideration_set_execution
                    and cur_node`s Condition is satisfied:

                    cur_consideration_set_has_changed <- True
                    add cur_node to cur_consideration_set_execution
                    increment execution and time counters
        while cur_consideration_set_has_changed

        if cur_consideration_set_execution is not empty or absolute time conditions are used:
            yield cur_consideration_set_execution

        increment cur_index
        increment time counters

    if all execution sets yielded were empty:
        yield an empty execution set

Execution

Note

This section covers normal scheduler execution (Scheduler.mode = SchedulingMode.STANDARD). See Exact Time Mode below for a description of exact time mode.

When a Scheduler is run, it provides a set of nodes that should be run next, based on their dependencies in the graph specification, and any Conditions, specified in the Scheduler’s constructor. For each call to the run method, the Scheduler sequentially evaluates its consideration sets in their order in the consideration_queue. For each set, it determines which nodes in the set are allowed to execute, based on whether their associated Condition has been met. Any node that does not have a Condition explicitly specified is assigned a Condition that causes it to be executed whenever it is under consideration and all its structural parents have been executed at least once since the node’s last execution. All of the nodes within a consideration_set that are allowed to execute comprise a TIME_STEP of execution. These nodes are considered as executing simultaneously.

Note

The ordering of the nodes specified within a TIME_STEP is arbitrary (and is irrelevant, as there are no graph dependencies among nodes within the same consideration_set). However, the execution of a node within a TIME_STEP may trigger the execution of another node within its consideration_set, as in the example below:

   C
  ↗ ↖
 A   B

scheduler.add_condition(B, graph_scheduler.EveryNCalls(A, 2))
scheduler.add_condition(C, graph_scheduler.EveryNCalls(B, 1))

execution sets: [{A}, {A, B}, {C}, ...]

Since there are no graph dependencies between A and B, they may execute in the same TIME_STEP. Morever, A and B are in the same consideration_set. Since B is specified to run every two times A runs, A’s second execution in the second TIME_STEP allows B to run within that TIME_STEP, rather than waiting for the next PASS.

For each TIME_STEP, the Scheduler evaluates whether any specified termination Conditions have been met, and terminates if so. Otherwise, it returns the set of nodes that should be executed in the current TIME_STEP. Each subsequent call to the run method returns the set of nodes in the following TIME_STEP.

Processing of all of the consideration_sets in the consideration_queue constitutes a PASS of execution, over which every node in the graph has been considered for execution. Subsequent calls to the run method cycle back through the consideration_queue, evaluating the consideration_sets in the same order as previously. Different subsets of nodes within the same consideration_set may be assigned to execute on each PASS, since different Conditions may be satisfied.

The Scheduler continues to make PASSes through the consideration_queue until a termination Condition is satisfied. If no termination Conditions are specified, by default the Scheduler terminates an TRIAL when every node has been specified for execution at least once (corresponding to the AllHaveRun Condition). However, other termination Conditions can be specified, that may cause the Scheduler to terminate an TRIAL earlier or later (e.g., when the Condition for a particular node or set of nodes is met). When the Scheduler terminates a TRIAL, the Composition begins processing the next input specified in the call to its run method. Thus, a TRIAL is defined as the scope of processing associated with a given input to the Composition.

Termination Conditions

Termination conditions are basic Conditions that specify when the open-ended units of time - TRIAL and RUN - have ended. By default, the termination condition for an TRIAL is AllHaveRun, which is satisfied when all nodes have run at least once within the environment state update, and the termination condition for an RUN is when all of its constituent environment state updates have terminated. These defaults may be overriden when running a Composition, by passing a dictionary mapping TimeScales to Conditions in the termination_processing argument of a call to Composition.run (to terminate the execution of processing):

Composition.run(
    ...,
    termination_processing={TimeScale.TRIAL: WhenFinished(ddm)}
    )

Graph structure conditions cannot be used as termination conditions.

Absolute Time

The scheduler supports scheduling of models of real-time systems in modes, both of which involve mapping real-time values to Time. The default mode is is most compatible with standard scheduling, but can cause some unexpected behavior in certain cases because it is inexact. The consideration queue remains intact, but as a result, actions specified by fixed times of absolute-time-based conditions (start and end of TimeInterval, and t of TimeTermination) may not occur at exactly the time specified. The simplest example of this situation involves a linear composition with two nodes:

>>> import psyneulink as pnl

>>> A = pnl.TransferMechanism()
>>> B = pnl.TransferMechanism()

>>> comp = pnl.Composition()
>>> pway = comp.add_linear_processing_pathway([A, B])

>>> comp.scheduler.add_condition(A, pnl.TimeInterval(start=10))
>>> comp.scheduler.add_condition(B, pnl.TimeInterval(start=10))

In standard mode, A and B are in different consideration sets, and so can never execute at the same time. At most one of A and B will start exactly at t=10ms, with the other starting at its next consideration after. There are many of these examples, and while it may be solveable in some cases, it is not a simple problem. So, Exact Time Mode exists as an alternative option for these cases, though it comes with its own drawbacks.

Note

Due to issues with floating-point precision, absolute time values in conditions and Time are limited to 8 decimal points. If more precision is needed, use fractions, where possible, or smaller units (e.g. microseconds instead of milliseconds).

Exact Time Mode

When Scheduler.mode is SchedulingMode.EXACT_TIME, the scheduler is capable of handling examples like the one above. In this mode, all nodes in the scheduler’s graph become members of the same consideration set, and may be executed at the same time for every consideration set execution, subject to the conditions specified. As a result, the guarantees in standard scheduling may not apply - that is, that all parent nodes get a chance to execute before their children, and that there exist no data dependencies (edges) between nodes in the same execution set. In exact time mode, all nodes will be in one [unordered] execution set. An ordering may be inferred by the original graph, however, using the indices in the original consideration queue. Compositions will execute nodes in this order, however independent usages of the scheduler may not. The user should be aware of this and set up defaults and inputs to nodes accordingly. Additionally, non-absolute conditions like EveryNCalls may behave unexpectedly in some cases.

Examples

Please see Condition for a list of all supported Conditions and their behavior.

• Basic phasing in a linear process:

  >>> import psyneulink as pnl
  
  >>> A = pnl.TransferMechanism(name='A')
  >>> B = pnl.TransferMechanism(name='B')
  >>> C = pnl.TransferMechanism(name='C')
  
  >>> comp = pnl.Composition()
  
  >>> pway = comp.add_linear_processing_pathway([A, B, C])
  >>> pway.pathway
  [(TransferMechanism A), (MappingProjection MappingProjection from A[RESULT] to B[InputPort-0]), (TransferMechanism B), (MappingProjection MappingProjection from B[RESULT] to C[InputPort-0]), (TransferMechanism C)]
  
  >>> # implicit condition of Always for A
  >>> comp.scheduler.add_condition(B, pnl.EveryNCalls(A, 2))
  >>> comp.scheduler.add_condition(C, pnl.EveryNCalls(B, 3))
  
  >>> # implicit AllHaveRun Termination condition
  >>> execution_sequence = list(comp.scheduler.run())
  >>> execution_sequence
  [{(TransferMechanism A)}, {(TransferMechanism A)}, {(TransferMechanism B)}, {(TransferMechanism A)}, {(TransferMechanism A)}, {(TransferMechanism B)}, {(TransferMechanism A)}, {(TransferMechanism A)}, {(TransferMechanism B)}, {(TransferMechanism C)}]
  

• Alternate basic phasing in a linear process:

  >>> comp = pnl.Composition()
  >>> pway = comp.add_linear_processing_pathway([A, B])
  >>> pway.pathway
  [(TransferMechanism A), (MappingProjection MappingProjection from A[RESULT] to B[InputPort-0]), (TransferMechanism B)]
  
  >>> comp.scheduler.add_condition(
  ...     A,
  ...     pnl.Any(
  ...         pnl.AtPass(0),
  ...         pnl.EveryNCalls(B, 2)
  ...     )
  ... )
  
  >>> comp.scheduler.add_condition(
  ...     B,
  ...     pnl.Any(
  ...         pnl.EveryNCalls(A, 1),
  ...         pnl.EveryNCalls(B, 1)
  ...     )
  ... )
  >>> termination_conds = {
  ...     pnl.TimeScale.TRIAL: pnl.AfterNCalls(B, 4, time_scale=pnl.TimeScale.TRIAL)
  ... }
  >>> execution_sequence = list(comp.scheduler.run(termination_conds=termination_conds))
  >>> execution_sequence 
  [{(TransferMechanism A)}, {(TransferMechanism B)}, {(TransferMechanism B)}, {(TransferMechanism A)}, {(TransferMechanism B)}, {(TransferMechanism B)}]
  

• Basic phasing in two processes:

  >>> comp = pnl.Composition()
  >>> pway = comp.add_linear_processing_pathway([A, C])
  >>> pway.pathway
  [(TransferMechanism A), (MappingProjection MappingProjection from A[RESULT] to C[InputPort-0]), (TransferMechanism C)]
  
  >>> pway = comp.add_linear_processing_pathway([B, C])
  >>> pway.pathway
  [(TransferMechanism B), (MappingProjection MappingProjection from B[RESULT] to C[InputPort-0]), (TransferMechanism C)]
  
  >>> comp.scheduler.add_condition(A, pnl.EveryNPasses(1))
  >>> comp.scheduler.add_condition(B, pnl.EveryNCalls(A, 2))
  >>> comp.scheduler.add_condition(
  ...     C,
  ...     pnl.Any(
  ...         pnl.AfterNCalls(A, 3),
  ...         pnl.AfterNCalls(B, 3)
  ...     )
  ... )
  >>> termination_conds = {
  ...     pnl.TimeScale.TRIAL: pnl.AfterNCalls(C, 4, time_scale=pnl.TimeScale.TRIAL)
  ... }
  >>> execution_sequence = list(comp.scheduler.run(termination_conds=termination_conds))
  >>> execution_sequence  
  [{(TransferMechanism A)}, {(TransferMechanism A), (TransferMechanism B)}, {(TransferMechanism A)}, {(TransferMechanism C)}, {(TransferMechanism A), (TransferMechanism B)}, {(TransferMechanism C)}, {(TransferMechanism A)}, {(TransferMechanism C)}, {(TransferMechanism A), (TransferMechanism B)}, {(TransferMechanism C)}]
  

Class Reference

class psyneulink.core.scheduling.scheduler.Scheduler(composition=None, graph=None, conditions=None, termination_conds=None, default_execution_id=None, mode=SchedulingMode.STANDARD, default_absolute_time_unit=<Quantity(1, 'millisecond')>, **kwargs)

Generates an order of execution for nodes in a graph or graph specification dictionary, possibly determined by a set of Conditions.

Parameters

• composition (Composition) – specifies the Components to be ordered for execution, and any dependencies among them, based on the Composition's graph.

• graph (Dict[object: set(object)], networkx.DiGraph) –

  a directed acyclic graph (DAG). Specified as either:

  • a graph specification dictionary: each entry of the

  dictionary must be an object, and the value of each entry must be a set of zero or more objects that project directly to the key. - a networkx.DiGraph

• conditions (ConditionSet) – set of Conditions that specify when individual nodes in graph execute and any dependencies among them.

• mode (SchedulingMode[STANDARD|EXACT_TIME] : SchedulingMode.STANDARD) – sets the mode of scheduling: standard or exact time

• default_absolute_time_unit (pint.Quantity : 1ms) – if not otherwise determined by any absolute conditions, specifies the absolute duration of a TIME_STEP

conditions

the set of Conditions the Scheduler uses when running

Type

ConditionSet

default_execution_id

the execution_id to use if none is supplied to run; a unique identifier to allow multiple schedulings independently. Must be hashable.

Type

object

execution_list

the full history of consideration set executions the Scheduler has produced

Type

list

consideration_queue

a list form of the Scheduler’s toposort ordering of its nodes

Type

list

consideration_queue_indices

a dict mapping nodes to their position in the consideration_queue

Type

dict

termination_conds

a mapping from TimeScales to Conditions that, when met, terminate the execution of the specified TimeScale. On set, update only for the TimeScales specified in the argument.

Type

Dict[TimeScale: Condition]

mode

sets the mode of scheduling: standard or exact time

Type

SchedulingMode

Default

SchedulingMode.STANDARD

default_absolute_time_unit

if not otherwise determined by any absolute conditions, specifies the absolute duration of a TIME_STEP

Type

pint.Quantity

Default

1ms

add_condition(owner, condition)

Adds a basic or graph structure Condition to the Scheduler.

If condition is basic, it will overwrite the current basic Condition for owner, if present. If you want to add multiple basic Conditions to a single owner, instead add a single Composite Condition to accurately specify the desired behavior.

If condition is structural, it will be applied on top of Scheduler.graph in the order it is added.

Parameters

• owner (node) – specifies the node with which the condition should be associated. condition will govern the execution behavior of owner

• condition (ConditionBase) – specifies the Condition, associated with the owner to be added to the ConditionSet.

add_condition_set(conditions)

Adds a set of basic or graph structure Conditions (in the form of a dict or another ConditionSet) to the Scheduler.

Any basic Condition added here will overwrite the current basic Condition for a given owner, if present. If you want to add multiple basic Conditions to a single owner, instead add a single Composite Condition to accurately specify the desired behavior.

Any structural Condition added here will be applied on top of Scheduler.graph in the order they are returned by iteration over conditions.

Parameters

conditions (dict[node: Condition], ConditionSet) –

specifies collection of Conditions to be added to this ConditionSet,

if a dict is provided:

each entry should map an owner node (the node whose execution behavior will be governed) to a Condition

remove_condition(owner_or_condition)

Removes the condition specified as or owned by owner_or_condition.

Parameters

owner_or_condition (Union[Hashable, ConditionBase]) – Either a condition or the owner of a condition

Returns

The condition removed, or None if no condition removed

Raises

ConditionError –

• when owner_or_condition is an owner and it owns multiple conditions - when owner_or_condition is a condition and its owner is None

run(termination_conds=None, context=None, base_context=<psyneulink.core.globals.context.Context object>, skip_trial_time_increment=False)

run is a python generator, that when iterated over provides the next CONSIDERATION_SET_EXECUTION of executions at each iteration

Parameters

termination_conds – (dict) - a mapping from TimeScales to Conditions that when met terminate the execution of the specified TimeScale

add_graph_edge(sender, receiver)

Adds an edge to the graph from sender to receiver. Equivalent to add_condition(sender, AddEdgeTo(receiver)).

Parameters

• sender (Hashable) – sender of the new edge

• receiver (Hashable) – receiver of the new edge

Returns

the new condition added to implement the edge

Return type

AddEdgeTo

remove_graph_edge(sender, receiver)

Removes an edge from the graph from sender to receiver if it exists. Equivalent to add_condition(receiver, RemoveEdgeFrom(sender)).

Parameters

• sender (Hashable) – sender of the edge to be removed

• receiver (Hashable) – receiver of the edge to be removed

Returns

the new condition added to implement the edge

Return type

RemoveEdgeFrom

class psyneulink.core.scheduling.scheduler.SchedulingMode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

STANDARD

Standard Mode

EXACT_TIME

Exact time Mode