Merge pull request #529 from atsign-foundation/fix/monitor-heartbeat-lifecycle

fix: daemonize the monitor heartbeat thread and scope its locks per instance

Author: cconstab

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()

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()
 

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()