use execute_request for parallel execute, instead of apply

IPython/parallel/client/client.py

@@ -628,7 +628,30 b' class Client(HasTraits):'
                 print ("got unknown result: %s"%msg_id)
         else:
             self.outstanding.remove(msg_id)
-        self.results[msg_id] = self._unwrap_exception(msg['content'])
+
+        content = msg['content']
+        header = msg['header']
+
+        # construct metadata:
+        md = self.metadata[msg_id]
+        md.update(self._extract_metadata(header, parent, content))
+        # is this redundant?
+        self.metadata[msg_id] = md
+        
+        e_outstanding = self._outstanding_dict[md['engine_uuid']]
+        if msg_id in e_outstanding:
+            e_outstanding.remove(msg_id)
+
+        # construct result:
+        if content['status'] == 'ok':
+            self.results[msg_id] = content
+        elif content['status'] == 'aborted':
+            self.results[msg_id] = error.TaskAborted(msg_id)
+        elif content['status'] == 'resubmitted':
+            # TODO: handle resubmission
+            pass
+        else:
+            self.results[msg_id] = self._unwrap_exception(content)
 
     def _handle_apply_reply(self, msg):
         """Save the reply to an apply_request into our results."""
@@ -1024,7 +1047,7 b' class Client(HasTraits):'
 
         return result
 
-    def send_apply_message(self, socket, f, args=None, kwargs=None, subheader=None, track=False,
+    def send_apply_request(self, socket, f, args=None, kwargs=None, subheader=None, track=False,
                             ident=None):
         """construct and send an apply message via a socket.
 
@@ -1066,6 +1089,41 b' class Client(HasTraits):'
 
         return msg
 
+    def send_execute_request(self, socket, code, silent=False, subheader=None, ident=None):
+        """construct and send an execute request via a socket.
+
+        """
+
+        assert not self._closed, "cannot use me anymore, I'm closed!"
+        # defaults:
+        subheader = subheader if subheader is not None else {}
+
+        # validate arguments
+        if not isinstance(code, basestring):
+            raise TypeError("code must be text, not %s" % type(code))
+        if not isinstance(subheader, dict):
+            raise TypeError("subheader must be dict, not %s" % type(subheader))
+        
+        content = dict(code=code, silent=bool(silent), user_variables=[], user_expressions={})
+
+
+        msg = self.session.send(socket, "execute_request", content=content, ident=ident,
+                            subheader=subheader)
+
+        msg_id = msg['header']['msg_id']
+        self.outstanding.add(msg_id)
+        if ident:
+            # possibly routed to a specific engine
+            if isinstance(ident, list):
+                ident = ident[-1]
+            if ident in self._engines.values():
+                # save for later, in case of engine death
+                self._outstanding_dict[ident].add(msg_id)
+        self.history.append(msg_id)
+        self.metadata[msg_id]['submitted'] = datetime.now()
+
+        return msg
+
     #--------------------------------------------------------------------------
     # construct a View object
     #--------------------------------------------------------------------------

IPython/parallel/client/view.py

@@ -195,7 +195,7 b' class View(HasTraits):'
     @sync_results
     @save_ids
     def _really_apply(self, f, args, kwargs, block=None, **options):
-        """wrapper for client.send_apply_message"""
+        """wrapper for client.send_apply_request"""
         raise NotImplementedError("Implement in subclasses")
 
     def apply(self, f, *args, **kwargs):
@@ -533,7 +533,7 b' class DirectView(View):'
         msg_ids = []
         trackers = []
         for ident in _idents:
-            msg = self.client.send_apply_message(self._socket, f, args, kwargs, track=track,
+            msg = self.client.send_apply_request(self._socket, f, args, kwargs, track=track,
                                     ident=ident)
             if track:
                 trackers.append(msg['tracker'])
@@ -547,6 +547,7 b' class DirectView(View):'
                 pass
         return ar
 
+
     @spin_after
     def map(self, f, *sequences, **kwargs):
         """view.map(f, *sequences, block=self.block) => list|AsyncMapResult
@@ -590,6 +591,8 b' class DirectView(View):'
         pf = ParallelFunction(self, f, block=block, **kwargs)
         return pf.map(*sequences)
 
+    @sync_results
+    @save_ids
     def execute(self, code, targets=None, block=None):
         """Executes `code` on `targets` in blocking or nonblocking manner.
 
@@ -604,6 +607,23 b' class DirectView(View):'
                 whether or not to wait until done to return
                 default: self.block
         """
+        block = self.block if block is None else block
+        targets = self.targets if targets is None else targets
+
+        _idents = self.client._build_targets(targets)[0]
+        msg_ids = []
+        trackers = []
+        for ident in _idents:
+            msg = self.client.send_execute_request(self._socket, code, ident=ident)
+            msg_ids.append(msg['header']['msg_id'])
+        ar = AsyncResult(self.client, msg_ids, fname='execute', targets=targets)
+        if block:
+            try:
+                ar.get()
+            except KeyboardInterrupt:
+                pass
+        return ar
+        
         return self._really_apply(util._execute, args=(code,), block=block, targets=targets)
 
     def run(self, filename, targets=None, block=None):
@@ -996,7 +1016,7 b' class LoadBalancedView(View):'
         follow = self._render_dependency(follow)
         subheader = dict(after=after, follow=follow, timeout=timeout, targets=idents, retries=retries)
 
-        msg = self.client.send_apply_message(self._socket, f, args, kwargs, track=track,
+        msg = self.client.send_apply_request(self._socket, f, args, kwargs, track=track,
                                 subheader=subheader)
         tracker = None if track is False else msg['tracker']