Send the full traceback with any exceptions in the queue process

util/queuefile.py

@@ -31,8 +31,14 @@ class QueueFile(object):
       if isinstance(result, Exception):
         self._closed = True
         self.raised_exception = True
+
+        handled = False
         for handler in self._exception_handlers:
           handler(result)
+          handled = True
+        
+        if handled:
+          return
 
         raise result
 

util/queueprocess.py

@@ -4,6 +4,8 @@ import multiprocessing
 import os
 import time
 import gipc
+import sys
+import traceback
 
 logger = multiprocessing.log_to_stderr()
 logger.setLevel(logging.INFO)
@@ -42,8 +44,8 @@ def _run(get_producer, queues, chunk_size, args):
   while True:
     try:
       data = producer(chunk_size) or None
-    except Exception as ex:
+    except Exception:
-      data = ex
+      data = Exception("".join(traceback.format_exception(*sys.exc_info())))
 
     for queue in queues:
       try: