diff --git a/at_client/atclient.py b/at_client/atclient.py index 7f0feac..4ca5ab0 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=".*", last_received_time=0): if self.queue != None: - global should_be_running_lock what = "" try: if self.monitor_connection == None: @@ -385,13 +383,13 @@ def start_monitor(self, regex=".*", last_received_time=0): verbose=self.verbose, regex=regex, last_received_time=last_received_time) 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() @@ -400,18 +398,17 @@ def start_monitor(self, regex=".*", last_received_time=0): 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 737ea95..cb909cb 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 @@ -26,44 +25,54 @@ def __init__(self, queue: queue.Queue, atsign: AtSign, address: Address, self.regex = regex self.last_received_time = last_received_time 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. + # 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): - global should_be_running_lock - while True: - should_be_running_lock.acquire() + while not self._heartbeat_stop_event.is_set(): + 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)) @@ -76,42 +85,39 @@ 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() - try: - time.sleep(self._heartbeat_interval_millis / 6000) # 6 * 1000 (from ms to s) - except Exception as ignore: - pass + self.should_be_running_lock.release() + 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() - 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() @@ -138,9 +144,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" @@ -206,26 +212,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..80f05ca --- /dev/null +++ b/test/monitor_heartbeat_test.py @@ -0,0 +1,45 @@ +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() + + 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()