More checks for skipped files on visit end time (#672)

* More logs on dormancy check
* Only skip files on completion if already skipped

---------

Co-authored-by: Eu Pin Tien <eu-pin.tien@diamond.ac.uk>

Author: stephen-riggs

src/murfey/client/multigrid_control.py

@@ -137,13 +137,20 @@ def is_ready_for_dormancy(self):
             for w in self._environment.watchers.values():
                 if w.is_safe_to_stop():
                     w.stop()
-            return (
-                all(r._finalised for r in self.rsync_processes.values())
-                and not any(a.thread.is_alive() for a in self.analysers.values())
-                and not any(
-                    w.thread.is_alive() for w in self._environment.watchers.values()
-                )
+            rsyncers_finalised = all(
+                r._finalised for r in self.rsync_processes.values()
+            )
+            analysers_alive = any(a.thread.is_alive() for a in self.analysers.values())
+            watchers_alive = any(
+                w.thread.is_alive() for w in self._environment.watchers.values()
+            )
+            log.debug(
+                "Dormancy check: \n"
+                f"  rsyncers: {rsyncers_finalised} \n"
+                f"  analysers: {not analysers_alive} \n"
+                f"  watchers: {not watchers_alive}"
             )
+            return rsyncers_finalised and not analysers_alive and not watchers_alive
         log.debug(f"Multigrid watcher for session {self.session_id} is still active")
         return False
 

src/murfey/client/rsync.py

@@ -79,6 +79,7 @@ def __init__(
         self._notify = notify
         self._finalised = False
         self._end_time = end_time
+        self._finalising = False
 
         self._skipped_files: List[Path] = []
 
@@ -185,7 +186,7 @@ def stop(self):
         if self.thread.is_alive():
             self.queue.put(None)
             self.thread.join()
-        logger.debug("RSync thread stop completed")
+        logger.debug("RSync thread successfully stopped")
 
     def request_stop(self):
         self._stopping = True
@@ -199,7 +200,8 @@ def finalise(
         self.stop()
         self._remove_files = True
         self._notify = False
-        self._end_time = datetime.now()
+        self._end_time = None
+        self._finalising = True
         if thread:
             self.thread = threading.Thread(
                 name=f"RSync finalisation {self._basepath}:{self._remote}",
@@ -284,7 +286,6 @@ def _process(self):
                 continue
 
         self._stop_callback(self._basepath, explicit_stop=self._stopping)
-        logger.info("RSync thread finished")
 
     def _fake_transfer(self, files: list[Path]) -> bool:
         previously_transferred = self._files_transferred
@@ -330,6 +331,9 @@ def _transfer(self, infiles: list[Path]) -> bool:
             ]
             self._skipped_files.extend(set(infiles).difference(set(files)))
             num_skipped_files = len(set(infiles).difference(set(files)))
+        elif self._finalising:
+            files = [f for f in infiles if f.is_file() and f not in self._skipped_files]
+            num_skipped_files = 0
         else:
             files = [f for f in infiles if f.is_file()]
             num_skipped_files = 0