DOP-2696: Pass exceptions through WorkerLauncher rather than deadlock

Author: i80and

snooty/parser.py

@@ -1229,15 +1229,16 @@ def merge_diagnostics(
 
     def flush(
         self,
-    ) -> queue.Queue[Union[PostprocessorResult, util.CancelledException]]:
+    ) -> queue.Queue[Tuple[Optional[PostprocessorResult], Optional[Exception]]]:
         """Run the postprocessor if and only if any pages have changed, and return postprocessing results."""
         postprocessor = self.postprocessor_factory()
         return self.worker.run(postprocessor)
 
     def flush_and_wait(self) -> PostprocessorResult:
-        result = self.flush().get()
-        if isinstance(result, util.CancelledException):
-            raise result
+        result, exception = self.flush().get()
+        if exception:
+            raise exception
+        assert result is not None
         return result
 
     def cancel(self) -> None:

snooty/test_util.py

@@ -171,3 +171,15 @@ def run_worker() -> None:
 
     finally:
         sys.setswitchinterval(initial_switch_interval)
+
+
+def test_worker_that_fails() -> None:
+    class SomeException(Exception):
+        pass
+
+    def start(event: threading.Event, arg: object) -> None:
+        raise SomeException()
+
+    worker = util.WorkerLauncher("worker-test", start)
+    with pytest.raises(SomeException):
+        worker.run_and_wait(None)

snooty/util.py

@@ -483,11 +483,13 @@ def __init__(
         self.__thread: Optional[threading.Thread] = None
         self.__cancel = threading.Event()
 
-    def run(self, arg: _T) -> queue.Queue[Union[_R, CancelledException]]:
+    def run(self, arg: _T) -> queue.Queue[Tuple[Optional[_R], Optional[Exception]]]:
         """Cancel any current thread of execution; block until it is cancelled; and re-launch the worker."""
         self.cancel()
 
-        result_queue: queue.Queue[Union[_R, CancelledException]] = queue.Queue(1)
+        result_queue: queue.Queue[
+            Tuple[Optional[_R], Optional[Exception]]
+        ] = queue.Queue(1)
 
         def inner() -> None:
             try:
@@ -497,9 +499,9 @@ def inner() -> None:
                 if self.__cancel.is_set():
                     raise CancelledException()
 
-                result_queue.put(result)
-            except CancelledException as cancelled:
-                result_queue.put(cancelled)
+                result_queue.put((result, None))
+            except Exception as exception:
+                result_queue.put((None, exception))
 
         thread = threading.Thread(name=self.name, target=inner, daemon=True)
         with self._lock:
@@ -509,11 +511,11 @@ def inner() -> None:
         return result_queue
 
     def run_and_wait(self, arg: _T) -> _R:
-        result = self.run(arg).get()
-        if isinstance(result, CancelledException):
-            raise result
+        result, exception = self.run(arg).get()
+        if exception:
+            raise exception
 
-        return result
+        return cast(_R, result)
 
     def cancel(self) -> None:
         """Instruct the worker to raise WorkerCanceled and abort execution."""