Reference

Contents

Reference#

Kernel#

class deltacycle.Kernel.State(*values)#

Transitions:

INIT -> RUNNING -> COMPLETED
                -> FINISHED
class deltacycle.Kernel#

Simulation Kernel.

Responsible for:

  • Scheduling and executing tasks

  • Updating model state

Time is an integer. Its initial value is -1, indicating “before time”. All tasks must be scheduled at a non-negative time.

All kernels have a main task, which is scheduled to run at time zero.

This is a low level API. User code is not expected to interact with it directly. To run a simulation, use the run and step functions.

state() → State#

Current simulation state.

time() → int#

Current simulation time.

property main: Task#

Parent task of all other tasks.

task() → Task#

Currently running task.

done() → bool#

Return True if the kernel is done.

A kernel that is “done” either:

  • Exhaused all tasks (COMPLETED), or

  • Called finish (FINISHED)

clear()#

Clear all simulation state.

abstractmethod call_soon(task: Task, args: TaskArgs) → None#

Schedule task to run soon, in current time slot.

abstractmethod call_later(delay: int, task: Task, args: TaskArgs) → None#

Schedule task to run later, after delay.

abstractmethod call_at(when: int, task: Task, args: TaskArgs) → None#

Schedule task to run at specified time: when.

abstractmethod create_main(coro: TaskCoro) → Task#

Create main task, and schedule it at time zero.

Returns:

Handle to the main task

abstractmethod create_task(coro: TaskCoro, name: str | None = None, **kwargs: Any) → Task#

Create child task, and schedule it soon.

Returns:

Handle to the created task

class deltacycle.DefaultKernel#

Default simulation kernel

Tasks are scheduled with a (heapq) priority queue.

Task ordering rules:

  • Tasks scheduled at different times run in time order.

  • Tasks scheduled at same time run in priority order.

  • Tasks scheduled at same time with same priority run in insertion order.

Priority is an arbitrary integer.

The main (parent) task will be assigned priority zero.

deltacycle.finish()#

Halt all incomplete coroutines, and immediately exit simulation.

Clear all kernel data, and transition state to FINISHED.

deltacycle.get_running_kernel() → Kernel#

Return currently running kernel.

May be used by a simulation task to access kernel state.

Returns:

Kernel instance.

Raises:

RuntimeError – No kernel, or kernel is not currently running.

deltacycle.get_kernel() → Kernel | None#

Get the current kernel.

DeltaCycle only supports one simulation kernel at a time. This function gets the handle to that kernel, which may be None.

May be used by high level code to manage multiple kernels.

Returns:

Kernel instance or None.

deltacycle.set_kernel(kernel: Kernel | None = None)#

Set the current kernel.

DeltaCycle only supports one simulation kernel at a time. This function sets the handle to that kernel, which may be None.

May be used by high level code to manage multiple kernels.

Parameters:

kernel – Kernel instance or None.

Tasks#

type deltacycle.TaskCoro = Coroutine[None, Sendable | None, ResultType]#
exception deltacycle.Throwable#

Throw a signal to a task.

exception deltacycle.Interrupt#

Bases: Throwable

Interrupt task.

class deltacycle.Task.State(*values)#

Transitions:

        PENDING
           |
INIT -> RUNNING -> RETURNED
                -> EXCEPTED
class deltacycle.Task(coro: TaskCoro[ResultType], name: str)#

Bases: KernelIf, Blocking, Sendable, Generic

Manage the life cycle of a coroutine.

Do NOT instantiate a Task directly. Use create_task function, or (better) TaskGroup.create_task method.

__await__() → Generator[None, Sendable, Any]#
property coro: TaskCoro[ResultType]#

Wrapped coroutine.

property name: str#

Task name.

Primarily for debug; no functional effect. There are no rules or restrictions for valid names. Give tasks unique and recognizable names to help identify them.

If not provided to the create_task function, a default name of Task-{index} will be assigned, where index is a monotonically increasing integer value, starting from 0.

property group: TaskGroup | None#

Return TaskGroup, or None.

