Fix ask#54: exits correctly on exception/ctrl-c

mode/threads.py

@@ -115,6 +115,11 @@ def __init__(self,
         super().__init__(loop=self.thread_loop, **kwargs)
         assert self._shutdown.loop is self.parent_loop
 
+    def __del__(self):
+        # Since services can restart, the thread loop must not be closed
+        # until the service is being fully removed, not just stopped.
+        self.thread_loop.close()
+
     async def on_thread_started(self) -> None:
         ...
 
@@ -175,8 +180,8 @@ async def start(self) -> None:
             self._thread_running = None
 
     async def crash(self, exc: BaseException) -> None:
-        if self._thread_running and not self._thread_running.done():
-            self._thread_running.set_exception(exc)  # <- .start() will raise
+        self.parent_loop.call_soon_threadsafe(
+            maybe_set_exception, self._thread_running, exc)
         await super().crash(exc)
 
     def _start_thread(self) -> None:
@@ -189,10 +194,14 @@ def _start_thread(self) -> None:
             # shutdown here, since _shutdown_thread will not execute.
             self.set_shutdown()
             raise
+        finally:
+            self.loop.stop()
 
     async def stop(self) -> None:
         if self._started.is_set():
             await super().stop()
+            if self._thread is not None:
+                self._thread.stop()
 
     async def _stop_children(self) -> None:
         ...   # called by thread instead of .stop()
@@ -208,8 +217,6 @@ async def _shutdown_thread(self) -> None:
         await self.on_thread_stop()
         self.set_shutdown()
         await self._default_stop_futures()
-        if self._thread is not None:
-            self._thread.stop()
         await self._default_stop_exit_stacks()
 
     async def _serve(self) -> None:

mode/utils/futures.py

@@ -155,7 +155,7 @@ def maybe_set_result(fut: Optional[asyncio.Future],
                      result: Any) -> bool:
     """Set future result if not already done."""
     if fut is not None and not fut.done():
-        fut.set_result(result)
+        notify(fut, result)
         return True
     return False
 
@@ -165,4 +165,5 @@ def notify(fut: Optional[asyncio.Future], result: Any = None) -> None:
     # can be used to turn a Future into a lockless, single-consumer condition,
     # for multi-consumer use asyncio.Condition
     if fut is not None and not fut.done():
-        fut.set_result(result)
+        loop = fut.get_loop()
+        loop.call_soon_threadsafe(fut.set_result, result)

mode/utils/locks.py

@@ -45,12 +45,13 @@ def set(self) -> None:
         All coroutines waiting for it to become true are awakened.
         Coroutine that call wait() once the flag is true will not block at all.
         """
+        from .futures import notify
+
         if not self._value:
             self._value = True
 
-            for fut in self._waiters:
-                if not fut.done():
-                    fut.set_result(True)
+            for fut in self._waiters.copy():
+                notify(fut, True)
 
     def clear(self) -> None:
         """Reset the internal flag to false.

t/functional/utils/test_futures.py

@@ -155,7 +155,9 @@ async def test_maybe_set_result():
     loop = asyncio.get_event_loop()
     future = loop.create_future()
     maybe_set_result(future, 42)
+    await asyncio.sleep(0.001)
     assert future.result() == 42
     maybe_set_result(future, 53)
     maybe_set_result(None, 57)
+    await asyncio.sleep(0.001)
     assert future.result() == 42

t/unit/test_supervisors.py

@@ -177,6 +177,8 @@ def test_wakeup(self, *, sup):
         wakeup = sup._please_wakeup = Mock()
         wakeup.done.return_value = False
         wakeup.cancelled.return_value = False
+        loop = wakeup.get_loop.return_value = Mock()
         sup.wakeup()
         assert sup._stopped.is_set()
-        wakeup.set_result.assert_called_with(None)
+        wakeup.get_loop.assert_called_once()
+        loop.call_soon_threadsafe.assert_called_with(wakeup.set_result, None)

t/unit/test_threads.py

@@ -161,7 +161,9 @@ async def test_stop(self, *, thread):
         await thread.stop()
         thread._started.set()
         thread._shutdown.set()
+        thread._thread = Mock()
         await thread.stop()
+        thread._thread.stop.assert_called_once_with()
 
     @pytest.mark.asyncio
     async def test_stop_children(self, *, thread):
@@ -183,10 +185,6 @@ async def test_shutdown_thread(self, *, thread):
         thread._default_stop_futures.assert_called_once_with()
         thread._shutdown.is_set()
 
-        thread._thread = Mock()
-        await thread._shutdown_thread()
-        thread._thread.stop.assert_called_once_with()
-
     @pytest.mark.asyncio
     async def test__thread_keepalive(self, *, thread):
 
@@ -262,6 +260,7 @@ async def test_crash(self, *, thread):
         thread._thread_running = Mock()
         thread._thread_running.done.return_value = False
         await thread.crash(exc)
+        await asyncio.sleep(0.001)
         thread._thread_running.set_exception.assert_called_with(exc)