Make the already-updating guard real and keep one pending check per data file

pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py

@@ -76,6 +76,10 @@ def __init__(self, pipeline = None):
         if pipeline != None:
             self.pipeline = pipeline
         self.on_complete_methods = []
+        # Held while the updating flag and the pending check of a data file
+        # are read and changed, so two threads cannot both start an update
+        # or both schedule the next check.
+        self._lock = threading.Lock()
 
     def __emit(self, status, data_file):
         for method in self.on_complete_methods:
@@ -94,10 +98,18 @@ def update_data_file(self, data_file):
         if not self.__is_registered(data_file):
             return False
 
-        # exit if already updating
+        # Exit if already updating.
+        with self._lock:
+            if data_file.updating:
+                return False
+            data_file.updating = True
+
+        try:
+            return self._update_data_file(data_file)
+        finally:
+            data_file.updating = False
 
-        if data_file.updating:
-            pass
+    def _update_data_file(self, data_file):
 
         # Get the update url
 
@@ -248,7 +260,13 @@ def check_next_update(self, data_file, run_after_error = False):
             t = threading.Timer(interval, self.async_update_data_file, args=[data_file])
             t.daemon = True
             t.name = "AutoUpdate - Scheduled"
-            data_file.update_timer = t
+            # One check is pending for a data file, so the one already
+            # waiting is cancelled, for example after a manual update.
+            with self._lock:
+                previous = getattr(data_file, "update_timer", None)
+                if previous is not None:
+                    previous.cancel()
+                data_file.update_timer = t
             t.start()
 
     def register_pipeline(self, pipeline):

pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py

@@ -0,0 +1,169 @@
+
+"""
+An update does not start while another is running, and one data file has one
+chain of scheduled checks, as the data updates specification asks of a
+refresh mechanism that MUST be thread safe.
+"""
+
+import threading
+import unittest
+from unittest import mock
+
+import requests
+
+from fiftyone_pipeline_engines.datafile_update_service import DataFileUpdateService
+
+from .test_datafile_update_service import RecordingPipeline
+
+URL = "https://example.com/update"
+# The longest a test waits for another thread.
+WAIT_SECONDS = 10
+
+
+class FakeEngine:
+    datakey = "fake"
+
+
+class MemoryDataFile:
+    """Stands in for a data file that is kept in memory and updated over
+    HTTP."""
+
+    def __init__(self):
+        self.flow_element = FakeEngine()
+        self.identifier = "fake-data"
+        self.update_on_start = False
+        self.auto_update = True
+        self.file_system_watcher = False
+        self.updating = False
+        self.verify_if_modified_since = False
+        self.verify_md5 = False
+        self.decompress = False
+        self.download = False
+        self.create_temp_copy = False
+        self.polling_interval = 30
+        self.update_time_maximum_randomisation = 0
+        self.data = b"old"
+
+    def get_update_url(self):
+        return URL
+
+    def get_next_update(self):
+        return None
+
+    def refresh(self):
+        pass
+
+
+def answer(status):
+    reply = mock.Mock()
+    reply.status_code = status
+    reply.content = b"new"
+    reply.headers = {}
+    return reply
+
+
+class FakeTimer:
+
+    def __init__(self, timers, interval, function, args):
+        self.cancelled = False
+        self.daemon = False
+        self.name = ""
+        self.timers = timers
+        timers.append(self)
+
+    def start(self):
+        pass
+
+    def cancel(self):
+        self.cancelled = True
+
+    @property
+    def live(self):
+        return not self.cancelled
+
+
+class UpdateGuardTests(unittest.TestCase):
+
+    def setUp(self):
+        self.pipeline = RecordingPipeline()
+        self.service = DataFileUpdateService(self.pipeline)
+        self.data_file = MemoryDataFile()
+        self.timers = []
+        timer = mock.patch.object(
+            threading, "Timer",
+            lambda interval, function, args=None: FakeTimer(
+                self.timers, interval, function, args))
+        timer.start()
+        self.addCleanup(timer.stop)
+
+    def live(self):
+        return [timer for timer in self.timers if timer.live]
+
+    def test_an_update_does_not_start_while_another_is_running(self):
+        started = threading.Event()
+        release = threading.Event()
+        calls = []
+
+        def get(url, headers=None, **kwargs):
+            calls.append(url)
+            if len(calls) == 1:
+                started.set()
+                release.wait(WAIT_SECONDS)
+            return answer(200)
+
+        results = []
+        with mock.patch.object(requests, "get", get):
+            first = threading.Thread(
+                target=lambda: results.append(
+                    self.service.update_data_file(self.data_file)))
+            first.start()
+            self.assertTrue(started.wait(WAIT_SECONDS))
+            second = self.service.update_data_file(self.data_file)
+            release.set()
+            first.join(WAIT_SECONDS)
+
+        self.assertFalse(second)
+        self.assertEqual(1, len(calls))
+        self.assertEqual([True], results)
+
+    def test_updating_is_clear_after_an_update_so_the_next_one_runs(self):
+        with mock.patch.object(requests, "get", lambda *a, **k: answer(200)):
+            self.assertTrue(self.service.update_data_file(self.data_file))
+            self.assertFalse(self.data_file.updating)
+            self.assertTrue(self.service.update_data_file(self.data_file))
+
+    def test_updating_is_clear_after_an_error(self):
+        def fail(*args, **kwargs):
+            raise RuntimeError("boom")
+
+        with mock.patch.object(requests, "get", fail):
+            with self.assertRaises(RuntimeError):
+                self.service.update_data_file(self.data_file)
+        self.assertFalse(self.data_file.updating)
+
+    def test_a_manual_update_leaves_one_pending_check(self):
+        self.service.register_data_file(self.data_file)
+        self.assertEqual(1, len(self.live()))
+        with mock.patch.object(requests, "get", lambda *a, **k: answer(304)):
+            self.service.update_data_file(self.data_file)
+        self.assertEqual(2, len(self.timers))
+        self.assertEqual(1, len(self.live()))
+        self.assertTrue(self.timers[0].cancelled)
+
+    def test_polling_carries_on_after_a_manual_update(self):
+        self.service.register_data_file(self.data_file)
+        with mock.patch.object(requests, "get", lambda *a, **k: answer(200)):
+            self.service.update_data_file(self.data_file)
+        self.assertEqual(1, len(self.live()))
+        self.assertIs(self.live()[0], self.data_file.update_timer)
+
+    def test_unregistering_cancels_every_pending_check(self):
+        self.service.register_data_file(self.data_file)
+        with mock.patch.object(requests, "get", lambda *a, **k: answer(200)):
+            self.service.update_data_file(self.data_file)
+        self.service.unregister_data_file(self.data_file)
+        self.assertEqual([], self.live())
+
+
+if __name__ == "__main__":
+    unittest.main()