add unordered iteration to AsyncMapResults

only exposed explicitly on LoadBalancedView.map

IPython/parallel/client/asyncresult.py

@@ -261,12 +261,19 b' class AsyncMapResult(AsyncResult):'
     """Class for representing results of non-blocking gathers.
 
     This will properly reconstruct the gather.
+    
+    This class is iterable at any time, and will wait on results as they come.
+    
+    If ordered=False, then the first results to arrive will come first, otherwise
+    results will be yielded in the order they were submitted.
+    
     """
 
-    def __init__(self, client, msg_ids, mapObject, fname=''):
+    def __init__(self, client, msg_ids, mapObject, fname='', ordered=True):
         AsyncResult.__init__(self, client, msg_ids, fname=fname)
         self._mapObject = mapObject
         self._single_result = False
+        self.ordered = ordered
 
     def _reconstruct_result(self, res):
         """Perform the gather on the actual results."""
@@ -274,6 +281,13 b' class AsyncMapResult(AsyncResult):'
 
     # asynchronous iterator:
     def __iter__(self):
+        it = self._ordered_iter if self.ordered else self._unordered_iter
+        for r in it():
+            yield r
+
+    # asynchronous ordered iterator:
+    def _ordered_iter(self):
+        """iterator for results *as they arrive*, preserving submission order."""
         try:
             rlist = self.get(0)
         except error.TimeoutError:
@@ -294,6 +308,42 b' class AsyncMapResult(AsyncResult):'
             for r in rlist:
                 yield r
 
+    # asynchronous unordered iterator:
+    def _unordered_iter(self):
+        """iterator for results *as they arrive*, on FCFS basis, ignoring submission order."""
+        try:
+            rlist = self.get(0)
+        except error.TimeoutError:
+            pending = set(self.msg_ids)
+            while pending:
+                try:
+                    self._client.wait(pending, 1e-3)
+                except error.TimeoutError:
+                    # ignore timeout error, because that only means
+                    # *some* jobs are outstanding
+                    pass
+                # update ready set with those no longer outstanding:
+                ready = pending.difference(self._client.outstanding)
+                # update pending to exclude those that are finished
+                pending = pending.difference(ready)
+                while ready:
+                    msg_id = ready.pop()
+                    ar = AsyncResult(self._client, msg_id, self._fname)
+                    rlist = ar.get()
+                    try:
+                        for r in rlist:
+                            yield r
+                    except TypeError:
+                        # flattened, not a list
+                        # this could get broken by flattened data that returns iterables
+                        # but most calls to map do not expose the `flatten` argument
+                        yield rlist
+        else:
+            # already done
+            for r in rlist:
+                yield r
+
+
 
 class AsyncHubResult(AsyncResult):
     """Class to wrap pending results that must be requested from the Hub.

IPython/parallel/client/remotefunction.py

@@ -46,7 +46,7 b' def remote(view, block=None, **flags):'
     return remote_function
 
 @skip_doctest
-def parallel(view, dist='b', block=None, **flags):
+def parallel(view, dist='b', block=None, ordered=True, **flags):
     """Turn a function into a parallel remote function.
 
     This method can be used for map:
@@ -57,7 +57,7 b" def parallel(view, dist='b', block=None, **flags):"
     """
 
     def parallel_function(f):
-        return ParallelFunction(view, f, dist=dist, block=block, **flags)
+        return ParallelFunction(view, f, dist=dist, block=block, ordered=ordered, **flags)
     return parallel_function
 
 #--------------------------------------------------------------------------
@@ -122,15 +122,19 b' class ParallelFunction(RemoteFunction):'
         to use the current `block` attribute of `view`
     chunksize : int or None
         The size of chunk to use when breaking up sequences in a load-balanced manner
+    ordered : bool [default: True]
+        Whether 
     **flags : remaining kwargs are passed to View.temp_flags
     """
 
     chunksize=None
+    ordered=None
     mapObject=None
 
-    def __init__(self, view, f, dist='b', block=None, chunksize=None, **flags):
+    def __init__(self, view, f, dist='b', block=None, chunksize=None, ordered=True, **flags):
         super(ParallelFunction, self).__init__(view, f, block=block, **flags)
         self.chunksize = chunksize
+        self.ordered = ordered
 
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
@@ -186,7 +190,10 b' class ParallelFunction(RemoteFunction):'
 
             msg_ids.append(ar.msg_ids[0])
 
-        r = AsyncMapResult(self.view.client, msg_ids, self.mapObject, fname=self.func.__name__)
+        r = AsyncMapResult(self.view.client, msg_ids, self.mapObject, 
+                            fname=self.func.__name__,
+                            ordered=self.ordered
+                        )
 
         if self.block:
             try:

IPython/parallel/client/view.py

@@ -992,7 +992,7 b' class LoadBalancedView(View):'
     @spin_after
     @save_ids
     def map(self, f, *sequences, **kwargs):
-        """view.map(f, *sequences, block=self.block, chunksize=1) => list|AsyncMapResult
+        """view.map(f, *sequences, block=self.block, chunksize=1, ordered=True) => list|AsyncMapResult
 
         Parallel version of builtin `map`, load-balanced by this View.
 
@@ -1009,14 +1009,20 b' class LoadBalancedView(View):'
             function to be mapped
         *sequences: one or more sequences of matching length
             the sequences to be distributed and passed to `f`
-        block : bool
-            whether to wait for the result or not [default self.block]
+        block : bool [default self.block]
+            whether to wait for the result or not
         track : bool
             whether to create a MessageTracker to allow the user to
             safely edit after arrays and buffers during non-copying
             sends.
-        chunksize : int
-            how many elements should be in each task [default 1]
+        chunksize : int [default 1]
+            how many elements should be in each task.
+        ordered : bool [default True]
+            Whether the results should be gathered as they arrive, or enforce
+            the order of submission.
+            
+            Only applies when iterating through AsyncMapResult as results arrive.
+            Has no effect when block=True.
 
         Returns
         -------
@@ -1034,6 +1040,7 b' class LoadBalancedView(View):'
         # default
         block = kwargs.get('block', self.block)
         chunksize = kwargs.get('chunksize', 1)
+        ordered = kwargs.get('ordered', True)
 
         keyset = set(kwargs.keys())
         extra_keys = keyset.difference_update(set(['block', 'chunksize']))
@@ -1042,7 +1049,7 b' class LoadBalancedView(View):'
 
         assert len(sequences) > 0, "must have some sequences to map onto!"
 
-        pf = ParallelFunction(self, f, block=block,  chunksize=chunksize)
+        pf = ParallelFunction(self, f, block=block, chunksize=chunksize, ordered=ordered)
         return pf.map(*sequences)
 
 __all__ = ['LoadBalancedView', 'DirectView']

IPython/parallel/tests/test_lbview.py

@@ -58,6 +58,42 b' class TestLoadBalancedView(ClusterTestCase):'
         r = self.view.map_sync(f, data)
         self.assertEquals(r, map(f, data))
 
+    def test_map_unordered(self):
+        def f(x):
+            return x**2
+        def slow_f(x):
+            import time
+            time.sleep(0.05*x)
+            return x**2
+        data = range(16,0,-1)
+        reference = map(f, data)
+        
+        amr = self.view.map_async(f, data, ordered=False)
+        self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
+        # check individual elements, retrieved as they come (uses __iter__)
+        astheycame = list(amr)
+        # Ensure that at least one result came out of order:
+        self.assertNotEquals(astheycame, reference, "should not have preserved order")
+        self.assertEquals(sorted(astheycame, reverse=True), reference, "result corrupted")
+
+    def test_map_ordered(self):
+        def f(x):
+            return x**2
+        def slow_f(x):
+            import time
+            time.sleep(0.05*x)
+            return x**2
+        data = range(16,0,-1)
+        reference = map(f, data)
+        
+        amr = self.view.map_async(f, data)
+        self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
+        # check individual elements, retrieved as they come (uses __iter__)
+        astheycame = list(amr)
+        # Ensure that results came in order
+        self.assertEquals(astheycame, reference)
+        self.assertEquals(amr.result, reference)
+    
     def test_abort(self):
         view = self.view
         ar = self.client[:].apply_async(time.sleep, .5)