Merge pull request #3654 from minrk/alltargets

fix single-result AsyncResults

`direct_view(targets='all')` could produce single-result behavior if only one engine is registered.

IPython/parallel/client/asyncresult.py

@@ -73,6 +73,9 class AsyncResult(object):
         if isinstance(msg_ids, basestring):
             # always a list
             msg_ids = [msg_ids]
+            self._single_result = True
+        else:
+            self._single_result = False
         if tracker is None:
             # default to always done
             tracker = finished_tracker
@@ -81,14 +84,11 class AsyncResult(object):
         self._fname=fname
         self._targets = targets
         self._tracker = tracker
+        
         self._ready = False
         self._outputs_ready = False
         self._success = None
         self._metadata = [ self._client.metadata.get(id) for id in self.msg_ids ]
-        if len(msg_ids) == 1:
-            self._single_result = not isinstance(targets, (list, tuple))
-        else:
-            self._single_result = False
 
     def __repr__(self):
         if self._ready:

IPython/parallel/client/client.py

@@ -1378,9 +1378,11 class Client(HasTraits):
         block = self.block if block is None else block
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
-
+        
+        single_result = False
         if not isinstance(indices_or_msg_ids, (list,tuple)):
             indices_or_msg_ids = [indices_or_msg_ids]
+            single_result = True
 
         theids = []
         for id in indices_or_msg_ids:
@@ -1392,6 +1394,11 class Client(HasTraits):
 
         local_ids = filter(lambda msg_id: msg_id in self.outstanding or msg_id in self.results, theids)
         remote_ids = filter(lambda msg_id: msg_id not in local_ids, theids)
+        
+        # given single msg_id initially, get_result shot get the result itself,
+        # not a length-one list
+        if single_result:
+            theids = theids[0]
 
         if remote_ids:
             ar = AsyncHubResult(self, msg_ids=theids)

IPython/parallel/client/view.py

@@ -549,8 +549,8 class DirectView(View):
         block = self.block if block is None else block
         track = self.track if track is None else track
         targets = self.targets if targets is None else targets
-
-        _idents = self.client._build_targets(targets)[0]
+        
+        _idents, _targets = self.client._build_targets(targets)
         msg_ids = []
         trackers = []
         for ident in _idents:
@@ -559,8 +559,10 class DirectView(View):
             if track:
                 trackers.append(msg['tracker'])
             msg_ids.append(msg['header']['msg_id'])
+        if isinstance(targets, int):
+            msg_ids = msg_ids[0]
         tracker = None if track is False else zmq.MessageTracker(*trackers)
-        ar = AsyncResult(self.client, msg_ids, fname=getname(f), targets=targets, tracker=tracker)
+        ar = AsyncResult(self.client, msg_ids, fname=getname(f), targets=_targets, tracker=tracker)
         if block:
             try:
                 return ar.get()
@@ -631,13 +633,15 class DirectView(View):
         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]
+        _idents, _targets = self.client._build_targets(targets)
         msg_ids = []
         trackers = []
         for ident in _idents:
             msg = self.client.send_execute_request(self._socket, code, silent=silent, ident=ident)
             msg_ids.append(msg['header']['msg_id'])
-        ar = AsyncResult(self.client, msg_ids, fname='execute', targets=targets)
+        if isinstance(targets, int):
+            msg_ids = msg_ids[0]
+        ar = AsyncResult(self.client, msg_ids, fname='execute', targets=_targets)
         if block:
             try:
                 ar.get()

IPython/parallel/error.py

@@ -43,15 +43,14 class IPythonError(Exception):
 class KernelError(IPythonError):
     pass
 
-
-
 class NoEnginesRegistered(KernelError):
     pass
 
-
 class TaskAborted(KernelError):
     pass
 
+class TaskTimeout(KernelError):
+    pass
 
 class TimeoutError(KernelError):
     pass

IPython/parallel/tests/test_asyncresult.py

@@ -308,4 +308,18 class AsyncResultTest(ClusterTestCase):
         ar.get(5)
         nt.assert_in(4, found)
         self.assertTrue(len(found) > 1, "should have seen data multiple times, but got: %s" % found)
+    
+    def test_not_single_result(self):
+        save_build = self.client._build_targets
+        def single_engine(*a, **kw):
+            idents, targets = save_build(*a, **kw)
+            return idents[:1], targets[:1]
+        ids = single_engine('all')[1]
+        self.client._build_targets = single_engine
+        for targets in ('all', None, ids):
+            dv = self.client.direct_view(targets=targets)
+            ar = dv.apply_async(lambda : 5)
+            self.assertEqual(ar.get(10), [5])
+        self.client._build_targets = save_build
+
 

IPython/parallel/tests/test_client.py

@@ -152,10 +152,10 class TestClient(ClusterTestCase):
         ar = c[t].apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
-        ahr = self.client.get_result(ar.msg_ids)
+        ahr = self.client.get_result(ar.msg_ids[0])
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEqual(ahr.get(), ar.get())
-        ar2 = self.client.get_result(ar.msg_ids)
+        ar2 = self.client.get_result(ar.msg_ids[0])
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.close()
     
@@ -171,10 +171,11 class TestClient(ClusterTestCase):
         ar = c[t].execute("import time; time.sleep(1)", silent=False)
         # give the monitor time to notice the message
         time.sleep(.25)
-        ahr = self.client.get_result(ar.msg_ids)
+        ahr = self.client.get_result(ar.msg_ids[0])
+        print ar.get(), ahr.get(), ar._single_result, ahr._single_result
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEqual(ahr.get().pyout, ar.get().pyout)
-        ar2 = self.client.get_result(ar.msg_ids)
+        ar2 = self.client.get_result(ar.msg_ids[0])
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.close()
     

IPython/parallel/tests/test_view.py

@@ -156,10 +156,10 class TestView(ClusterTestCase, ParametricTestCase):
         ar = v.apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
-        ahr = v2.get_result(ar.msg_ids)
+        ahr = v2.get_result(ar.msg_ids[0])
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEqual(ahr.get(), ar.get())
-        ar2 = v2.get_result(ar.msg_ids)
+        ar2 = v2.get_result(ar.msg_ids[0])
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.spin()
         c.close()