1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
#!/usr/bin/env python3
from abc import ABC, abstractmethod
import datetime
import logging
import time
import threading
from typing import Dict, Optional
import pytz
from thread_utils import background_thread
logger = logging.getLogger(__name__)
class StateTracker(ABC):
"""A base class that maintains and updates a global state via an
update routine. Instances of this class should be periodically
invoked via the heartbeat() method. This method, in turn, invokes
update() with update_ids according to a schedule / periodicity
provided to the c'tor.
"""
def __init__(self, update_ids_to_update_secs: Dict[str, float]) -> None:
"""The update_ids_to_update_secs dict parameter describes one or more
update types (unique update_ids) and the periodicity(ies), in
seconds, at which it/they should be invoked.
Note that, when more than one update is overdue, they will be
invoked in order by their update_ids so care in choosing these
identifiers may be in order.
"""
self.update_ids_to_update_secs = update_ids_to_update_secs
self.last_reminder_ts: Dict[str, Optional[datetime.datetime]] = {}
for x in update_ids_to_update_secs.keys():
self.last_reminder_ts[x] = None
@abstractmethod
def update(
self,
update_id: str,
now: datetime.datetime,
last_invocation: Optional[datetime.datetime],
) -> None:
"""Put whatever you want here. The update_id will be the string
passed to the c'tor as a key in the Dict. It will only be
tapped on the shoulder, at most, every update_secs seconds.
The now param is the approximate current timestamp and the
last_invocation param is the last time you were invoked (or
None on the first invocation)
"""
pass
def heartbeat(self, *, force_all_updates_to_run: bool = False) -> None:
"""Invoke this method to cause the StateTracker instance to identify
and invoke any overdue updates based on the schedule passed to
the c'tor. In the base StateTracker class, this method must
be invoked manually with a thread from external code.
If more than one type of update (update_id) are overdue,
they will be invoked in order based on their update_ids.
Setting force_all_updates_to_run will invoke all updates
(ordered by update_id) immediately ignoring whether or not
they are due.
"""
self.now = datetime.datetime.now(tz=pytz.timezone("US/Pacific"))
for update_id in sorted(self.last_reminder_ts.keys()):
if force_all_updates_to_run:
logger.debug('Forcing all updates to run')
self.update(
update_id, self.now, self.last_reminder_ts[update_id]
)
self.last_reminder_ts[update_id] = self.now
return
refresh_secs = self.update_ids_to_update_secs[update_id]
last_run = self.last_reminder_ts[update_id]
if last_run is None: # Never run before
logger.debug(
f'id {update_id} has never been run; running it now'
)
self.update(
update_id, self.now, self.last_reminder_ts[update_id]
)
self.last_reminder_ts[update_id] = self.now
else:
delta = self.now - last_run
if delta.total_seconds() >= refresh_secs: # Is overdue?
logger.debug(f'id {update_id} is overdue; running it now')
self.update(
update_id,
self.now,
self.last_reminder_ts[update_id],
)
self.last_reminder_ts[update_id] = self.now
class AutomaticStateTracker(StateTracker):
"""Just like HeartbeatCurrentState but you don't need to pump the
heartbeat; it runs on a background thread. Call .shutdown() to
terminate the updates.
"""
@background_thread
def pace_maker(self, should_terminate) -> None:
"""Entry point for a background thread to own calling heartbeat()
at regular intervals so that the main thread doesn't need to do
so."""
while True:
if should_terminate.is_set():
logger.debug('pace_maker noticed event; shutting down')
return
self.heartbeat()
logger.debug(f'pace_maker is sleeping for {self.sleep_delay}s')
time.sleep(self.sleep_delay)
def __init__(
self,
update_ids_to_update_secs: Dict[str, float],
*,
override_sleep_delay: Optional[float] = None,
) -> None:
import math_utils
super().__init__(update_ids_to_update_secs)
if override_sleep_delay is not None:
logger.debug(f'Overriding sleep delay to {override_sleep_delay}')
self.sleep_delay = override_sleep_delay
else:
periods_list = list(update_ids_to_update_secs.values())
self.sleep_delay = math_utils.gcd_float_sequence(periods_list)
logger.info(f'Computed sleep_delay={self.sleep_delay}')
(thread, stop_event) = self.pace_maker()
self.should_terminate = stop_event
self.updater_thread = thread
def shutdown(self):
"""Terminates the background thread and waits for it to tear down.
This may block for as long as self.sleep_delay.
"""
logger.debug(
'Setting shutdown event and waiting for background thread.'
)
self.should_terminate.set()
self.updater_thread.join()
logger.debug('Background thread terminated.')
class WaitableAutomaticStateTracker(AutomaticStateTracker):
def __init__(
self,
update_ids_to_update_secs: Dict[str, float],
*,
override_sleep_delay: Optional[float] = None,
) -> None:
self._something_changed = threading.Event()
super().__init__(update_ids_to_update_secs,
override_sleep_delay=override_sleep_delay)
def something_changed(self):
self._something_changed.set()
def did_something_change(self) -> bool:
return self._something_changed.is_set()
def reset(self):
self._something_changed.clear()
def wait(self,
*,
timeout=None):
return self._something_changed.wait(
timeout=timeout
)
|
