handle 'after' state in launchers

- don't reserialize clusters that have stopped
- load_clusters purges stopped clusters

Author: minrk

ipyparallel/cluster/cluster.py

@@ -378,7 +378,7 @@ def __repr__(self):
             fields["profile_dir"] = repr(profile_dir)
 
         if self.controller:
-            fields["controller"] = "<running>"
+            fields["controller"] = f"<{self.controller.state}>"
         if self.engines:
             fields["engine_sets"] = list(self.engines)
 
@@ -398,7 +398,7 @@ def _cls_str(cls):
 
         cluster_info["class"] = _cls_str(self.__class__)
 
-        if self.controller:
+        if self.controller and self.controller.state != 'after':
             d["controller"] = {
                 "class": _cls_str(self.controller_launcher_class),
                 "state": None,
@@ -411,7 +411,8 @@ def _cls_str(cls):
         }
         sets = d["engines"]["sets"]
         for engine_set_id, engine_launcher in self.engines.items():
-            sets[engine_set_id] = engine_launcher.to_dict()
+            if engine_launcher.state != 'after':
+                sets[engine_set_id] = engine_launcher.to_dict()
         return d
 
     @classmethod
@@ -447,13 +448,15 @@ def from_dict(cls, d, **kwargs):
                     )
                 except launcher.NotRunning as e:
                     self.log.error(f"Controller for {cluster_key} not running: {e}")
+                else:
+                    self.controller.on_stop(self._controller_stopped)
 
         engine_info = d.get("engines")
         if engine_info:
             cls = self.engine_launcher_class = import_item(engine_info["class"])
             for engine_set_id, engine_state in engine_info.get("sets", {}).items():
                 try:
-                    self.engines[engine_set_id] = cls.from_dict(
+                    self.engines[engine_set_id] = engine_set = cls.from_dict(
                         engine_state,
                         engine_set_id=engine_set_id,
                         parent=self,
@@ -462,6 +465,9 @@ def from_dict(cls, d, **kwargs):
                     self.log.error(
                         f"Engine set {cluster_key}{engine_set_id} not running: {e}"
                     )
+                else:
+                    engine_set.on_stop(partial(self._engines_stopped, engine_set_id))
+
         # check if state changed
         if self.to_dict() != d:
             # if so, update our cluster file
@@ -527,7 +533,9 @@ def update_cluster_file(self):
             # setting cluster_file='' disables saving to disk
             return
 
-        if not self.controller and not self.engines:
+        if (not self.controller or self.controller.state == 'after') and not any(
+            es.state == 'after' for es in self.engines.values()
+        ):
             self.remove_cluster_file()
         else:
             self.write_cluster_file()
@@ -594,6 +602,7 @@ def add_args(args):
     def _controller_stopped(self, stop_data=None):
         """Callback when a controller stops"""
         self.log.info(f"Controller stopped: {stop_data}")
+        self.update_cluster_file()
 
     async def start_engines(self, n=None, engine_set_id=None, **kwargs):
         """Start an engine set
@@ -839,6 +848,17 @@ def load_clusters(
         - single profile by name
         - all IPython profiles, if nothing else specified
         """
+
+        # first, check our current 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}")
+                self.clusters.pop(key)
+
         if profile_dirs is None:
             if profile_dir is not None:
                 profile_dirs = [profile_dir]

ipyparallel/cluster/launcher.py

@@ -112,7 +112,7 @@ class BaseLauncher(LoggingConfigurable):
     start_data = Any()
     stop_data = Any()
 
-    identifier = Any(
+    identifier = Unicode(
         help="Used for lookup in e.g. EngineSetLauncher during notify_stop and default log files"
     )
 
@@ -521,7 +521,7 @@ def get_output(self, remove=False):
                     with open(self.output_file) as f:
                         self._output = f.read()
                 except FileNotFoundError:
-                    self.log.error(f"Missing output file: {self.output_file}")
+                    self.log.debug(f"Missing output file: {self.output_file}")
                     self._output = ""
             else:
                 self._output = ""
@@ -658,13 +658,14 @@ def from_dict(cls, d, **kwargs):
         n = 0
         for i, engine_dict in d['engines'].items():
             try:
-                self.launchers[i] = self.launcher_class.from_dict(
-                    engine_dict, parent=self
+                self.launchers[i] = el = self.launcher_class.from_dict(
+                    engine_dict, identifier=i, parent=self
                 )
             except NotRunning as e:
                 self.log.error(f"Engine {i} not running: {e}")
             else:
                 n += 1
+                el.on_stop(self._notice_engine_stopped)
         if n == 0:
             raise NotRunning("No engines left")
         else:
@@ -676,15 +677,16 @@ def start(self, n):
         self.n = n
         dlist = []
         for i in range(n):
+            identifier = str(i)
             if i > 0:
                 time.sleep(self.delay)
-            el = self.launchers[i] = self.launcher_class(
+            el = self.launchers[identifier] = self.launcher_class(
                 work_dir=self.work_dir,
                 parent=self,
                 log=self.log,
                 profile_dir=self.profile_dir,
                 cluster_id=self.cluster_id,
-                identifier=i,
+                identifier=identifier,
                 output_file=os.path.join(
                     self.profile_dir,
                     "log",

ipyparallel/tests/test_cluster.py

@@ -307,3 +307,23 @@ async def test_default_from_file(Cluster):
         assert cluster2.cluster_file == cluster.cluster_file
         with await cluster.connect_client() as rc:
             assert len(rc) == 1
+
+
+async def test_cluster_manager_notice_stop(Cluster):
+    cm = cluster.ClusterManager()
+    cm.load_clusters()
+    c = Cluster(n=1)
+    key = cm._cluster_key(c)
+    assert key not in cm.clusters
+
+    await c.start_cluster()
+    cm.load_clusters()
+    assert key in cm.clusters
+    c_copy = cm.clusters[key]
+
+    await c.stop_cluster()
+    # give it a moment to notice
+    time.sleep(5)
+    # refresh list, cleans out stopped clusters
+    cm.load_clusters()
+    assert key not in cm.clusters

lab/src/clusters.tsx

@@ -717,6 +717,9 @@ function ClusterListingItem(props: IClusterListingItemProps) {
   let cluster_state = "Stopped";
   if (cluster.controller) {
     cluster_state = cluster.controller.state.state;
+    if (cluster_state == "after") {
+      cluster_state = "Stopped";
+    }
   }
 
   // stop action is 'delete' for already-stopped clusters