Re-raise queued exceptions with traceback

salt/transport/ipc.py

@@ -11,6 +11,7 @@ import socket
 import weakref
 import time
 import threading
+import sys
 
 # Import 3rd-party libs
 import msgpack
@@ -85,6 +86,11 @@ class FutureWithTimeout(tornado.concurrent.Future):
             self.set_exception(exc)
 
 
+class IPCExceptionProxy(object):
+    def __init__(self, orig_info):
+        self.orig_info = orig_info
+
+
 class IPCServer(object):
     '''
     A Tornado IPC server very similar to Tornado's TCPServer class
@@ -694,6 +700,7 @@ class IPCMessageSubscriberService(IPCClient):
                 break
             except Exception as exc:
                 log.error('Exception occurred in Subscriber while handling stream: %s', exc)
+                exc = IPCExceptionProxy(sys.exc_info())
                 self._feed_subscribers([exc])
                 break
 
@@ -801,13 +808,19 @@ class IPCMessageSubscriber(object):
                 raise tornado.gen.Return(None)
             if data is None:
                 break
-            elif isinstance(data, Exception):
-                raise data
+            elif isinstance(data, IPCExceptionProxy):
+                self.reraise(data.orig_info)
             elif callback:
                 self.service.io_loop.spawn_callback(callback, data)
             else:
                 raise tornado.gen.Return(data)
 
+    def reraise(self, exc_info):
+        if six.PY2:
+            raise exc_info[0], exc_info[1], exc_info[2]
+        else:
+            raise exc_info[0].with_traceback(exc_info[1], exc_info[2])
+
     def read_sync(self, timeout=None):
         '''
         Read a message from an IPC socket