From 17299415052cf3644f3d54ca4e6bbe5131aa1688 Mon Sep 17 00:00:00 2001 From: Colin Constable Date: Tue, 7 Jul 2026 11:51:12 -0700 Subject: [PATCH 1/2] fix: daemonize the monitor heartbeat thread and scope its locks per instance Two defects in AtMonitorConnection's lifecycle: 1. The heartbeat thread was non-daemon and runs an infinite loop, so any process that ever constructed a monitor connection could not exit, and every abandoned/replaced connection leaked a live thread. 2. should_be_running_lock and running_lock were module-level singletons (in atconstants, star-imported here and used from AtClient), so every monitor connection in a process shared one lock pair and serialized against the others' state transitions. The heartbeat thread is now created with daemon=True and kept on the instance (_heartbeat_thread). The locks are now instance attributes; AtClient uses the connection's own locks. The now-unused atconstants star imports are removed from both files; the definitions in atconstants are left in place so any external importer keeps working. Adds network-free tests: daemon flag, per-instance lock identity, and that one connection's lock does not block another's. --- at_client/atclient.py | 15 ++--- at_client/connections/atmonitorconnection.py | 62 ++++++++++---------- test/monitor_heartbeat_test.py | 37 ++++++++++++ 3 files changed, 75 insertions(+), 39 deletions(-) create mode 100644 test/monitor_heartbeat_test.py diff --git a/at_client/atclient.py b/at_client/atclient.py index 32a6227..afbcc2e 100644 --- a/at_client/atclient.py +++ b/at_client/atclient.py @@ -17,7 +17,6 @@ from .connections.atrootconnection import AtRootConnection from .connections.atsecondaryconnection import AtSecondaryConnection from .connections.atmonitorconnection import AtMonitorConnection -from .util.atconstants import * from .connections.address import Address from .common.keys import AtKey, Keys, SharedKey, PrivateHiddenKey, PublicKey, SelfKey from .util.authutil import AuthUtil @@ -375,7 +374,6 @@ def __del__(self): def start_monitor(self, regex=".*"): if self.queue != None: - global should_be_running_lock what = "" try: if self.monitor_connection == None: @@ -383,13 +381,13 @@ def start_monitor(self, regex=".*"): self.monitor_connection = AtMonitorConnection(queue=self.queue, atsign=self.atsign, address=self.secondary_address, verbose=self.verbose, regex=regex) self.monitor_connection.connect() AuthUtil.authenticate_with_pkam(self.monitor_connection, self.atsign, self.keys) - should_be_running_lock.acquire(blocking=1) + self.monitor_connection.should_be_running_lock.acquire(blocking=1) if not self.monitor_connection.running: - should_be_running_lock.release() + self.monitor_connection.should_be_running_lock.release() what = "call monitor_connection.start_monitor()" self.monitor_connection.start_monitor() else: - should_be_running_lock.release() + self.monitor_connection.should_be_running_lock.release() except Exception as e: print("SEVERE: failed to " + what + " : " + str(e)) traceback.print_exc() @@ -398,18 +396,17 @@ def start_monitor(self, regex=".*"): def stop_monitor(self): if self.queue != None: - global should_be_running_lock what = "" try: if self.monitor_connection == None: return - should_be_running_lock.acquire(blocking=1) + self.monitor_connection.should_be_running_lock.acquire(blocking=1) if not self.monitor_connection.running: - should_be_running_lock.release() + self.monitor_connection.should_be_running_lock.release() what = "call monitor_connection.stop_monitor()" self.monitor_connection.stop_monitor() else: - should_be_running_lock.release() + self.monitor_connection.should_be_running_lock.release() except Exception as e: print("SEVERE: failed to " + what + " : " + str(e)) traceback.print_exc() diff --git a/at_client/connections/atmonitorconnection.py b/at_client/connections/atmonitorconnection.py index a0d61ca..e2413d4 100644 --- a/at_client/connections/atmonitorconnection.py +++ b/at_client/connections/atmonitorconnection.py @@ -9,7 +9,6 @@ from .notification.atevents import AtEvent, AtEventType from ..util.syncdecorator import synchronized from ..util.timeutil import TimeUtil -from ..util.atconstants import * from .address import Address from .atsecondaryconnection import AtSecondaryConnection import queue @@ -24,44 +23,47 @@ def __init__(self, queue:queue.Queue, atsign:AtSign, address: Address, context:s self.queue = queue self.regex = regex self._verbose = verbose + self.should_be_running_lock = threading.Lock() + self.running_lock = threading.Lock() super().__init__(address, context, verbose) self._last_heartbeat_sent_time = TimeUtil.current_time_millis() self._last_heartbeat_ack_time = TimeUtil.current_time_millis() self._heartbeat_interval_millis = 30000 self.start_heart_beat() - + def start_heart_beat(self): - threading.Thread(target=self._start_heart_beat).start() + # Daemon: an abandoned connection's heartbeat must not keep the process alive. + self._heartbeat_thread = threading.Thread(target=self._start_heart_beat, daemon=True) + self._heartbeat_thread.start() def _start_heart_beat(self): - global should_be_running_lock while True: - should_be_running_lock.acquire() + self.should_be_running_lock.acquire() if self.should_be_running: - should_be_running_lock.release() + self.should_be_running_lock.release() if (not self.running) or (self._last_heartbeat_sent_time - self._last_heartbeat_ack_time >= self._heartbeat_interval_millis): try: print("Monitor heartbeats not being received") self.stop_monitor() wait_start_time = TimeUtil.current_time_millis() - running_lock.acquire(blocking=1) + self.running_lock.acquire(blocking=1) entered = False print((TimeUtil.current_time_millis() - wait_start_time) < 5000) while self.running and ((TimeUtil.current_time_millis() - wait_start_time) < 5000): entered = True - running_lock.release() + self.running_lock.release() print("Wait 5 seconds for monitor to stop") try: time.sleep(1) except Exception as ignore: pass if not entered: - running_lock.release() + self.running_lock.release() entered = False - running_lock.acquire(blocking=1) + self.running_lock.acquire(blocking=1) if self.running: print("Monitor thread has not stopped, but going to start another one anyway") - running_lock.release() + self.running_lock.release() self.start_monitor() except Exception as e: print("Monitor restart failed "+ str(e)) @@ -74,7 +76,7 @@ def _start_heart_beat(self): # Can't do anything, the heartbeat loop will take care of restarting the monitor connection pass else: - should_be_running_lock.release() + self.should_be_running_lock.release() try: time.sleep(self._heartbeat_interval_millis / 6000) # 6 * 1000 (from ms to s) except Exception as ignore: @@ -83,33 +85,33 @@ def _start_heart_beat(self): def start_monitor(self): self._last_heartbeat_sent_time = self._last_heartbeat_ack_time = TimeUtil.current_time_millis() - should_be_running_lock.acquire(blocking=1) + self.should_be_running_lock.acquire(blocking=1) self.should_be_running = True - should_be_running_lock.release() + self.should_be_running_lock.release() - running_lock.acquire(blocking=1) + self.running_lock.acquire(blocking=1) if not self.running: self.running = True - running_lock.release() + self.running_lock.release() if not self._connected: try: self._connect() except Exception as e: print("startMonitor failed to connect to secondary : " + str(e)) traceback.print_exc() - running_lock.acquire(blocking=1) + self.running_lock.acquire(blocking=1) self.running = False - running_lock.release() + self.running_lock.release() return False self._run() else: - running_lock.release() + self.running_lock.release() return True def stop_monitor(self): - should_be_running_lock.acquire(blocking=1) + self.should_be_running_lock.acquire(blocking=1) self.should_be_running = False - should_be_running_lock.release() + self.should_be_running_lock.release() self._last_heartbeat_sent_time = self._last_heartbeat_ack_time = TimeUtil.current_time_millis() self.disconnect() @@ -132,9 +134,9 @@ def _run(self): self.execute_command(command=monitor_cmd, retry_on_exception=True, read_the_response=False) print("Monitor started on " + str(self.atsign.to_string())) entered = False - should_be_running_lock.acquire(blocking=1) + self.should_be_running_lock.acquire(blocking=1) while self.should_be_running: - should_be_running_lock.release() + self.should_be_running_lock.release() entered = True first = False what = "read from connection" @@ -200,26 +202,26 @@ def _run(self): at_event = AtEvent(event_type, event_data) self.queue.put(at_event) - should_be_running_lock.acquire(blocking=1) + self.should_be_running_lock.acquire(blocking=1) entered = False if not entered: - should_be_running_lock.release() + self.should_be_running_lock.release() entered = False except Exception as e: traceback.print_exc() - should_be_running_lock.acquire(blocking=1) + self.should_be_running_lock.acquire(blocking=1) if not self.should_be_running: - should_be_running_lock.release() + self.should_be_running_lock.release() else: - should_be_running_lock.release() + self.should_be_running_lock.release() print("Monitor failed to " + what + " : " + str(e)) traceback.print_exc() print("Monitor ending. Monitor heartbeat thread should restart the monitor shortly") self.disconnect() finally: - running_lock.acquire(blocking=1) + self.running_lock.acquire(blocking=1) self.running = False - running_lock.release() + self.running_lock.release() self.disconnect() diff --git a/test/monitor_heartbeat_test.py b/test/monitor_heartbeat_test.py new file mode 100644 index 0000000..a80e91c --- /dev/null +++ b/test/monitor_heartbeat_test.py @@ -0,0 +1,37 @@ +import unittest +from queue import Queue + +from at_client.common import AtSign +from at_client.connections.address import Address +from at_client.connections.atmonitorconnection import AtMonitorConnection + + +class MonitorHeartbeatTest(unittest.TestCase): + """Network-free tests for heartbeat thread lifecycle and per-instance locks.""" + + def _connection(self): + return AtMonitorConnection(queue=Queue(), atsign=AtSign("@alice"), + address=Address("localhost", 64), verbose=False) + + def test_heartbeat_thread_is_daemon(self): + """A non-daemon heartbeat thread would keep the process alive forever.""" + conn = self._connection() + self.assertTrue(conn._heartbeat_thread.daemon) + + def test_locks_are_per_instance(self): + """Module-level locks made every monitor in a process share one lock pair.""" + a = self._connection() + b = self._connection() + self.assertIsNot(a.should_be_running_lock, b.should_be_running_lock) + self.assertIsNot(a.running_lock, b.running_lock) + + def test_one_connections_lock_does_not_block_another(self): + a = self._connection() + b = self._connection() + with a.should_be_running_lock: + self.assertTrue(b.should_be_running_lock.acquire(blocking=False)) + b.should_be_running_lock.release() + + +if __name__ == "__main__": + unittest.main() From 74b734fa55b51756fb634d640912ce76b94974ad Mon Sep 17 00:00:00 2001 From: Colin Constable Date: Tue, 7 Jul 2026 12:12:50 -0700 Subject: [PATCH 2/2] fix: make the heartbeat loop stoppable via an Event Aligns with the reference Dart SDK, whose heartbeat is a cancellable Timer (stopHeartbeat() in monitor.dart): the loop now waits on a threading.Event instead of time.sleep, and stop_heart_beat() sets it, so the thread exits promptly rather than only at process exit. Adds a test that the thread ends within the join timeout after stop_heart_beat(). --- at_client/connections/atmonitorconnection.py | 16 ++++++++++------ test/monitor_heartbeat_test.py | 8 ++++++++ 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/at_client/connections/atmonitorconnection.py b/at_client/connections/atmonitorconnection.py index e2413d4..69a4188 100644 --- a/at_client/connections/atmonitorconnection.py +++ b/at_client/connections/atmonitorconnection.py @@ -33,11 +33,18 @@ def __init__(self, queue:queue.Queue, atsign:AtSign, address: Address, context:s def start_heart_beat(self): # Daemon: an abandoned connection's heartbeat must not keep the process alive. + # The stop event makes the loop's wait interruptible, so stop_heart_beat() + # takes effect promptly (the Dart SDK's heartbeat is a cancellable Timer). + self._heartbeat_stop_event = threading.Event() self._heartbeat_thread = threading.Thread(target=self._start_heart_beat, daemon=True) self._heartbeat_thread.start() - + + def stop_heart_beat(self): + """Stop this connection's heartbeat/restart loop.""" + self._heartbeat_stop_event.set() + def _start_heart_beat(self): - while True: + while not self._heartbeat_stop_event.is_set(): self.should_be_running_lock.acquire() if self.should_be_running: self.should_be_running_lock.release() @@ -77,10 +84,7 @@ def _start_heart_beat(self): pass else: self.should_be_running_lock.release() - try: - time.sleep(self._heartbeat_interval_millis / 6000) # 6 * 1000 (from ms to s) - except Exception as ignore: - pass + self._heartbeat_stop_event.wait(self._heartbeat_interval_millis / 6000) # 6 * 1000 (from ms to s) def start_monitor(self): self._last_heartbeat_sent_time = self._last_heartbeat_ack_time = TimeUtil.current_time_millis() diff --git a/test/monitor_heartbeat_test.py b/test/monitor_heartbeat_test.py index a80e91c..80f05ca 100644 --- a/test/monitor_heartbeat_test.py +++ b/test/monitor_heartbeat_test.py @@ -32,6 +32,14 @@ def test_one_connections_lock_does_not_block_another(self): self.assertTrue(b.should_be_running_lock.acquire(blocking=False)) b.should_be_running_lock.release() + def test_stop_heart_beat_ends_the_thread(self): + """The heartbeat loop must exit promptly when stopped, not at process exit.""" + conn = self._connection() + self.assertTrue(conn._heartbeat_thread.is_alive()) + conn.stop_heart_beat() + conn._heartbeat_thread.join(timeout=2) + self.assertFalse(conn._heartbeat_thread.is_alive()) + if __name__ == "__main__": unittest.main()