From e08cc23c3d9d826d39a2bd3a73e48cfb84d628c5 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 23:57:34 +0000 Subject: [PATCH 1/5] Test that updates do not overlap and one chain of checks is scheduled --- .../tests/test_datafile_update_guard.py | 167 ++++++++++++++++++ 1 file changed, 167 insertions(+) create mode 100644 pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py new file mode 100644 index 00000000..4e662ac6 --- /dev/null +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py @@ -0,0 +1,167 @@ + +""" +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" + + +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(10) + 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(10)) + second = self.service.update_data_file(self.data_file) + release.set() + first.join(10) + + 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() From d2f6eae61f4fd7403c6a6106f40b77b1db67d976 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 23:58:47 +0000 Subject: [PATCH 2/5] Make the already-updating guard real and keep one pending check per data file An update returns False while another is running for the same data file, with the flag set under a lock, and the flag is cleared when an update ends, however it ends. The pending scheduled check is cancelled before the next is scheduled, so a manual update no longer leaves two chains of checks. --- .../datafile_update_service.py | 26 ++++++++++++++++--- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index c4c6b047..fa8f648c 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/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): From 0270803aa25ef041341421fd4fc87f7f0028c84b Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 10:33:29 +0000 Subject: [PATCH 3/5] Name the guard test's wait as a constant --- .../tests/test_datafile_update_guard.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py index 4e662ac6..8bd6bcb0 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py @@ -16,6 +16,8 @@ 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: @@ -106,7 +108,7 @@ def get(url, headers=None, **kwargs): calls.append(url) if len(calls) == 1: started.set() - release.wait(10) + release.wait(WAIT_SECONDS) return answer(200) results = [] @@ -115,10 +117,10 @@ def get(url, headers=None, **kwargs): target=lambda: results.append( self.service.update_data_file(self.data_file))) first.start() - self.assertTrue(started.wait(10)) + self.assertTrue(started.wait(WAIT_SECONDS)) second = self.service.update_data_file(self.data_file) release.set() - first.join(10) + first.join(WAIT_SECONDS) self.assertFalse(second) self.assertEqual(1, len(calls)) From 7709feb54d7efdd0098b8b9a0dbeee455c95cb20 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Fri, 13 Nov 2026 17:56:10 +0000 Subject: [PATCH 4/5] Let a second update of a data file wait for the first and make its own check, and take the same lock in update_from_memory --- .../datafile_update_service.py | 48 ++++++---- .../tests/test_datafile_update_guard.py | 88 +++++++++++++++---- 2 files changed, 101 insertions(+), 35 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index c3af1eed..ec1916bc 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -93,10 +93,21 @@ 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. + # Held while a data file's update lock is made and while its pending + # check is replaced, so two threads cannot make two locks or leave + # two checks pending. self._lock = threading.Lock() + # The update lock of each data file, by the data file's id. + self._update_locks = {} + + def _update_lock(self, data_file): + # One update of a data file runs at a time. A second one waits for + # the first and then makes its own check, so every check ends and is + # reported. Re-entrant, so that an update complete callback can start + # an update of the same data file. + with self._lock: + return self._update_locks.setdefault( + id(data_file), threading.RLock()) def __emit(self, status, data_file): # A callback that raises stops the callbacks after it, as .NET's service @@ -123,16 +134,15 @@ def update_data_file(self, data_file): if not self.__is_registered(data_file): return False - # Exit if already updating. - with self._lock: - if data_file.updating: - return False + # The file system watcher reads updating to leave a changed file + # alone while an update runs. + with self._update_lock(data_file): + was_updating = data_file.updating data_file.updating = True - - try: - return self._update_data_file(data_file) - finally: - data_file.updating = False + try: + return self._update_data_file(data_file) + finally: + data_file.updating = was_updating def _update_data_file(self, data_file): @@ -278,8 +288,6 @@ def _update_data_file(self, data_file): status = self._refresh(data_file) - data_file.updating = False - self.__emit(status, data_file) self.check_next_update(data_file) @@ -307,6 +315,13 @@ def update_from_memory(self, data_file, data): if not self.__is_registered(data_file): return UpdateStatus.AUTO_UPDATE_NOT_NEEDED + # Waits for an update of the same data file that is running, as a + # second update does. + with self._update_lock(data_file): + return self._update_from_memory(data_file, data) + + def _update_from_memory(self, data_file, data): + was_updating = data_file.updating if data_file.download and getattr(data_file, "path", None): data_file.updating = True try: @@ -314,7 +329,7 @@ def update_from_memory(self, data_file, data): if status is None: status = self._refresh(data_file) finally: - data_file.updating = False + data_file.updating = was_updating self.__emit(status, data_file) return status @@ -324,7 +339,7 @@ def update_from_memory(self, data_file, data): try: status = self._refresh(data_file, from_file=False) finally: - data_file.updating = False + data_file.updating = was_updating if status != UpdateStatus.AUTO_UPDATE_SUCCESS: data_file.data = previous @@ -402,7 +417,6 @@ def _refresh(self, data_file, from_file=True): def _update_failed(self, data_file, status): # An update that ends without new data. The next check is the next # scheduled one, never an immediate repeat of a check on start-up. - data_file.updating = False self.__emit(status, data_file) self.check_next_update(data_file, run_after_error=True) return False diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py index 8bd6bcb0..f231342c 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py @@ -1,8 +1,8 @@ """ -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. +An update of a data file waits for one that is running and then makes its own +check, so every check ends and is reported, and one data file has one chain +of scheduled checks. """ import threading @@ -18,6 +18,9 @@ URL = "https://example.com/update" # The longest a test waits for another thread. WAIT_SECONDS = 10 +# How long a test gives an update that should be waiting to show that it is. +# An update that does not wait has made its request and ended well within it. +STILL_WAITING_SECONDS = 0.5 class FakeEngine: @@ -51,7 +54,7 @@ def get_next_update(self): return None def refresh(self): - pass + self.refreshes = getattr(self, "refreshes", 0) + 1 def answer(status): @@ -99,32 +102,81 @@ def setUp(self): 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): + def start_blocked_update(self, calls, statuses): + # Starts an update whose request waits until the event returned is + # set, and returns once that request has been made. Each request + # adds whether the data file was updating to calls, and each report + # adds its status to statuses. started = threading.Event() release = threading.Event() - calls = [] def get(url, headers=None, **kwargs): - calls.append(url) + calls.append(self.data_file.updating) if len(calls) == 1: started.set() release.wait(WAIT_SECONDS) return answer(200) + self.service.on_complete( + lambda status, data_file: statuses.append(status)) + patch = mock.patch.object(requests, "get", get) + patch.start() + self.addCleanup(patch.stop) + # A test that fails before it releases the request still lets the + # thread end. + self.addCleanup(release.set) + first = threading.Thread( + target=self.service.update_data_file, args=[self.data_file]) + first.start() + self.assertTrue(started.wait(WAIT_SECONDS)) + return first, release + + def test_a_second_update_waits_for_the_first_then_makes_its_own_check(self): + calls = [] + statuses = [] 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) + first, release = self.start_blocked_update(calls, statuses) + second = threading.Thread( + target=lambda: results.append( + self.service.update_data_file(self.data_file))) + second.start() + second.join(STILL_WAITING_SECONDS) + self.assertTrue(second.is_alive()) self.assertEqual(1, len(calls)) + release.set() + first.join(WAIT_SECONDS) + second.join(WAIT_SECONDS) + + self.assertEqual(2, len(calls)) self.assertEqual([True], results) + self.assertEqual(2, len(statuses)) + + def test_updating_is_set_while_an_update_runs(self): + calls = [] + first, release = self.start_blocked_update(calls, []) + self.assertTrue(self.data_file.updating) + release.set() + first.join(WAIT_SECONDS) + self.assertEqual([True], calls) + self.assertFalse(self.data_file.updating) + + def test_an_update_from_memory_waits_for_a_running_update(self): + first, release = self.start_blocked_update([], []) + results = [] + memory = threading.Thread( + target=lambda: results.append( + self.service.update_from_memory(self.data_file, b"given"))) + memory.start() + memory.join(STILL_WAITING_SECONDS) + self.assertTrue(memory.is_alive()) + self.assertEqual(0, getattr(self.data_file, "refreshes", 0)) + release.set() + first.join(WAIT_SECONDS) + memory.join(WAIT_SECONDS) + + self.assertEqual(2, self.data_file.refreshes) + self.assertEqual(b"given", self.data_file.data) + self.assertEqual(1, len(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)): From 2687dcb2653b69651e3c3a4821e3e0fd2ed8c7c9 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Fri, 13 Nov 2026 22:31:13 +0000 Subject: [PATCH 5/5] Hold the update lock in the file system watcher's refresh, so an update waits for it --- .../datafile_update_service.py | 26 ++++++++--- .../tests/test_datafile_update_guard.py | 46 +++++++++++++++++++ 2 files changed, 65 insertions(+), 7 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index ec1916bc..3ff40401 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -101,10 +101,10 @@ def __init__(self, pipeline = None): self._update_locks = {} def _update_lock(self, data_file): - # One update of a data file runs at a time. A second one waits for - # the first and then makes its own check, so every check ends and is - # reported. Re-entrant, so that an update complete callback can start - # an update of the same data file. + # One update or file system watcher refresh of a data file runs at a + # time. A second update waits for the first and then makes its own + # check, so every check ends and is reported. Re-entrant, so that an + # update complete callback can start an update of the same data file. with self._lock: return self._update_locks.setdefault( id(data_file), threading.RLock()) @@ -632,14 +632,26 @@ def track_local_data_file_changes(self, data_file): # told by the operating system and say nothing whilst a # file is untouched, and this check stands in for that. return - if data_file.updating: + lock = self._update_lock(data_file) + if data_file.updating or not lock.acquire(blocking=False): # A newer file found whilst an update is being applied is # reported as in progress, and the time is not taken, so # the next check looks at the file again. self.__emit(UpdateStatus.AUTO_UPDATE_IN_PROGRESS, data_file) return - data_file.last_changed_time = changed_time - self.__emit(self._refresh(data_file), data_file) + # The refresh holds the data file's update lock, so an update + # asked for meanwhile waits for it, as it waits for another + # update. + try: + data_file.updating = True + try: + data_file.last_changed_time = changed_time + status = self._refresh(data_file) + finally: + data_file.updating = False + self.__emit(status, data_file) + finally: + lock.release() finally: if self.__is_registered(data_file): t = threading.Timer( diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py index f231342c..3b4c56ce 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_guard.py @@ -5,6 +5,8 @@ of scheduled checks. """ +import os +import tempfile import threading import unittest from unittest import mock @@ -178,6 +180,50 @@ def test_an_update_from_memory_waits_for_a_running_update(self): self.assertEqual(b"given", self.data_file.data) self.assertEqual(1, len(results)) + def test_an_update_waits_for_a_file_system_watcher_refresh(self): + # A data file on disk that the watcher finds changed, whose refresh + # waits until the test lets it go. + handle, path = tempfile.mkstemp() + os.close(handle) + self.addCleanup(os.remove, path) + self.data_file.path = path + self.data_file.last_changed_time = 0 + refreshing = threading.Event() + release = threading.Event() + self.addCleanup(release.set) + + def refresh(): + refreshing.set() + release.wait(WAIT_SECONDS) + + self.data_file.refresh = refresh + calls = [] + patch = mock.patch.object( + requests, "get", + lambda *a, **k: calls.append(self.data_file.updating) or + answer(304)) + patch.start() + self.addCleanup(patch.stop) + + watcher = threading.Thread( + target=self.service.track_local_data_file_changes, + args=[self.data_file]) + watcher.start() + self.assertTrue(refreshing.wait(WAIT_SECONDS)) + self.assertTrue(self.data_file.updating) + update = threading.Thread( + target=self.service.update_data_file, args=[self.data_file]) + update.start() + update.join(STILL_WAITING_SECONDS) + self.assertTrue(update.is_alive()) + self.assertEqual([], calls) + release.set() + watcher.join(WAIT_SECONDS) + update.join(WAIT_SECONDS) + + self.assertEqual([True], calls) + self.assertFalse(self.data_file.updating) + 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))