Merge pull request #514 from minrk/signal-engines

add KernelNanny for remote signaling

Author: minrk

docs/source/tutorial/direct.rst

@@ -266,16 +266,16 @@ and blocks until all of the associated results are ready:
 
 .. sourcecode:: ipython
 
-    In [72]: dview.block=False
+    In [72]: dview.block = False
 
     # A trivial list of AsyncResults objects
-    In [73]: pr_list = [dview.apply_async(wait, 3) for i in range(10)]
+    In [73]: ar_list = [dview.apply_async(wait, 3) for i in range(10)]
 
     # Wait until all of them are done
-    In [74]: dview.wait(pr_list)
+    In [74]: dview.wait(ar_list)
 
     # Then, their results are ready using get()
-    In [75]: pr_list[0].get()
+    In [75]: ar_list[0].get()
     Out[75]: [2.9982571601867676, 2.9982588291168213, 2.9987530708312988, 2.9990990161895752]
 
 
@@ -297,7 +297,7 @@ The following examples demonstrate how to use the instance attributes:
 
 .. sourcecode:: ipython
 
-    In [16]: dview.targets = [0,2]
+    In [16]: dview.targets = [0, 2]
 
     In [17]: dview.block = False
 
@@ -406,6 +406,56 @@ between engines, MPI, pyzmq, or some other direct interconnect should be used.
 Other things to look at
 =======================
 
+Signaling engines
+-----------------
+
+New in IPython Parallel 7.0 is the :meth:`Client.send_signal` method.
+This lets you directly interrupt engines, which might be running a blocking task
+that you want to cancel.
+
+This is also available via the Cluster API.
+Unlike the Cluster API, though,
+which only allows interrupting whole engine 'sets' (usally all engines in the cluster),
+the client API allows interrupting individual engines.
+
+.. sourcecode:: ipython
+
+    In [9]: ar = rc[:].apply_async(time.sleep, 5)
+
+    In [10]: rc.send_signal(signal.SIGINT)
+    Out[10]: <Future at 0x7f91a9489fd0 state=pending>
+
+    In [11]: ar.get()
+    [12:apply]:
+    ---------------------------------------------------------------------------
+    KeyboardInterrupt                         Traceback (most recent call last)
+    <string> in <module>
+
+    KeyboardInterrupt:
+
+    [13:apply]:
+    ---------------------------------------------------------------------------
+    KeyboardInterrupt                         Traceback (most recent call last)
+    <string> in <module>
+
+    KeyboardInterrupt:
+
+    [14:apply]:
+    ---------------------------------------------------------------------------
+    KeyboardInterrupt                         Traceback (most recent call last)
+    <string> in <module>
+
+    KeyboardInterrupt:
+
+    [15:apply]:
+    ---------------------------------------------------------------------------
+    KeyboardInterrupt                         Traceback (most recent call last)
+    <string> in <module>
+
+    KeyboardInterrupt:
+
+
+
 Remote function decorators
 --------------------------
 

ipyparallel/apps/baseapp.py

@@ -202,6 +202,8 @@ def reinit_logging(self):
             # Start logging to the new log file
             log_filename = self.name + u'-' + str(os.getpid()) + u'.log'
             logfile = os.path.join(log_dir, log_filename)
+            if sys.__stderr__:
+                print(f"Sending logs to {logfile}", file=sys.__stderr__)
             open_log_file = open(logfile, 'w')
         else:
             open_log_file = None

ipyparallel/client/client.py

@@ -1494,14 +1494,13 @@ def wait_interactive(self, jobs=None, interval=1.0, timeout=-1.0):
     # Control methods
     # --------------------------------------------------------------------------
 
-    def clear(self, targets=None, block=None):
-        """Clear the namespace in target(s)."""
-        block = self.block if block is None else block
-        targets = self._build_targets(targets)[0]
+    def _send_control_request(self, targets, msg_type, content, block):
+        """Send a request on the control channel"""
+        target_identities = self._build_targets(targets)[0]
         futures = []
-        for t in targets:
+        for ident in target_identities:
             futures.append(
-                self._send(self._control_stream, 'clear_request', content={}, ident=t)
+                self._send(self._control_stream, msg_type, content=content, ident=ident)
             )
         if not block:
             return multi_future(futures)
@@ -1511,6 +1510,34 @@ def clear(self, targets=None, block=None):
             if msg['content']['status'] != 'ok':
                 raise self._unwrap_exception(msg['content'])
 
