Support return_exceptions=True in map, get

Allows exceptions to be returned instead of raised

so that partial errors can be handled and successful values from partial success extracted

Author: minrk

ipyparallel/client/asyncresult.py

@@ -73,6 +73,7 @@ def __init__(
         fname='unknown',
         targets=None,
         owner=False,
+        return_exceptions=False,
     ):
         super(AsyncResult, self).__init__()
         if not isinstance(children, list):
@@ -81,6 +82,8 @@ def __init__(
         else:
             self._single_result = False
 
+        self._return_exceptions = return_exceptions
+
         if isinstance(children[0], string_types):
             self.msg_ids = children
             self._children = []
@@ -204,22 +207,35 @@ def _reconstruct_result(self, res):
         else:
             return res
 
-    def get(self, timeout=-1):
+    def get(self, timeout=None, return_exceptions=None):
         """Return the result when it arrives.
 
-        If `timeout` is not ``None`` and the result does not arrive within
-        `timeout` seconds then ``TimeoutError`` is raised. If the
-        remote call raised an exception then that exception will be reraised
-        by get() inside a `RemoteError`.
+        Arguments:
+
+        timeout : int [default None]
+            If `timeout` is not ``None`` and the result does not arrive within
+            `timeout` seconds then ``TimeoutError`` is raised. If the
+            remote call raised an exception then that exception will be reraised
+            by get() inside a `RemoteError`.
+        return_exceptions : bool [default False]
+            If True, return Exceptions instead of raising them.
         """
         if not self.ready():
             self.wait(timeout)
 
+        if return_exceptions is None:
+            # default to attribute, if AsyncResult was created with return_exceptions=True
+            return_exceptions = self._return_exceptions
+
         if self._ready:
             if self._success:
                 return self.result()
             else:
-                raise self.exception()
+                e = self.exception()
+                if return_exceptions:
+                    return self._reconstruct_result(self._raw_results)
+                else:
+                    raise e
         else:
             raise error.TimeoutError("Result not ready.")
 
@@ -270,22 +286,37 @@ def wait(self, timeout=-1):
     def _resolve_result(self, f=None):
         if self.done():
             return
+        if f:
+            results = f.result()
+        else:
+            results = list(map(self._client.results.get, self.msg_ids))
+
+        # store raw results
+        self._raw_results = results
+
         try:
-            if f:
-                results = f.result()
-            else:
-                results = list(map(self._client.results.get, self.msg_ids))
             if self._single_result:
                 r = results[0]
                 if isinstance(r, Exception):
                     raise r
             else:
-                results = error.collect_exceptions(results, self._fname)
-            self._success = True
-            self.set_result(self._reconstruct_result(results))
+                results = self._collect_exceptions(results)
         except Exception as e:
             self._success = False
             self.set_exception(e)
+        else:
+            self._success = True
+            self.set_result(self._reconstruct_result(results))
+
+    def _collect_exceptions(self, results):
+        """Wrap Exceptions in a CompositeError
+
+        if self._return_exceptions is True, this is a no-op
+        """
+        if self._return_exceptions:
+            return results
+        else:
+            return error.collect_exceptions(results, self._fname)
 
     def _finalize_result(self, f):
         if self.owner:
@@ -424,10 +455,10 @@ def __getitem__(self, key):
         """getitem returns result value(s) if keyed by int/slice, or metadata if key is str."""
         if isinstance(key, int):
             self._check_ready()
-            return error.collect_exceptions([self.result()[key]], self._fname)[0]
+            return self._collect_exceptions([self.result()[key]])[0]
         elif isinstance(key, slice):
             self._check_ready()
-            return error.collect_exceptions(self.result()[key], self._fname)
+            return self._collect_exceptions(self.result()[key])
         elif isinstance(key, string_types):
             # metadata proxy *does not* require that results are done
             self.wait(0)
@@ -473,7 +504,7 @@ def __iter__(self):
             for child in self._children:
                 self._wait_for_child(child, evt=evt)
                 result = child.result()
-                error.collect_exceptions([result], self._fname)
+                self._collect_exceptions([result])
                 yield result
         else:
             # already done
@@ -583,15 +614,15 @@ def wait_interactive(self, interval=0.1, timeout=-1, widget=None):
             Override default context-detection behavior for whether a widget-based progress bar
             should be used.
         """
-        if timeout is None:
-            timeout = -1
+        if timeout and timeout < 0:
+            timeout = None
         N = len(self)
         tic = time.perf_counter()
         progress_bar = progress(widget=widget, total=N, unit='tasks', desc=self._fname)
 
         n_prev = 0
         while not self.ready() and (
-            timeout < 0 or time.perf_counter() - tic <= timeout
+            timeout is None or time.perf_counter() - tic <= timeout
         ):
             self.wait(interval)
             progress_bar.update(self.progress - n_prev)
@@ -751,25 +782,50 @@ def display_outputs(self, groupby="type", result_only=False):
 
 
 class AsyncMapResult(AsyncResult):
-    """Class for representing results of non-blocking gathers.
+    """Class for representing results of non-blocking maps.
 
-    This will properly reconstruct the gather.
+    AsyncMapResult.get() will properly reconstruct gathers into single object.
 
-    This class is iterable at any time, and will wait on results as they come.
+    AsyncMapResult 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, children, mapObject, fname='', ordered=True):
+    def __init__(
+        self,
+        client,
+        children,
+        mapObject,
+        fname='',
+        ordered=True,
+        return_exceptions=False,
+    ):
         self._mapObject = mapObject
         self.ordered = ordered
-        AsyncResult.__init__(self, client, children, fname=fname)
+        AsyncResult.__init__(
+            self,
+            client,
+            children,
+            fname=fname,
+            return_exceptions=return_exceptions,
+        )
         self._single_result = False
 
     def _reconstruct_result(self, res):
         """Perform the gather on the actual results."""
+        if self._return_exceptions:
+            if any(isinstance(r, Exception) for r in res):
+                # running with _return_exceptions,
+                # cannot reconstruct original
+                # use simple chain iterable
+                flattened = []
+                for r in res:
+                    if isinstance(r, Exception):
+                        flattened.append(r)
+                    else:
+                        flattened.extend(r)
+                return flattened
         return self._mapObject.joinPartitions(res)
 
     # asynchronous iterator:
@@ -786,7 +842,7 @@ def _yield_child_results(self, child):
         rlist = child.result()
         if not isinstance(rlist, list):
             rlist = [rlist]
-        error.collect_exceptions(rlist, self._fname)
+        self._collect_exceptions(rlist)
         for r in rlist:
             yield r
 
@@ -841,6 +897,8 @@ def _init_futures(self):
     def wait(self, timeout=-1):
         """wait for result to complete."""
         start = time.time()
+        if timeout and timeout < 0:
+            timeout = None
         if self._ready:
             return
         local_ids = [m for m in self.msg_ids if m in self._client.outstanding]
@@ -852,7 +910,7 @@ def wait(self, timeout=-1):
             else:
                 rdict = self._client.result_status(remote_ids, status_only=False)
                 pending = rdict['pending']
-                while pending and (timeout < 0 or time.time() < start + timeout):
+                while pending and (timeout is None or time.time() < start + timeout):
                     rdict = self._client.result_status(remote_ids, status_only=False)
                     pending = rdict['pending']
                     if pending:
@@ -865,11 +923,10 @@ def wait(self, timeout=-1):
                 results = list(map(self._client.results.get, self.msg_ids))
                 if self._single_result:
                     r = results[0]
-                    if isinstance(r, Exception):
+                    if isinstance(r, Exception) and not self._return_exceptions:
                         raise r
-                        self.set_result(r)
                 else:
-                    results = error.collect_exceptions(results, self._fname)
+                    results = self._collect_exceptions(results)
                 self._success = True
                 self.set_result(self._reconstruct_result(results))
             except Exception as e:

ipyparallel/client/remotefunction.py

@@ -186,6 +186,7 @@ class ParallelFunction(RemoteFunction):
     ordered : bool [default: True]
         Whether the result should be kept in order. If False,
         results become available as they arrive, regardless of submission order.
+    return_exceptions : bool [default: False]
     **flags
         remaining kwargs are passed to View.temp_flags
     """
@@ -195,11 +196,20 @@ class ParallelFunction(RemoteFunction):
     mapObject = None
 
     def __init__(
-        self, view, f, dist='b', block=None, chunksize=None, ordered=True, **flags
+        self,
+        view,
+        f,
+        dist='b',
+        block=None,
+        chunksize=None,
+        ordered=True,
+        return_exceptions=False,
+        **flags,
     ):
         super(ParallelFunction, self).__init__(view, f, block=block, **flags)
         self.chunksize = chunksize
         self.ordered = ordered
+        self.return_exceptions = return_exceptions
 
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
@@ -296,6 +306,7 @@ def __call__(self, *sequences, **kwargs):
             self.mapObject,
             fname=getname(self.func),
             ordered=self.ordered,
+            return_exceptions=self.return_exceptions,
         )
 
         if self.block:

