Programmable Timers

Programmable timers provide a convenient way for reactors to schedule future behavior. While ports can be used to send messages to other reactors, programmable times can be seen as a mechanism for a reactor to send a delayed messages to its future self.

The following example replicates the behavior of a Periodic Timers using a programmable timer.

import datetime

import xronos


class Clock(xronos.Reactor):
    _tick = xronos.ProgrammableTimerDeclaration()

    def __init__(self, period: datetime.timedelta):
        super().__init__()
        self._period = period

    def _schedule_next_tick(self, tick_effect):
        tick_effect.schedule(value=None, delay=self._period)

    @xronos.reaction
    def next(self, interface):
        interface.add_trigger(self.startup)
        interface.add_trigger(self._tick)
        tick_effect = interface.add_effect(self._tick)

        return lambda: self._schedule_next_tick(tick_effect)

    @xronos.reaction
    def tick(self, interface):
        interface.add_trigger(self._tick)

        return lambda: print(f"Tick at {self.get_time_since_startup()}")


def main():
    env = xronos.Environment()
    env.create_reactor("clock", Clock, period=datetime.timedelta(seconds=1))
    env.execute()


if __name__ == "__main__":
    main()

The Clock reactor declares a programmable timer called _tick using a ProgrammableTimerDeclaration. The reactor further defines two reactions. The on_tick reaction is triggered by _tick and prints a message that includes the current time. The next reaction is triggered both by startup and by _tick. It also declares an effect on _tick. Using the ProgrammableTimerEffect object, the reaction may call schedule() to schedule a future event. In this example, the reaction handler calls _schedule_next_tick() which in turn calls schedule() using self._period as delay.

The next reaction ensures that _tick triggers in regular intervals. The first invocation of the next reaction handler is triggered by startup. In this first invocation, the handler schedules the first event on _tick. All subsequent invocations of next will be triggered by _tick itself.

When executed, the program produces the following output.

$ python clock.py
Tick at 0:00:01
Tick at 0:00:02
Tick at 0:00:03
Tick at 0:00:04
Tick at 0:00:05
...

While we can use programmable timers to replicate the behavior of periodic timers, they are a lot more versatile. Since new events are scheduled by reactions, the reaction handlers gain full control over when events occur. This allows us to implement more interesting behavior. For instance, the following example implements a clock that slows down with each tick.

class SlowingClock(Clock):
    def __init__(self, period, increment):
        super().__init__(period)
        self._increment = increment

    def _schedule_next_tick(self, tick_effect):
        tick_effect.schedule(value=None, delay=self._period)
        self._period += self._increment


def main():
    env = xronos.Environment()
    env.create_reactor(
        "clock",
        SlowingClock,
        period=datetime.timedelta(seconds=1),
        increment=datetime.timedelta(milliseconds=200),
    )
    env.execute()


if __name__ == "__main__":
    main()

With this modification, the program produces the following output.

$ python clock.py
Tick at 0:00:01
Tick at 0:00:02.200000
Tick at 0:00:03.600000
Tick at 0:00:05.200000
Tick at 0:00:07
...