If the task was started by a TaskGroup’s create_task method, it will assign this property to point to the TaskGroup instance.

state() → State#
done() → bool#

Return True if the task is done.

A task that is “done” either:

  • Completed normally, or

  • Raised an exception.

result() → ResultType#

Return the task’s result, or raise an exception.

Returns:

If the task ran to completion, return its result.

Raises:

  • Exception – If the task raise any other type of exception.

  • RuntimeError – If the task is not done.

exception() → Exception | None#

Return the task’s exception.

Returns:

If the task raised an exception, return it. Otherwise, return None.

Raises:

RuntimeError – If the task is not done.

interrupt(*args: Any) → bool#

Interrupt task.

If a task is already done: return False.

If a task is pending:

  1. Renege from all queues

  2. Reschedule to raise Interrupt in the current time slot

  3. Return True

If a task is running, immediately raise Interrupt.

Parameters:

args – Arguments passed to Interrupt instance

Returns:

bool success indicator

Raises:

Interrupt – If the task interrupts itself

class deltacycle.TaskQueue#
abstractmethod __bool__() → bool#

Return True if the queue has tasks ready to run.

abstractmethod push(item: Any) → None#

Push item to queue tail.

abstractmethod pop() → Any#

Pop item from queue head.

abstractmethod drop(task: Task) → None#

Drop task from queue.

class deltacycle.TaskGroup#

Bases: KernelIf

Group of tasks.

async __aenter__() → Self#
async __aexit__(exc_type: type[BaseException] | None, exc: BaseException | None, traceback: TracebackType | None)#
create_task(coro: TaskCoro, name: str | None = None, **kwargs: Any) → Task#
deltacycle.create_task(coro: TaskCoro, name: str | None = None, **kwargs: Any) → Task#

Create a task, and schedule it to start soon.

Parameters:

  • coro – Coroutine function instance.

  • name – Identify the task for logging/debugging purposes. If not given, a default name like Task-{index} will be assigned. Not guaranteed to be unique.

  • kwargs – Arguments passed to the kernel to customize task execution.

Returns:

Created Task instance.

Raises:

RuntimeError – No kernel, or kernel is not currently running.

deltacycle.get_current_task() → Task#

Return currently running task.

Returns:

Task instance.

Raises:

RuntimeError – No kernel, or kernel is not currently running.

Variables#

type deltacycle.Predicate = Callable[[], bool]#
class deltacycle.Variable#

Bases: KernelIf, Blocking, Sendable

Model component that changes over time.

The instantaneous state of a simulation is represented by a collection of variables.

There are two types of variables: singular, and aggregate.

Children:

       Variable
          |
   +------+------+
   |             |
Singular     Aggregate

  • A singular variable has one value.

  • An aggregate variable is a mapping of key, value pairs.

Variables are always blocking. Tasks may schedule updates to variables. Changes to variable values may unblock tasks, which may in turn schedule updates to other variables.

__await__() → Generator[None, Sendable, Self]#

Await variable change:

For variable v:

  1. Suspend the current task.

  2. When another task invokes v.set_next(...) and v.changed() evaluates to True, unblock all tasks waiting for that event.

pred(p: Predicate) → PredVar#

Return blocking, predicated variable.

Parameters:

p – Prediate function w/ no args and bool return type.

Returns:

Predicated Variable (PredVar) object.

abstractmethod changed() → bool#

Return True if changed during the current time slot.

abstractmethod update() → None#

Update variable value.

class deltacycle.PredVar(var: Variable, p: Predicate)#

Predicated Variable.

A lightweight wrapper around a Variable instance. Implements Awaitable and Blocking. Can be used in await, await AllOf and await AnyOf statements.

Predicate functions allow fine-grained control of variable dependencies. Sometimes waiting tasks can be woken up when there is any change to the variables’s value. However, it is often desirable to only trigger on particular types of changes. For example, in digital logic a flip-flop might only update its state when 1) reset is inactive, 2) clock is transitioning from low to high (a positive edge), AND 3) a data enable signal is active. A predicate function may be used to evaluate when those conditions are all true.

