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

@@ -93,6 +93,21 @@ def __init__(self, pipeline = None):
         if pipeline != None:
             self.pipeline = pipeline
         self.on_complete_methods = []
+        # 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 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())
 
     def __emit(self, status, data_file):
         # A callback that raises stops the callbacks after it, as .NET's service
@@ -119,10 +134,17 @@ def update_data_file(self, data_file):
         if not self.__is_registered(data_file):
             return False
 
-        # exit if already updating
+        # 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 = was_updating
 
-        if data_file.updating:
-            pass
+    def _update_data_file(self, data_file):
 
         self._log("info", "Checking for update")
 
@@ -266,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)
 
@@ -295,14 +315,21 @@ 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:
                 status = self._write_data_file(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
 
@@ -312,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
@@ -390,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
@@ -524,7 +550,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):
@@ -600,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(