+    def send_signal(self, sig, targets=None, block=None):
+        """Send a signal target(s).
+
+        Parameters
+        ----------
+
+        sig: int or str
+            The signal number or name to send.
+            If a str, will evaluate to getattr(signal, sig) on the engine,
+            which is useful for sending signals cross-platform.
+
+        .. versionadded:: 7.0
+        """
+        block = self.block if block is None else block
+        return self._send_control_request(
+            targets=targets,
+            msg_type='signal_request',
+            content={'sig': sig},
+            block=block,
+        )
+
+    def clear(self, targets=None, block=None):
+        """Clear the namespace in target(s)."""
+        block = self.block if block is None else block
+        return self._send_control_request(
+            targets=targets, msg_type='clear_request', content={}, block=block
+        )
+
     def abort(self, jobs=None, targets=None, block=None):
         """Abort specific jobs from the execution queues of target(s).
 
@@ -1531,7 +1558,6 @@ def abort(self, jobs=None, targets=None, block=None):
         """
         block = self.block if block is None else block
         jobs = jobs if jobs is not None else list(self.outstanding)
-        targets = self._build_targets(targets)[0]
 
         msg_ids = []
         if isinstance(jobs, string_types + (AsyncResult,)):
@@ -1549,22 +1575,13 @@ def abort(self, jobs=None, targets=None, block=None):
             else:
                 msg_ids.append(j)
         content = dict(msg_ids=msg_ids)
-        futures = []
-        for t in targets:
-            futures.append(
-                self._send(
-                    self._control_stream, 'abort_request', content=content, ident=t
-                )
-            )
 
-        if not block:
-            return multi_future(futures)
-        else:
-            for f in futures:
-                f.wait()
-                msg = f.result()
-                if msg['content']['status'] != 'ok':
-                    raise self._unwrap_exception(msg['content'])
+        return self._send_control_request(
+            targets,
+            msg_type='abort_request',
+            content=content,
+            block=block,
+        )
 
     def shutdown(self, targets='all', restart=False, hub=False, block=None):
         """Terminates one or more engine processes, optionally including the hub.

ipyparallel/cluster/_win32support.py

@@ -250,15 +250,14 @@ def _classes_default(self):
     daemonize = Bool(
         False,
         config=True,
-        help="""Daemonize the ipcluster program. This implies --log-to-file.
+        help="""Launch the cluster and immediately exit.
+
+        .. versionchanged:: 7.0
+            No longer leaves the ipcluster process itself running.
+            Prior to 7.0, --daemonize did not work on Windows.
         """,
     )
 
-    @observe('daemonize')
-    def _daemonize_changed(self, change):
-        if change['new']:
-            self.log_to_file = True
-
     early_shutdown = Integer(
         30,
         config=True,
@@ -382,6 +381,9 @@ def engines_stopped_early(self, stop_data):
             a public interface.
             """
             )
+            engine_output = self.engine_launcher.get_output(remove=True)
+            if engine_output:
+                self.log.error(f"Engine output:\n{engine_output}")
             self.loop.add_callback(self.stop_cluster)
 
         return self.engines_stopped(stop_data)
@@ -471,7 +473,9 @@ def engines_stopped(self, r):
     async def start_cluster(self):
         await self.cluster.start_cluster()
         if self.daemonize:
-            self.log.info(f"Leaving cluster running: {self.cluster.cluster_file}")
+            print(
+                f"Leaving cluster running: {self.cluster.cluster_file}", file=sys.stderr
+            )
             self.loop.add_callback(self.loop.stop)
         self.cluster._controller.on_stop(self.controller_stopped)
         self.watch_engines()

ipyparallel/cluster/app.py

@@ -357,7 +357,9 @@ def from_dict(cls, d, **kwargs):
             cls = self.engine_launcher_class = import_item(engine_info["class"])
             for engine_set_id, engine_state in engine_info.get("sets", {}).items():
                 self._engine_sets[engine_set_id] = cls.from_dict(
-                    engine_state, parent=self
+                    engine_state,
+                    engine_set_id=engine_set_id,
+                    parent=self,
                 )
 
         return self
@@ -503,6 +505,7 @@ async def start_engines(self, n=None, engine_set_id=None, **kwargs):
             log=self.log,
             profile_dir=self.profile_dir,
             cluster_id=self.cluster_id,
+            engine_set_id=engine_set_id,
             **kwargs,
         )
         if n is None:
@@ -547,6 +550,8 @@ async def stop_engines(self, engine_set_id=None):
         r = engine_set.stop()
         if inspect.isawaitable(r):
             await r
+        # retrieve and cleanup output files
+        engine_set.get_output(remove=True)
         self._engine_sets.pop(engine_set_id)
         self.update_cluster_file()
 
@@ -608,6 +613,9 @@ async def stop_controller(self):
             if inspect.isawaitable(r):
                 await r
 
+        if self._controller:
+            self._controller.get_output(remove=True)
+
         self._controller = None
         self.update_cluster_file()