__await__() → Generator[None, Sendable, Variable]#

Await variable change:

For variable v, and predicate function p:

  1. Suspend the current task.

  2. When another task invokes v.set_next(...) and p evaluates to True, unblock all tasks waiting for that event.

class deltacycle.Value#

Bases: ABC, Generic

Variable value.

abstractmethod get_prev() → T#

Return value at the end of the previous timeslot.

abstract property prev: T#

Return value at the end of the previous timeslot.

abstractmethod set_next(value: T) → None#

Schedule update to value in the current timeslot.

abstract property next#
class deltacycle.Singular(value: T)#

Bases: Variable, Value, Generic

Model state organized as a single unit.

get_value() → T#

Return present value.

When performing multiple updates to a variable during the same timeslot, this method will always return the value of the latest update.

property value: T#

Return present value.

When performing multiple updates to a variable during the same timeslot, this method will always return the value of the latest update.

class deltacycle.Aggregate(value: T)#

Bases: Variable, Generic

Model state organized as multiple units.

__getitem__(key: Hashable) → AggrItem[T]#
get_prev(key: Hashable) → T#

Return value at the end of the previous timeslot.

set_next(key: Hashable, value: T)#

Schedule update to value in the current timeslot.

get_value() → AggrValue[T]#

Return present value.

When performing multiple updates to a variable during the same timeslot, this method will always return the value of the latest update.

property value: AggrValue[T]#

Return present value.

When performing multiple updates to a variable during the same timeslot, this method will always return the value of the latest update.

class deltacycle.AggrItem(aggr: Aggregate[T], key: Hashable)#

Bases: Value, Generic

Wrap Aggregate __getitem__.

class deltacycle.AggrValue(aggr: Aggregate[T])#

Wrap Aggregate value.

__getitem__(key: Hashable) → T#

Synchronization Primitives#

class deltacycle.Event#

Bases: KernelIf, Blocking, Sendable

Notify multiple tasks that some event has happened.

An event instance is lightweight. It consists of a flag, and a FIFO of waiting tasks.

When the event is created, its flag defaults to False. In this state, the event will block all tasks that await it. When a task invokes the event’s set method, that sets the flag (to True), and unblocks all waiting tasks.

If the event’s flag is set, it will not block awaiting tasks. When a task invokes the event’s clear method, that clears the flag (to False), and the event will go back to blocking awaiting tasks.

__await__() → Generator[None, Sendable, Self]#

Await event set.

__bool__() → bool#

Return flag state.

set()#

Set the flag. Stop blocking waiting tasks.

clear()#

Clear the flag. Start blocking waiting tasks.

class deltacycle.Semaphore(value: int = 0, capacity: int = 0)#

Bases: KernelIf, Sendable

__len__() → int#
property capacity: int | None#
req(priority: int = 0) → ReqSemaphore#
put()#
try_get() → bool#
async get(priority: int = 0)#
class deltacycle.ReqSemaphore(sem: Semaphore, priority: int)#
async __aenter__() → Self#
async __aexit__(exc_type: type[BaseException] | None, exc: BaseException | None, traceback: TracebackType | None)#
class deltacycle.Lock#

Bases: Semaphore

class deltacycle.CreditPool(value: int = 0, capacity: int = 0)#

Bases: KernelIf, Sendable

__len__() → int#
property capacity: int | None#
req(n: int = 1, priority: int = 0) → ReqCredit#
put(n: int = 1)#
try_get(n: int = 1) → bool#
async get(n: int = 1, priority: int = 0)#
class deltacycle.ReqCredit(credits: CreditPool, n: int, priority: int)#
async __aenter__() → Self#
async __aexit__(exc_type: type[BaseException] | None, exc: BaseException | None, traceback: TracebackType | None)#
class deltacycle.Queue(capacity: int = 0)#

Bases: KernelIf, Generic

Producer / Consumer FIFO Queue.

Has both blocking and non-blocking put and get interfaces. If capacity is a positive number, the queue has capacity slots. If capacity is zero or a negative number, the queue has infinite slots.

