Multigrid controller can now passively request the shutdown of all watchers and analysers, and finalisation of all rsyncers. Each of these is given a callback that checks the state of all relevant threads before labelling the multigrid controller as dormant once all threads have shutdown

Author: d-j-hatton

src/murfey/client/analyser.py

@@ -404,6 +404,7 @@ def _analyse(self):
                     )
                     self.post_transfer(transferred_file)
             self.queue.task_done()
+        self.notify(final=True)
 
     def _xml_file(self, data_file: Path) -> Path:
         if not self._environment:
@@ -432,6 +433,10 @@ def start(self):
         logger.info(f"Analyser thread starting for {self}")
         self.thread.start()
 
+    def request_stop(self):
+        self._stopping = True
+        self._halt_thread = True
+
     def stop(self):
         logger.debug("Analyser thread stop requested")
         self._stopping = True

src/murfey/client/multigrid_control.py

@@ -100,6 +100,25 @@ def __post_init__(self):
     def _multigrid_watcher_finalised(self):
         self.multigrid_watcher_active = False
 
+    def dormancy_check(self):
+        if not self.multigrid_watcher_active:
+            if (
+                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()
+                )
+            ):
+                self.dormant = True
+
+    def finalise(self):
+        for a in self.analysers.values():
+            a.request_stop()
+        for w in self._environment.watchers.values():
+            w.request_stop()
+        for p in self.rsync_processes.keys():
+            self._finalise_rsyncer(p)
+
     def _start_rsyncer_multigrid(
         self,
         source: Path,
@@ -170,7 +189,9 @@ def _rsyncer_stopped(self, source: Path, explicit_stop: bool = False):
     def _finalise_rsyncer(self, source: Path):
         finalise_thread = threading.Thread(
             name=f"Controller finaliser thread ({source})",
-            target=self.rsync_processes[source].finalise,
+            target=partial(
+                self.rsync_processes[source].finalise, callback=self.dormancy_check
+            ),
             kwargs={"thread": False},
             daemon=True,
         )
@@ -302,6 +323,7 @@ def rsync_result(update: RSyncerUpdate):
                 )
             else:
                 self.analysers[source].subscribe(self._data_collection_form)
+            self.analysers[source].subscribe(self.dormancy_check, final=True)
             self.analysers[source].start()
             if transfer:
                 self.rsync_processes[source].subscribe(self.analysers[source].enqueue)
@@ -341,6 +363,9 @@ def _rsync_update_converter(p: Path) -> None:
                     ),
                     secondary=True,
                 )
+                self._environment.watchers[source].subscribe(
+                    self.dormancy_check, final=True
+                )
                 self._environment.watchers[source].start()
 
     def _data_collection_form(self, response: dict):

src/murfey/client/rsync.py

@@ -15,7 +15,7 @@
 import time
 from enum import Enum
 from pathlib import Path
-from typing import Callable, List, NamedTuple
+from typing import Awaitable, Callable, List, NamedTuple
 from urllib.parse import ParseResult
 
 from murfey.client.tui.status_bar import StatusBar
@@ -75,6 +75,7 @@ def __init__(
         self._local = local
         self._server_url = server_url
         self._notify = notify
+        self._finalised = False
 
         # Set rsync destination
         if local:
@@ -181,7 +182,11 @@ def stop(self):
             self.thread.join()
         logger.debug("RSync thread stop completed")
 
-    def finalise(self, thread: bool = True):
+    def finalise(
+        self,
+        thread: bool = True,
+        callback: Callable[..., Awaitable[None] | None] | None = None,
+    ):
         self.stop()
         self._remove_files = True
         self._notify = False
@@ -196,6 +201,9 @@ def finalise(self, thread: bool = True):
             self.stop()
         else:
             self._transfer(list(self._basepath.glob("**/*")))
+        self._finalised = True
+        if callback:
+            callback()
 
     def enqueue(self, file_path: Path):
         if not self._stopping:

src/murfey/client/watchdir.py

@@ -63,6 +63,10 @@ def start(self):
         log.info(f"DirWatcher thread starting for {self}")
         self.thread.start()
 
+    def request_stop(self):
+        self._stopping = True
+        self._halt_thread = True
+
     def stop(self):
         log.debug("DirWatcher thread stop requested")
         self._stopping = True
@@ -90,6 +94,7 @@ def _process(self):
                 modification_time=modification_time, transfer_all=self._transfer_all
             )
             time.sleep(15)
+        self.notify(final=True)
 
     def scan(self, modification_time: float | None = None, transfer_all: bool = False):
         """