ipyparallel/client/view.py

@@ -1216,7 +1216,15 @@ def _really_apply(
 
     @sync_results
     @save_ids
-    def map(self, f, *sequences, block=None, chunksize=1, ordered=True):
+    def map(
+        self,
+        f,
+        *sequences,
+        block=None,
+        chunksize=1,
+        ordered=True,
+        return_exceptions=False,
+    ):
         """Parallel version of builtin `map`, load-balanced by this View.
 
         `block`, and `chunksize` can be specified by keyword only.
@@ -1242,6 +1250,9 @@ def map(self, f, *sequences, block=None, chunksize=1, ordered=True):
             Only applies when iterating through AsyncMapResult as results arrive.
             Has no effect when block=True.
 
+        return_exceptions: bool [default False]
+            Return Exceptions instead of raising on the first exception.
+
         Returns
         -------
         if block=False
@@ -1260,7 +1271,12 @@ def map(self, f, *sequences, block=None, chunksize=1, ordered=True):
         assert len(sequences) > 0, "must have some sequences to map onto!"
 
         pf = ParallelFunction(
-            self, f, block=block, chunksize=chunksize, ordered=ordered
+            self,
+            f,
+            block=block,
+            chunksize=chunksize,
+            ordered=ordered,
+            return_exceptions=return_exceptions,
         )
         return pf.map(*sequences)
 
@@ -1270,6 +1286,7 @@ def imap(
         *sequences,
         ordered=True,
         max_outstanding='auto',
+        return_exceptions=False,
     ):
         """Parallel version of lazily-evaluated `imap`, load-balanced by this View.
 
@@ -1308,6 +1325,9 @@ def imap(
 
             Use this to tune how greedily input generator should be consumed.
 
+        return_exceptions : bool [default False]
+            Return Exceptions instead of raising them.
+
         Returns
         -------
 
@@ -1380,22 +1400,22 @@ def should_yield():
             # yielding immediately means
             if should_yield():
                 for ready_ar in wait_for_ready():
-                    yield ready_ar.get()
+                    yield ready_ar.get(return_exceptions=return_exceptions)
 
             # we've filled the buffer, wait for at least one result before continuing
             if len(outstanding) == max_outstanding:
                 for ready_ar in wait_for_ready():
-                    yield ready_ar.get()
+                    yield ready_ar.get(return_exceptions=return_exceptions)
 
         # yield any remaining results
         if ordered:
             for ar in outstanding:
-                yield ar.get()
+                yield ar.get(return_exceptions=return_exceptions)
         else:
             while outstanding:
                 done, outstanding = concurrent.futures.wait(outstanding)
                 for ar in done:
-                    yield ar.get()
+                    yield ar.get(return_exceptions=return_exceptions)
 
     def register_joblib_backend(self, name='ipyparallel', make_default=False):
         """Register this View as a joblib parallel backend

ipyparallel/tests/clienttest.py

@@ -137,7 +137,7 @@ def wait_on_engines(self, timeout=5):
 
     def client_wait(self, client, jobs=None, timeout=-1):
         """my wait wrapper, sets a default finite timeout to avoid hangs"""
-        if timeout < 0:
+        if timeout is None or timeout < 0:
             timeout = self.timeout
         return Client.wait(client, jobs, timeout)