Cancel the running watchdog timer when it is started again

pyhik/hikvision.py

@@ -918,6 +918,10 @@ def alert_stream(self, reset_event, kill_event):
                     # We need to reset the connection.
                     raise ValueError('Watchdog failed.')
 
+                # Some devices end the response normally from time to time.
+                _LOGGING.debug('%s Stream closed by device, reconnecting.',
+                               self.name)
+
             # RequestException is the base class of every requests error and
             # covers read timeouts, which is how a silently dropped connection
             # surfaces. Anything not caught here kills the stream thread and

pyhik/watchdog.py

@@ -6,7 +6,7 @@
 Licensed under the MIT license.
 """
 
-from threading import Timer
+from threading import Lock, Timer
 
 
 class Watchdog(object):
@@ -17,22 +17,54 @@ def __init__(self, timeout, handler):
         self.time = timeout
         self.handler = handler
         self._timer = None
+        # Identifies the current timer, so an expiry from one that was
+        # replaced or stopped can tell it is stale.
+        self._token = None
+        # start(), pet() and stop() run on the stream thread while expiries
+        # run on timer threads.
+        self._lock = Lock()
         return
 
     def start(self):
-        """ Starts the watchdog timer. """
-        self._timer = Timer(self.time, self.handler)
-        self._timer.daemon = True
-        self._timer.start()
+        """ Starts the watchdog timer, replacing any timer already running. """
+        with self._lock:
+            # Replacing a running timer without cancelling it leaves it
+            # orphaned: it can no longer be pet or stopped, and fires anyway.
+            self._cancel()
+            token = object()
+            self._token = token
+            self._timer = Timer(self.time, self._expire, args=(token,))
+            self._timer.daemon = True
+            self._timer.start()
         return
 
     def pet(self):
         """ Reset watchdog timer. """
-        self.stop()
         self.start()
         return
 
     def stop(self):
         """ Stops the watchdog timer. """
+        with self._lock:
+            self._cancel()
+        return
+
+    def _cancel(self):
+        """ Cancel the current timer. Callers hold the lock. """
         if self._timer is not None:
             self._timer.cancel()
+        self._timer = None
+        self._token = None
+
+    def _expire(self, token):
+        """ Call the handler, unless this timer was replaced or stopped. """
+        with self._lock:
+            # cancel() cannot stop a timer that has already fired, so a timer
+            # replaced or stopped at the moment it expired still gets here.
+            if token is not self._token:
+                return
+            self._timer = None
+            self._token = None
+        # Outside the lock, so a slow handler cannot hold up pet() and stop(),
+        # and the handler may call them itself.
+        self.handler()

test/test_hikvision.py

@@ -1002,6 +1002,37 @@ def unreachable(*args, **kwargs):
         # The backoff it was sitting in is 5s, and grows with every failure.
         self.assertLess(time.monotonic() - started, 4)
 
+    def test_stream_ended_by_device_leaves_one_watchdog_timer(self):
+        """Some NVRs end the response normally and the loop reconnects at
+        once. Each connection starts the watchdog; a timer left behind by the
+        previous one would reset the stream 300s later (#149)."""
+        camera = make_camera()
+        camera.hik_request_stream = MagicMock()
+        kill_event = threading.Event()
+        connections = []
+
+        def lines():
+            connections.append(1)
+            if len(connections) == 2:
+                kill_event.set()
+            return iter([])
+
+        camera.hik_request_stream.get.return_value = MagicMock(
+            status_code=requests.codes.ok, iter_lines=lines)
+
+        timers = []
+
+        def new_timer(*args, **kwargs):
+            timers.append(MagicMock())
+            return timers[-1]
+
+        with patch("pyhik.watchdog.Timer", side_effect=new_timer):
+            camera.alert_stream(threading.Event(), kill_event)
+
+        self.assertEqual(len(timers), 2)
+        for timer in timers:
+            timer.cancel.assert_called()
+
     def test_a_killed_stream_does_not_reconnect(self):
         """The kill event can arrive while the thread is between attempts.
         Reconnecting then leaves a thread nobody is going to join again."""

test/test_watchdog.py

@@ -0,0 +1,110 @@
+"""Tests for the watchdog timer."""
+
+import threading
+import time
+import unittest
+
+from pyhik.watchdog import Watchdog
+
+# Short enough to keep the tests fast, long enough for a loaded CI runner to
+# get through a few calls before it fires.
+TIMEOUT = 0.2
+
+
+class WatchdogTestCase(unittest.TestCase):
+    def setUp(self):
+        self.fired = threading.Event()
+        self.calls = []
+
+        def handler():
+            self.calls.append(time.monotonic())
+            self.fired.set()
+
+        self.watchdog = Watchdog(TIMEOUT, handler)
+        self.addCleanup(self.watchdog.stop)
+
+    def test_fires_after_timeout(self):
+        self.watchdog.start()
+        self.assertTrue(self.fired.wait(TIMEOUT * 10))
+        self.assertEqual(len(self.calls), 1)
+
+    def test_restart_replaces_the_running_timer(self):
+        """Starting again must cancel the running timer. Otherwise it is
+        orphaned, can no longer be pet or stopped, and fires anyway (#149)."""
+        self.watchdog.start()
+        self.watchdog.start()
+        self.assertTrue(self.fired.wait(TIMEOUT * 10))
+        time.sleep(TIMEOUT * 2)
+        self.assertEqual(len(self.calls), 1)
+
+    def test_stop_after_restart_cancels_every_timer(self):
+        self.watchdog.start()
+        self.watchdog.start()
+        self.watchdog.stop()
+        time.sleep(TIMEOUT * 3)
+        self.assertEqual(self.calls, [])
+
+    def test_pet_postpones_expiry(self):
+        self.watchdog.start()
+        for _ in range(3):
+            time.sleep(TIMEOUT / 4)
+            self.watchdog.pet()
+        self.assertEqual(self.calls, [])
+        self.assertTrue(self.fired.wait(TIMEOUT * 10))
+        self.assertEqual(len(self.calls), 1)
+
+    def test_expiry_of_a_replaced_timer_is_ignored(self):
+        """cancel() cannot stop a timer that has already fired, so an expiry
+        can still arrive from a timer that was just replaced."""
+        self.watchdog.start()
+        stale = self.watchdog._token
+        self.watchdog.start()
+        self.watchdog._expire(stale)
+        self.assertEqual(self.calls, [])
+
+    def test_expiry_after_stop_is_ignored(self):
+        self.watchdog.start()
+        stale = self.watchdog._token
+        self.watchdog.stop()
+        self.watchdog._expire(stale)
+        self.assertEqual(self.calls, [])
+
+    def test_handler_may_stop_the_watchdog(self):
+        """HikCamera.watchdog_handler calls stop() from the timer thread."""
+        returned = threading.Event()
+
+        def handler():
+            self.watchdog.stop()
+            returned.set()
+
+        self.watchdog.handler = handler
+        self.watchdog.start()
+        self.assertTrue(returned.wait(TIMEOUT * 10))
+
+    def test_slow_handler_does_not_block_pet_and_stop(self):
+        """Watchdog is public, so a handler may take its time. The stream
+        thread pets and stops the watchdog and must not wait on it."""
+        entered = threading.Event()
+        release = threading.Event()
+
+        def handler():
+            entered.set()
+            release.wait(TIMEOUT * 20)
+
+        self.addCleanup(release.set)
+        self.watchdog.handler = handler
+        self.watchdog.start()
+        self.assertTrue(entered.wait(TIMEOUT * 10))
+        returned = threading.Event()
+
+        def pet_then_stop():
+            self.watchdog.pet()
+            self.watchdog.stop()
+            returned.set()
+
+        threading.Thread(target=pet_then_stop, daemon=True).start()
+        self.assertTrue(returned.wait(TIMEOUT * 5))
+
+
+if __name__ == '__main__':
+    unittest.main()