Skip to content

Event

gradysim.simulator.event

As an event-based simulator one of the main components in the simulation is an event loop, that's the focus of this module. Events are compact classes containing a timestamp and a callback. Events are inserted into the event loop which is organized as a heap to keep the events with the smallest timestamps on top. At every simulation iteration the simulator class grabs the event with the smallest timestamp and executes its callback.

Events are created by handlers. Protocols indirectly interact with them through the provider interface they have access to. These events, when executed, cause effects on the network nodes, mainly observed through calls to the protocol interface methods like handle_timer.

Event dataclass

Dataclass representing a single event. Will be placed inside the simulation loop. Shouldn't be instantiated directly, but through the EventLoop.schedule_event method.

Source code in gradysim/simulator/event.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@dataclass
class Event:
    """
    Dataclass representing a single event. Will be placed inside the simulation loop. Shouldn't be instantiated
    directly, but through the `EventLoop.schedule_event` method.
    """

    timestamp: float
    """Simulation time in seconds when the event will fire"""
    callback: Callable
    """Any callable with no parameters. Will be executed when the event fires"""
    context: str
    """
    Context in which the event will be executed. 
    This is used in logging to identify where the callback is being executed.
    """

    def __lt__(self, other):
        return self.timestamp < other.timestamp

callback: Callable instance-attribute

Any callable with no parameters. Will be executed when the event fires

context: str instance-attribute

Context in which the event will be executed. This is used in logging to identify where the callback is being executed.

timestamp: float instance-attribute

Simulation time in seconds when the event will fire

EventLoop

Event loop central to the event-based simulation. Is implemented as a min-heap populated by Event instances ordered by their timestamps. Generally only the Simulator will call the pop_event method, a handler should only need to use the schedule_event method.

Source code in gradysim/simulator/event.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class EventLoop:
    """
    Event loop central to the event-based simulation. Is implemented as a min-heap populated by `Event` instances ordered
    by their timestamps. Generally only the [`Simulator`][gradysim.simulator.simulation.Simulator] will call the `pop_event`
    method, a handler should only need to use the `schedule_event` method.
    """
    _event_heap: List[Event]
    _current_time: float

    def __init__(self):
        """
        Creates an event loop
        """
        self._event_heap = []
        self._current_time = 0

    def schedule_event(self, timestamp: float, callback: Callable, context: str = "") -> None:
        """
        Creates an event instance with the information provided as args and inserts it into the event heap.

        Args:
            timestamp: Simulation time in seconds when the event should fire
            callback: Any callable with no arguments
            context: Context where the callable executes. Useful for logging.
        """
        if timestamp < self.current_time:
            raise EventLoopException(f"Could not schedule event: tried to schedule at {timestamp} which is "
                                     f"earlier than the current time {self.current_time}. ")

        heapq.heappush(self._event_heap, Event(timestamp, callback, context))

    def pop_event(self) -> Event:
        """
        Removes an event from the top of the event heap and returns it. If called when the event heap is empty it will
        raise EventLoopException.

        Returns: The event popped.

        """
        if len(self._event_heap) == 0:
            raise EventLoopException("Could not pop event: the event queue is empty")

        event = heapq.heappop(self._event_heap)
        self._current_time = event.timestamp
        return event

    def peek_event(self) -> Optional[Event]:
        """
        Peeks at the event at the top of the event heap without removing it. Returns None if the event heap is empty.

        Returns:
            The event at the top of the event heap or None if it's empty
        """
        if len(self._event_heap) == 0:
            return None
        return self._event_heap[0]

    def clear(self) -> None:
        """
        Clears the event heap
        """
        self._event_heap.clear()

    @property
    def current_time(self) -> float:
        """
        Returns the timestamp of the last event to be removed from the heap.
        This represents the current simulation time.

        Returns:
            The current simulation fime
        """
        return self._current_time

    def __len__(self) -> int:
        """
        By calling len() on the EventLoop you can check how many events are queued in the event heap.

        Returns:
            Number of events in the event heap
        """
        return len(self._event_heap)

current_time: float property

Returns the timestamp of the last event to be removed from the heap. This represents the current simulation time.

Returns:

Type Description
float

The current simulation fime

__init__()

Creates an event loop

Source code in gradysim/simulator/event.py 
51
52
53
54
55
56
def __init__(self):
    """
    Creates an event loop
    """
    self._event_heap = []
    self._current_time = 0

__len__()

By calling len() on the EventLoop you can check how many events are queued in the event heap.

Returns:

Type Description
int

Number of events in the event heap
Source code in gradysim/simulator/event.py 
116
117
118
119
120
121
122
123
def __len__(self) -> int:
    """
    By calling len() on the EventLoop you can check how many events are queued in the event heap.

    Returns:
        Number of events in the event heap
    """
    return len(self._event_heap)

clear()

Clears the event heap

Source code in gradysim/simulator/event.py 
 99
100
101
102
103
def clear(self) -> None:
    """
    Clears the event heap
    """
    self._event_heap.clear()

peek_event()

Peeks at the event at the top of the event heap without removing it. Returns None if the event heap is empty.

Returns:

Type Description
Optional[Event]

The event at the top of the event heap or None if it's empty
Source code in gradysim/simulator/event.py 
88
89
90
91
92
93
94
95
96
97
def peek_event(self) -> Optional[Event]:
    """
    Peeks at the event at the top of the event heap without removing it. Returns None if the event heap is empty.

    Returns:
        The event at the top of the event heap or None if it's empty
    """
    if len(self._event_heap) == 0:
        return None
    return self._event_heap[0]

pop_event()

Removes an event from the top of the event heap and returns it. If called when the event heap is empty it will raise EventLoopException.

Returns: The event popped.

Source code in gradysim/simulator/event.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def pop_event(self) -> Event:
    """
    Removes an event from the top of the event heap and returns it. If called when the event heap is empty it will
    raise EventLoopException.

    Returns: The event popped.

    """
    if len(self._event_heap) == 0:
        raise EventLoopException("Could not pop event: the event queue is empty")

    event = heapq.heappop(self._event_heap)
    self._current_time = event.timestamp
    return event

schedule_event(timestamp, callback, context='')

Creates an event instance with the information provided as args and inserts it into the event heap.

Parameters:

Name Type Description Default
timestamp float

Simulation time in seconds when the event should fire

 required
callback Callable

Any callable with no arguments

 required
context str

Context where the callable executes. Useful for logging.

 ''
Source code in gradysim/simulator/event.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def schedule_event(self, timestamp: float, callback: Callable, context: str = "") -> None:
    """
    Creates an event instance with the information provided as args and inserts it into the event heap.

    Args:
        timestamp: Simulation time in seconds when the event should fire
        callback: Any callable with no arguments
        context: Context where the callable executes. Useful for logging.
    """
    if timestamp < self.current_time:
        raise EventLoopException(f"Could not schedule event: tried to schedule at {timestamp} which is "
                                 f"earlier than the current time {self.current_time}. ")

    heapq.heappush(self._event_heap, Event(timestamp, callback, context))