Merge pull request #549 from minrk/simpler-ids

Simpler default cluster / engine set ids

Author: minrk

ipyparallel/apps/baseapp.py

@@ -118,13 +118,6 @@ def _work_dir_changed(self, change):
         """,
     )
 
-    @observe('cluster_id')
-    def _cluster_id_changed(self, change):
-        if change['new']:
-            self.name = '{}-{}'.format(self.__class__.name, change['new'])
-        else:
-            self.name = self.__class__.name
-
     loop = Instance(IOLoop)
 
     def _loop_default(self):
@@ -197,7 +190,7 @@ def reinit_logging(self):
                         pass
         if self.log_to_file:
             # Start logging to the new log file
-            log_filename = self.name + u'-' + str(os.getpid()) + u'.log'
+            log_filename = f"{self.name}-{self.cluster_id}-{os.getpid()}.log"
             logfile = os.path.join(log_dir, log_filename)
             if sys.__stderr__:
                 print(f"Sending logs to {logfile}", file=sys.__stderr__)

ipyparallel/cluster/cluster.py

@@ -101,7 +101,7 @@ class Cluster(AsyncFirst, LoggingConfigurable):
 
     @default("cluster_id")
     def _default_cluster_id(self):
-        return f"{socket.gethostname()}-{int(time.time())}-{''.join(random.choice(_suffix_chars) for i in range(4))}"
+        return f"{int(time.time())}-{''.join(random.choice(_suffix_chars) for i in range(4))}"
 
     profile_dir = Unicode(
         help="""The profile directory.
@@ -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()
@@ -604,14 +611,23 @@ def _controller_stopped(self, stop_data=None):
         self.log.info(f"Controller stopped: {stop_data}")
         self.update_cluster_file()
 
+    def _new_engine_set_id(self):
+        """Generate a new engine set id"""
+        engine_set_id = base = f"{int(time.time())}"
+        i = 1
+        while engine_set_id in self.engines:
+            engine_set_id = f"{base}-{i}"
+            i += 1
+        return engine_set_id
+
     async def start_engines(self, n=None, engine_set_id=None, **kwargs):
         """Start an engine set
 
         Returns an engine set id which can be used in stop_engines
         """
         # TODO: send engines connection info
         if engine_set_id is None:
-            engine_set_id = f"{int(time.time())}-{''.join(random.choice(_suffix_chars) for i in range(4))}"
+            engine_set_id = self._new_engine_set_id()
         engine_set = self.engines[engine_set_id] = self.engine_launcher_class(
             work_dir=u'.',
             parent=self,
@@ -853,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/controller/app.py

@@ -235,9 +235,11 @@ class IPController(BaseParallelApplication):
 
     @observe('cluster_id')
     def _cluster_id_changed(self, change):
-        super()._cluster_id_changed(change)
-        self.engine_json_file = "%s-engine.json" % self.name
-        self.client_json_file = "%s-client.json" % self.name
+        base = 'ipcontroller'
+        if change.new:
+            base = f"{base}-{change.new}"
+        self.engine_json_file = f"{base}-engine.json"
+        self.client_json_file = f"{base}-client.json"
 
     # internal
     children = List()

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