Make futures raise exceptions with full traceback (fix #10)

docs/changes.rst

@@ -3,6 +3,20 @@ Changes
 =======
 
 
+v0.15 (in development)
+======================
+
+- Change the argument of :meth:`Future.set_exception()
+  <pykka.future.Future.set_exception>` from an exception instance to a
+  ``exc_info`` three-tuple. Passing just an exception instance to the method
+  still works, but it is deprecated and may be unsupported in a future release.
+
+- Due to the above change, :meth:`Future.get() <pykka.future.Future.get>` will
+  now reraise exceptions with complete traceback from the point when the
+  exception was first raised, and not just a traceback from when it was
+  reraised by :meth:`get`. (Fixes: :issue:`10`)
+
+
 v0.14 (2012-04-22)
 ==================
 

pykka/actor.py

@@ -172,7 +172,7 @@ def _run(self):
                 if 'reply_to' in message:
                     _logger.debug('Exception returned from %s to caller:' %
                         self, exc_info=_sys.exc_info())
-                    message['reply_to'].set_exception(exception)
+                    message['reply_to'].set_exception()
                 else:
                     self._handle_failure(*_sys.exc_info())
             except BaseException as exception:

pykka/future.py

@@ -1,3 +1,5 @@
+import sys as _sys
+
 try:
     # Python 2.x
     import Queue as _queue
@@ -51,12 +53,25 @@ def set(self, value=None):
         """
         raise NotImplementedError
 
-    def set_exception(self, exception):
+    def set_exception(self, exc_info=None):
         """
         Set an exception as the encapsulated value.
 
-        :param exception: the encapsulated exception
-        :type exception: exception
+        You can pass an ``exc_info`` three-tuple, as returned by
+        :func:`sys.exc_info`. If you don't pass ``exc_info``,
+        ``sys.exc_info()`` will be called and the value returned by it used.
+
+        In other words, if you're calling :meth:`set_exception`, without any
+        arguments, from an except block, the exception you're currently
+        handling will automatically be set on the future.
+
+        .. versionchanged:: 0.15
+            Previously, :meth:`set_exception` accepted an exception
+            instance as its only argument. This still works, but it is
+            deprecated and will be removed in a future release.
+
+        :param exc_info: the encapsulated exception
+        :type exc_info: three-tuple of (exc_class, exc_instance, traceback)
         """
         raise NotImplementedError
 
@@ -81,26 +96,27 @@ class ThreadingFuture(Future):
     def __init__(self):
         super(ThreadingFuture, self).__init__()
         self._queue = _queue.Queue()
-        self._value_received = False
-        self._value = None
+        self._data = None
 
     def get(self, timeout=None):
         try:
-            if not self._value_received:
-                self._value = self._queue.get(True, timeout)
-                self._value_received = True
-            if isinstance(self._value, BaseException):
-                raise self._value  # pylint: disable = E0702
+            if self._data is None:
+                self._data = self._queue.get(True, timeout)
+            if 'exc_info' in self._data:
+                exc_info = self._data['exc_info']
+                raise exc_info[0], exc_info[1], exc_info[2]
             else:
-                return self._value
+                return self._data['value']
         except _queue.Empty:
             raise _Timeout('%s seconds' % timeout)
 
     def set(self, value=None):
-        self._queue.put(value)
+        self._queue.put({'value': value})
 
-    def set_exception(self, exception):
-        self.set(exception)
+    def set_exception(self, exc_info=None):
+        if isinstance(exc_info, BaseException):
+            exc_info = (exc_info.__class__, exc_info, None)
+        self._queue.put({'exc_info': exc_info or _sys.exc_info()})
 
 
 def get_all(futures, timeout=None):

pykka/gevent.py

@@ -1,5 +1,7 @@
 from __future__ import absolute_import
 
+import sys as _sys
+
 # pylint: disable = E0611, W0406
 import gevent as _gevent
 import gevent.event as _gevent_event
@@ -39,7 +41,12 @@ def get(self, timeout=None):
     def set(self, value=None):
         self.async_result.set(value)
 
-    def set_exception(self, exception):
+    def set_exception(self, exc_info=None):
+        if isinstance(exc_info, BaseException):
+            exception = exc_info
+        else:
+            exc_info = exc_info or _sys.exc_info()
+            exception = exc_info[1]
         self.async_result.set_exception(exception)
 
 

tests/future_test.py

@@ -1,5 +1,6 @@
 import os
 import sys
+import traceback
 import unittest
 
 from pykka import Timeout
@@ -55,6 +56,36 @@ def test_future_in_future_works(self):
         outer_future.set(inner_future)
         self.assertEqual(outer_future.get().get(), 'foo')
 
+    def test_get_raises_exception_with_full_traceback(self):
+        future = self.future_class()
+
+        try:
+            raise NameError('foo')
+        except NameError as error:
+            exc_class_set, exc_instance_set, exc_traceback_set = sys.exc_info()
+            future.set_exception()
+
+        # We could move to another thread at this point
+
+        try:
+            future.get()
+        except NameError:
+            exc_class_get, exc_instance_get, exc_traceback_get = sys.exc_info()
+
+        self.assertEqual(exc_class_set, exc_class_get)
+        self.assertEqual(exc_instance_set, exc_instance_get)
+
+        exc_traceback_list_set = list(reversed(
+            traceback.extract_tb(exc_traceback_set)))
+        exc_traceback_list_get = list(reversed(
+            traceback.extract_tb(exc_traceback_get)))
+
+        # All frames from the first traceback should be included in the
+        # traceback from the future.get() reraise
+        self.assert_(len(exc_traceback_list_set) < len(exc_traceback_list_get))
+        for i, frame in enumerate(exc_traceback_list_set):
+            self.assertEquals(frame, exc_traceback_list_get[i])
+
 
 class ThreadingFutureTest(FutureTest, unittest.TestCase):
     future_class = ThreadingFuture
@@ -71,3 +102,8 @@ def test_can_wrap_existing_async_result(self):
             async_result = AsyncResult()
             future = GeventFuture(async_result)
             self.assertEquals(async_result, future.async_result)
+
+        def test_get_raises_exception_with_full_traceback(self):
+            # gevent prints the first half of the traceback instead of
+            # passing it through to the other side of the AsyncResult
+            pass