simpler checks for stopped clusters

- join wait thread, if running, to ensure stop callbacks have been called
- log updates to the cluster file
- async waits in load_clusters to give stop events a chance to fire

Author: minrk

ipyparallel/cluster/cluster.py

@@ -511,6 +511,7 @@ def from_file(
     def write_cluster_file(self):
         """Write cluster info to disk for later loading"""
         os.makedirs(os.path.dirname(self.cluster_file), exist_ok=True)
+        self.log.debug(f"Updating {self.cluster_file}")
         with open(self.cluster_file, "w") as f:
             json.dump(self.to_dict(), f)
 
@@ -523,6 +524,14 @@ def remove_cluster_file(self):
         else:
             self.log.debug(f"Removed cluster file: {self.cluster_file}")
 
+    def _is_running(self):
+        """Return if we have any running components"""
+        if self.controller and self.controller.state != 'after':
+            return True
+        if any(es.state != 'after' for es in self.engines.values()):
+            return True
+        return False
+
     def update_cluster_file(self):
         """Update my cluster file
 
@@ -533,9 +542,7 @@ def update_cluster_file(self):
             # setting cluster_file='' disables saving to disk
             return
 
-        if (not self.controller or self.controller.state == 'after') and not any(
-            es.state == 'after' for es in self.engines.values()
-        ):
+        if not self._is_running():
             self.remove_cluster_file()
         else:
             self.write_cluster_file()
@@ -862,10 +869,10 @@ def load_clusters(
         for key, cluster in list(self.clusters.items()):
             # remove stopped clusters
             # but not *new* clusters that haven't started yet
-            if (cluster.controller and cluster.controller.state == 'after') and all(
-                es.state == 'after' for es in cluster.engines.values()
-            ):
-                self.log.info("Removing stopped cluster {key}")
+            # if `cluster.controller` is present
+            # that means it was running at some point
+            if cluster.controller and not cluster._is_running():
+                self.log.info(f"Removing stopped cluster {key}")
                 self.clusters.pop(key)
 
         if profile_dirs is None:

ipyparallel/cluster/launcher.py

@@ -490,6 +490,11 @@ async def join(self, timeout=None):
                 raise TimeoutError(
                     f"Process {self.pid} did not complete in {timeout} seconds."
                 )
+        if getattr(self, '_stop_waiting', None) and getattr(self, "_wait_thread", None):
+            self._stop_waiting.set()
+            # got here, should be done
+            # wait for wait_thread to cleanup
+            self._wait_thread.join()
 
     def _stream_file(self, path):
         """Stream one file"""
@@ -1226,6 +1231,11 @@ async def join(self, timeout=None):
                 wait()
             else:
                 await asyncio.wrap_future(future)
+        if getattr(self, '_stop_waiting', None) and getattr(self, "_wait_thread", None):
+            self._stop_waiting.set()
+            # got here, should be done
+            # wait for wait_thread to cleanup
+            self._wait_thread.join()
 
     def signal(self, sig):
         if self.state == 'running':

ipyparallel/tests/test_cluster.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import logging
 import os
 import signal
 import sys
@@ -310,9 +311,9 @@ async def test_default_from_file(Cluster):
 
 
 async def test_cluster_manager_notice_stop(Cluster):
-    cm = cluster.ClusterManager()
+    cm = cluster.ClusterManager(log=logging.getLogger())
     cm.load_clusters()
-    c = Cluster(n=1)
+    c = Cluster(n=1, log=cm.log)
     key = cm._cluster_key(c)
     assert key not in cm.clusters
 
@@ -326,8 +327,8 @@ async def test_cluster_manager_notice_stop(Cluster):
     # refresh list, cleans out stopped clusters
     # can take some time to notice
     tic = time.perf_counter()
-    deadline = time.perf_counter() + 30
+    deadline = time.perf_counter() + _timeout
     while time.perf_counter() < deadline and key in cm.clusters:
-        time.sleep(0.2)
+        await asyncio.sleep(0.2)
         cm.load_clusters()
     assert key not in cm.clusters