pool.submit cannot be called during process teardown

improve teardown of clusters during process exit

log errors and advance in case of failure to stop one cluster so it doesn't prevent stopping others

Author: minrk

ipyparallel/cluster/cluster.py

@@ -16,6 +16,7 @@
 import string
 import sys
 import time
+import traceback
 from functools import partial
 from multiprocessing import cpu_count
 from weakref import WeakSet
@@ -56,7 +57,11 @@ def _atexit_cleanup_clusters(*args):
             continue
         if cluster.controller or cluster.engines:
             print(f"Stopping cluster {cluster}", file=sys.stderr)
-            cluster.stop_cluster_sync()
+            try:
+                cluster.stop_cluster_sync()
+            except Exception:
+                print(f"Error stopping cluster {cluster}", file=sys.stderr)
+                traceback.print_exception(*sys.exc_info())
 
 
 _atexit_cleanup_clusters.registered = False

ipyparallel/cluster/launcher.py

@@ -465,10 +465,18 @@ def start(self):
     async def join(self, timeout=None):
         """Wait for the process to exit"""
         with ThreadPoolExecutor(1) as pool:
+            wait = partial(self.process.wait, timeout)
             try:
-                await asyncio.wrap_future(
-                    pool.submit(partial(self.process.wait, timeout))
-                )
+                try:
+                    future = pool.submit(wait)
+                except RuntimeError:
+                    # e.g. called during process shutdown,
+                    # which raises
+                    # RuntimeError: cannot schedule new futures after interpreter shutdown
+                    # Instead, do the blocking call
+                    wait()
+                else:
+                    await asyncio.wrap_future(future)
             except psutil.TimeoutExpired:
                 raise TimeoutError(
                     f"Process {self.pid} did not complete in {timeout} seconds."
@@ -1184,9 +1192,17 @@ def wait_one(self, timeout):
 
     async def join(self, timeout=None):
         with ThreadPoolExecutor(1) as pool:
-            await asyncio.wrap_future(
-                pool.submit(partial(self.wait_one, timeout=timeout))
-            )
+            wait = partial(self.wait_one, timeout=timeout)
+            try:
+                future = pool.submit(wait)
+            except RuntimeError:
+                # e.g. called during process shutdown,
+                # which raises
+                # RuntimeError: cannot schedule new futures after interpreter shutdown
+                # Instead, do the blocking call
+                wait()
+            else:
+                await asyncio.wrap_future(future)
 
     def signal(self, sig):
         if self.state == 'running':