Merge pull request #473 from minrk/broadcast_outstanding

[BroadcastView] hook up derivative message futures before send

Author: minrk

ipyparallel/client/client.py

@@ -1104,6 +1104,7 @@ def _send(
         header=None,
         metadata=None,
         track_outstanding=False,
+        message_future_hook=None,
     ):
         """Send a message in the IO thread
 
@@ -1136,6 +1137,8 @@ def _send(
             async_result=msg_type in {'execute_request', 'apply_request'},
             track=track,
         )
+        if message_future_hook is not None:
+            message_future_hook(futures[0])
 
         def cleanup(f):
             """Purge caches on Future resolution"""
@@ -1574,7 +1577,15 @@ def _maybe_raise(self, result):
         return result
 
     def send_apply_request(
-        self, socket, f, args=None, kwargs=None, metadata=None, track=False, ident=None
+        self,
+        socket,
+        f,
+        args=None,
+        kwargs=None,
+        metadata=None,
+        track=False,
+        ident=None,
+        message_future_hook=None,
     ):
         """construct and send an apply message via a socket.
 
@@ -1617,13 +1628,20 @@ def send_apply_request(
             metadata=metadata,
             track=track,
             track_outstanding=True,
+            message_future_hook=message_future_hook,
         )
         msg_id = future.msg_id
 
         return future
 
     def send_execute_request(
-        self, socket, code, silent=True, metadata=None, ident=None
+        self,
+        socket,
+        code,
+        silent=True,
+        metadata=None,
+        ident=None,
+        message_future_hook=None,
     ):
         """construct and send an execute request via a socket."""
 
@@ -1650,6 +1668,7 @@ def send_execute_request(
             ident=ident,
             metadata=metadata,
             track_outstanding=True,
+            message_future_hook=message_future_hook,
         )
 
         return future

ipyparallel/client/view.py

@@ -912,11 +912,23 @@ def _really_apply(
         s_idents = [ident.decode("utf8") for ident in idents]
 
         metadata = self._init_metadata(s_idents)
-        message_future = self.client.send_apply_request(
-            self._socket, pf, pargs, pkwargs, track=track, metadata=metadata
-        )
-        ar = self._make_async_result(
-            message_future, s_idents, fname=getname(f), targets=targets
+
+        ar = None
+
+        def make_asyncresult(message_future):
+            nonlocal ar
+            ar = self._make_async_result(
+                message_future, s_idents, fname=getname(f), targets=_targets
+            )
+
+        self.client.send_apply_request(
+            self._socket,
+            pf,
+            pargs,
+            pkwargs,
+            track=track,
+            metadata=metadata,
+            message_future_hook=make_asyncresult,
         )
 
         if block:
@@ -948,14 +960,21 @@ def execute(self, code, silent=True, targets=None, block=None):
         s_idents = [ident.decode("utf8") for ident in _idents]
 
         metadata = self._init_metadata(s_idents)
+
+        ar = None
+
+        def make_asyncresult(message_future):
+            nonlocal ar
+            ar = self._make_async_result(
+                message_future, s_idents, fname='execute', targets=_targets
+            )
+
         message_future = self.client.send_execute_request(
             self._socket,
             code,
             silent=silent,
             metadata=metadata,
-        )
-        ar = self._make_async_result(
-            message_future, s_idents, fname='execute', targets=_targets
+            message_future_hook=make_asyncresult,
         )
         if block:
             try: