add KernelNanny for remote signaling

Nanny proxies control channel
- add Client.send_signal for signaling engines
- notify Hub immediately on process exit instead of waiting for heart failure
- exit Engine on any error during completion of registration
- consolidate `_send_control_request` for control-channel requests
- fix some mixed str/bytes uuid/identity logic in the Hub

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/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/controller/hub.py

@@ -23,6 +23,7 @@
 from tornado.gen import coroutine
 from tornado.gen import maybe_future
 from traitlets import Any
+from traitlets import Bytes
 from traitlets import Dict
 from traitlets import HasTraits
 from traitlets import Instance
@@ -122,8 +123,9 @@ class EngineConnector(HasTraits):
     stallback: tornado timeout for stalled registration
     """
 
-    id = Integer(0)
+    id = Integer()
     uuid = Unicode()
+    ident = Bytes()
     pending = Set()
     stallback = Any()
 
@@ -152,17 +154,15 @@ class Hub(LoggingConfigurable):
 
     # internal data structures:
     ids = Set()  # engine IDs
-    keytable = Dict()
-    by_ident = Dict()
-    engines = Dict()
-    clients = Dict()
-    hearts = Dict()
+    by_ident = Dict()  # map bytes identities : int engine id
+    engines = Dict()  # map int engine id : EngineConnector
+    hearts = Dict()  # map bytes identities : int engine id, only for active heartbeats
     pending = Set()
     queues = Dict()  # pending msg_ids keyed by engine_id
     tasks = Dict()  # pending msg_ids submitted as tasks, keyed by client_id
     completed = Dict()  # completed msg_ids keyed by engine_id
     all_completed = Set()  # completed msg_ids keyed by engine_id
-    unassigned = Set()  # set of task msg_ds not yet assigned a destination
+    unassigned = Set()  # set of task msg_ids not yet assigned a destination
     incoming_registrations = Dict()
     registration_timeout = Integer()
     _idcounter = Integer(0)
@@ -830,8 +830,8 @@ def connection_request(self, client_id, msg):
         self.log.info("client::client %r connected", client_id)
         content = dict(status='ok')
         jsonable = {}
-        for k, v in iteritems(self.keytable):
-            jsonable[str(k)] = v
+        for eid, ec in self.engines.items():
+            jsonable[str(eid)] = ec.uuid
         content['engines'] = jsonable
         self.session.send(
             self.query, 'connection_reply', content, parent=msg, ident=client_id
@@ -889,7 +889,9 @@ def register_engine(self, reg, msg):
         if content['status'] == 'ok':
             if heart in self.heartmonitor.hearts:
                 # already beating
-                self.incoming_registrations[heart] = EngineConnector(id=eid, uuid=uuid)
+                self.incoming_registrations[heart] = EngineConnector(
+                    id=eid, uuid=uuid, ident=heart
+                )
                 self.finish_registration(heart)
             else:
                 purge = lambda: self._purge_stalled_registration(heart)
@@ -898,7 +900,7 @@ def register_engine(self, reg, msg):
                     purge,
                 )
                 self.incoming_registrations[heart] = EngineConnector(
-                    id=eid, uuid=uuid, stallback=t
+                    id=eid, uuid=uuid, ident=heart, stallback=t
                 )
         else:
             self.log.error(
@@ -920,23 +922,22 @@ def unregister_engine(self, ident, msg):
             return
         self.log.info("registration::unregister_engine(%r)", eid)
 
-        uuid = self.keytable[eid]
-        content = dict(id=eid, uuid=uuid)
+        ec = self.engines[eid]
+        content = dict(id=eid, uuid=ec.uuid)
 
         # stop the heartbeats
-        self.hearts.pop(uuid, None)
-        self.heartmonitor.responses.discard(uuid)
-        self.heartmonitor.hearts.discard(uuid)
+        self.hearts.pop(ec.ident, None)
+        self.heartmonitor.responses.discard(ec.ident)
+        self.heartmonitor.hearts.discard(ec.ident)
 
         self.loop.add_timeout(
             self.loop.time() + self.registration_timeout,
-            lambda: self._handle_stranded_msgs(eid, uuid),
+            lambda: self._handle_stranded_msgs(eid, ec.uuid),
         )
 
         # cleanup mappings
-        self.by_ident.pop(uuid, None)
+        self.by_ident.pop(ec.ident, None)
         self.engines.pop(eid, None)
-        self.keytable.pop(eid, None)
 
         self._save_engine_state()
 
@@ -993,14 +994,13 @@ def finish_registration(self, heart):
             self.loop.remove_timeout(ec.stallback)
         eid = ec.id
         self.ids.add(eid)
-        self.keytable[eid] = ec.uuid
         self.engines[eid] = ec
-        self.by_ident[cast_bytes(ec.uuid)] = ec.id
+        self.by_ident[ec.ident] = ec.id
         self.queues[eid] = list()
         self.tasks[eid] = list()
         self.completed[eid] = list()
         self.hearts[heart] = eid
-        content = dict(id=eid, uuid=self.engines[eid].uuid)
+        content = dict(id=eid, uuid=ec.uuid)
         if self.notifier:
             self.session.send(
                 self.notifier, "registration_notification", content=content
@@ -1067,7 +1067,9 @@ def _load_engine_state(self):
             self.heartmonitor.responses.add(heart)
             self.heartmonitor.hearts.add(heart)
 
-            self.incoming_registrations[heart] = EngineConnector(id=int(eid), uuid=uuid)
+            self.incoming_registrations[heart] = EngineConnector(
+                id=int(eid), uuid=uuid, ident=heart
+            )
             self.finish_registration(heart)
 
         self.notifier = save_notifier