The put interface will block only when it is full. The get interface will block only when it is empty.

An infinite queue will never be full. Its size is subject only to the machine’s memory limitations.

__len__() → int#
property capacity: int | None#
empty() → bool#
full() → bool#
try_put(item: T) → bool#

Nonblocking put: Return True if a put attempt is successful.

async put(item: T, priority: int = 0)#

Block until there is space for an item.

try_get() → tuple[bool, T | None]#

Nonblocking get.

Returns:

If the get is successful, (True, item); If unsuccessful, (False, None).

async get(priority: int = 0) → T#

Block until an item is available.

class deltacycle.Container(capacity: int = 0)#

Bases: KernelIf

Producer / Consumer Resource Container.

Has both blocking and non-blocking put and get interfaces. If capacity is a positive number, the container has capacity slots. If capacity is zero or a negative number, the container has infinite slots.

The put interface will block only when it is full. The get interface will block only when it is empty.

An infinite container will never be full. Its size is subject only to the machine’s memory limitations.

__len__() → int#
property capacity: int | None#
try_put(n: int = 1) → bool#
async put(n: int = 1, priority: int = 0)#
try_get(n: int = 1) → bool#
async get(n: int = 1, priority: int = 0)#

Scheduling#

class deltacycle.Blocking#
class deltacycle.Sendable#
class deltacycle.AllOf(*bs: Blocking)#
class deltacycle.AnyOf(*bs: Blocking)#
deltacycle.run(coro: TaskCoro | None = None, kernel: Kernel | None = None, kernel_type: type[Kernel] = <class 'deltacycle._kernel.DefaultKernel'>, ticks: int | None = None, until: int | None = None) → MainResultType | None#

Run a simulation.

If a simulation hits the run limit, it will exit and return None. That simulation may be resumed any number of times. If all tasks are exhausted, return the main coroutine result.

Parameters:

  • coro – Main coroutine function instance. Required if creating a new kernel. Ignored if using an existing kernel.

  • kernel – Optional Kernel instance. If not provided, a new kernel will be created.

  • ticks – Optional relative run limit. If provided, run for ticks simulation time steps.

  • until – Optional absolute run limit. If provided, run until ticks simulation time steps.

Returns:

If the main coroutine runs til completion, return its result. Otherwise, return None.

Raises:

  • ValueError – Creating a new kernel, but no main coroutine provided.

  • RuntimeError – The kernel is in an invalid state.

deltacycle.step(coro: TaskCoro | None = None, kernel: Kernel | None = None, kernel_type: type[Kernel] = <class 'deltacycle._kernel.DefaultKernel'>) → Generator[int, None, MainResultType]#

Step (iterate) a simulation.

Iterated simulations do not have a run limit. It is the user’s responsibility to break at the desired time. If all tasks are exhausted, return the main coroutine result.

Parameters:

  • coro – Main coroutine function instance. Required if creating a new kernel. Ignored if using an existing kernel.

  • kernel – Optional Kernel instance. If not provided, a new kernel will be created.

Yields:

Time immediately before the next time slot executes.

Returns:

Main coroutine result.

Raises:

  • ValueError – Creating a new kernel, but no main coroutine provided.

  • RuntimeError – The kernel is in an invalid state.

deltacycle.now() → int#

Return current simulation time.

Returns:

Current simulation time.

Raises:

RuntimeError – No kernel, or kernel is not currently running.

async deltacycle.sleep(delay: int)#

Suspend the current task, and wake up after a delay.

async deltacycle.all_of(*bs: Blocking) → tuple[Sendable, ...]#

Block forward progress until all items are unblocked.

Parameters:

bs – Sequence of blocking items.

Returns:

Return a tuple of items in unblocking order.

async deltacycle.any_of(*bs: Blocking) → Sendable | None#

Block forward progress until at least one item is unblocked.

Parameters:

bs – Sequence of blocking items.

Returns:

If the input is empty, return None. Otherwise, return the item that unblocked first.

Contents