notice nesting of `sync_results` decorator

ensures the sync operation only fires once, after the outermost call.

avoids O(n) calls to set.difference

IPython/parallel/client/remotefunction.py

@@ -21,6 +21,7 b' from __future__ import division'
 import sys
 import warnings
 
+from IPython.external.decorator import decorator
 from IPython.testing.skipdoctest import skip_doctest
 
 from . import map as Map
@@ -79,6 +80,24 b' def getname(f):'
     
     return str(f)
 
+@decorator
+def sync_view_results(f, self, *args, **kwargs):
+    """sync relevant results from self.client to our results attribute.
+    
+    This is a clone of view.sync_results, but for remote functions
+    """
+    view = self.view
+    if view._in_sync_results:
+        return f(self, *args, **kwargs)
+    print 'in sync results', f
+    view._in_sync_results = True
+    try:
+        ret = f(self, *args, **kwargs)
+    finally:
+        view._in_sync_results = False
+        view._sync_results()
+    return ret
+    
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
@@ -158,6 +177,7 b' class ParallelFunction(RemoteFunction):'
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
 
+    @sync_view_results
     def __call__(self, *sequences):
         client = self.view.client
         

IPython/parallel/client/view.py

@@ -56,10 +56,14 b' def save_ids(f, self, *args, **kwargs):'
 @decorator
 def sync_results(f, self, *args, **kwargs):
     """sync relevant results from self.client to our results attribute."""
-    ret = f(self, *args, **kwargs)
-    delta = self.outstanding.difference(self.client.outstanding)
-    completed = self.outstanding.intersection(delta)
-    self.outstanding = self.outstanding.difference(completed)
+    if self._in_sync_results:
+        return f(self, *args, **kwargs)
+    self._in_sync_results = True
+    try:
+        ret = f(self, *args, **kwargs)
+    finally:
+        self._in_sync_results = False
+        self._sync_results()
     return ret
 
 @decorator
@@ -115,6 +119,7 b' class View(HasTraits):'
 
     _socket = Instance('zmq.Socket')
     _flag_names = List(['targets', 'block', 'track'])
+    _in_sync_results = Bool(False)
     _targets = Any()
     _idents = Any()
 
@@ -198,6 +203,15 b' class View(HasTraits):'
     # apply
     #----------------------------------------------------------------
 
+    def _sync_results(self):
+        """to be called by @sync_results decorator
+        
+        after submitting any tasks.
+        """
+        delta = self.outstanding.difference(self.client.outstanding)
+        completed = self.outstanding.intersection(delta)
+        self.outstanding = self.outstanding.difference(completed)
+        
     @sync_results
     @save_ids
     def _really_apply(self, f, args, kwargs, block=None, **options):
@@ -323,6 +337,7 b' class View(HasTraits):'
     # Map
     #-------------------------------------------------------------------
 
+    @sync_results
     def map(self, f, *sequences, **kwargs):
         """override in subclasses"""
         raise NotImplementedError
@@ -554,7 +569,7 b' class DirectView(View):'
         return ar
 
 
-    @spin_after
+    @sync_results
     def map(self, f, *sequences, **kwargs):
         """view.map(f, *sequences, block=self.block) => list|AsyncMapResult
 
@@ -1031,7 +1046,7 b' class LoadBalancedView(View):'
                 pass
         return ar
 
-    @spin_after
+    @sync_results
     @save_ids
     def map(self, f, *sequences, **kwargs):
         """view.map(f, *sequences, block=self.block, chunksize=1, ordered=True) => list|AsyncMapResult