Periodic Timers

The xronos framework provides powerful mechanisms for implementing timed behavior. The simplest mechanism is provided by periodic timers. Copy the following example into a new file called timer.py.

plainwith type hintsdiagram

import datetime

import xronos

MAX_COUNT = 5


class Timed(xronos.Reactor):
    _timer = xronos.PeriodicTimerDeclaration()

    def __init__(self, period):
        super().__init__()
        self._timer.period = period
        self._count = 0

    @xronos.reaction
    def hello(self, interface):
        interface.add_trigger(self.startup)
        return lambda: print(f"{self.get_time()}: {self.name} says 'Hello!'")

    @xronos.reaction
    def on_timer(self, interface):
        interface.add_trigger(self._timer)

        def handler():
            self._count += 1
            print(
                f"{self.get_time()}: {self.name}'s timer triggered "
                f"(count={self._count})"
            )
            if self._count == MAX_COUNT:
                self.request_shutdown()

        return handler

    @xronos.reaction
    def goodbye(self, interface):
        interface.add_trigger(self.shutdown)
        return lambda: print(f"{self.get_time()}: {self.name} says 'Goodbye!'")


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


if __name__ == "__main__":
    main()

In xronos, a periodic timer can be created by instantiating xronos.PeriodicTimerDeclaration as a class attribute. This declares a PeriodicTimer object that is automatically initialized by the xronos runtime. We can interact with the timer object simply by using self._timer.

Note that the __init__ method of Timed configures the timer’s period according to an argument that it receives. It further defines the state variable self._count and sets it to 0.

The Timed reactor defines three reactions. The hello reaction is similar to the reaction in our “Hello, World!” example. It prints the current time, the reactors name and ‘Hello!’. The goodbye reaction produces a similar output in response to the builtin shutdown event.

The reaction on_timer declares the timer as its trigger and defines a reaction handler that is a little bit more complex than the simple lambda function we have seen so far. The reaction defines a nested handler function that it then returns.

def handler():
    self._count += 1
    print(
        f"{self.get_time()}: {self.name}'s timer triggered "
        f"(count={self._count})"
    )
    if self._count == MAX_COUNT:
        self.request_shutdown()

return handler

The handler performs multiple steps. First, it increments the self._count variable. It then prints the current time, the timer’s name and the current count. Finally, it checks if the count has reached MAX_COUNT, in which case request_shutdown() is called to terminate the program.

The program runs for about 4 seconds and produces output similar to the following:

$ python timer.py
2024-11-29 10:17:13.893071: timed says 'Hello!'
2024-11-29 10:17:13.893071: timed's timer triggered (count=1)
2024-11-29 10:17:14.893071: timed's timer triggered (count=2)
2024-11-29 10:17:15.893071: timed's timer triggered (count=3)
2024-11-29 10:17:16.893071: timed's timer triggered (count=4)
2024-11-29 10:17:17.893071: timed's timer triggered (count=5)
2024-11-29 10:17:17.893071: timed says 'Goodbye!'

Carefully look at the printed timestamps. The timestamp printed by hello is identical to the first line printed by on_timer. This is, because we did not configure an offset and the first event triggers immediately at startup. In other words: the first timer event is simultaneous to the startup event. Also note that the individual timer events are perfectly spaced one second apart.

How can xronos achieve such a precision? In xronos, time behaves differently than the wall-clock time that computer programs usually use. The xronos runtime manages its own clock, and this clock is key to controlling how a program executes. One property of the internal clock is that the time observed by a reaction is equal to the timestamp of the triggering event. Moreover, any two calls to get_time() within the same reaction handler will return the same time value.

The synchrony of timestamps is also preserved across multiple reactors. Modify the instantiation code as shown in the following, so that the program consists of three Timed reactors using a period of 1 second, 2 seconds and 3 seconds, respectively.

plainwith type hintsdiagram

def main():
    env = xronos.Environment()
    env.create_reactor("timed1", Timed, period=datetime.timedelta(seconds=1))
    env.create_reactor("timed2", Timed, period=datetime.timedelta(seconds=2))
    env.create_reactor("timed3", Timed, period=datetime.timedelta(seconds=3))
    env.execute()

Running the program produces output similar to the following:

$ python timer.py
2024-11-29 10:48:57.423695: timed2 says 'Hello!'
2024-11-29 10:48:57.423695: timed3 says 'Hello!'
2024-11-29 10:48:57.423695: timed1 says 'Hello!'
2024-11-29 10:48:57.423695: timed2's timer triggered (count=1)
2024-11-29 10:48:57.423695: timed3's timer triggered (count=1)
2024-11-29 10:48:57.423695: timed1's timer triggered (count=1)
2024-11-29 10:48:58.423695: timed1's timer triggered (count=2)
2024-11-29 10:48:59.423695: timed2's timer triggered (count=2)
2024-11-29 10:48:59.423695: timed1's timer triggered (count=3)
2024-11-29 10:49:00.423695: timed3's timer triggered (count=2)
2024-11-29 10:49:00.423695: timed1's timer triggered (count=4)
2024-11-29 10:49:01.423695: timed2's timer triggered (count=3)
2024-11-29 10:49:01.423695: timed1's timer triggered (count=5)
2024-11-29 10:49:01.423695: timed2 says 'Goodbye!'
2024-11-29 10:49:01.423695: timed3 says 'Goodbye!'
2024-11-29 10:49:01.423695: timed1 says 'Goodbye!'

Note how all three reactors say ‘Hello!’ or ‘Goodbye!’ at the same time, and how the timestamps of individual timer events align between the reactors.