Repository: agronholm/anyio

Issue:
asyncio worker threads are prematurely shut-down on SIGINT
```python
import anyio
import time

def thread():
    print("thread start", time.time())
    time.sleep(5)
    print("thread end", time.time())

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(anyio.to_thread.run_sync, thread)
        try:
            await anyio.sleep_forever()
        finally:
            await anyio.to_thread.run_sync(thread)

anyio.run(main, backend="asyncio")
```
If i run this and press CTRL-C while sleeping, it does this:

```
thread start 1623770620.0201666
^Cthread end 1623770625.0241108
thread start 1623770625.0242603
thread end 1623770630.028106
```

with `backend="trio"` I get what I expect:
```
thread start 1623770659.6473012
^Cthread start 1623770660.5315428
thread end 1623770664.652128
thread end 1623770665.5361295
```

One workaround for this issue is to use a signal receiver:
```python
import anyio
import time
import signal

def thread():
    print("thread start", time.time())
    time.sleep(5)
    print("thread end", time.time())

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(anyio.to_thread.run_sync, thread)
        try:
            with anyio.open_signal_receiver(signal.SIGINT) as signals:
                async for signum in signals:
                    break
        finally:
            await anyio.to_thread.run_sync(thread)

anyio.run(main, backend="asyncio")
```

Another workaround (which I'm using) is to spawn threads manually (`threading.Thread(...).start()`) for tear-down.  Both of these are asyncio-specific workarounds, and I think that anyio `to_thread.run_sync` should behave the same if running on asyncio or trio, if possible.

My use case is this is working with something that is synchronous and must use worker threads.  I must call the appropriate `close()` methods on the synchronous API from a thread so as to not block my async application.

Discussion:
agronholm: This PR still lacks a regression test.

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 57e4e92..509bd8b 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -10,6 +10,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   synchronous context manager)
 - Fixed ``to_thread.run_sync()`` hanging on the second call on asyncio when used with
   ``loop.run_until_complete()``
+- Fixed ``to_thread.run_sync()`` prematurely marking a worker thread inactive when a task await on
+  the result is cancelled
 - Changed the default value of the ``use_uvloop`` asyncio backend option to ``False`` to prevent
   unsafe event loop policy changes in different threads
 - Fixed the type annotation of ``open_signal_receiver()`` as a synchronous context manager
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 11bee85..556c2b9 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -712,6 +712,16 @@ class WorkerThread(Thread):
         self.queue: Queue[Union[Tuple[Callable, tuple, asyncio.Future], None]] = Queue(2)
         self.idle_since = current_time()
 
+    def _report_result(self, future: asyncio.Future, result: Any,
+                       exc: Optional[BaseException]) -> None:
+        self.idle_since = current_time()
+        self.idle_workers.append(self)
+        if not future.cancelled():
+            if exc is not None:
+                future.set_exception(exc)
+            else:
+                future.set_result(result)
+
     def run(self) -> None:
         with claim_worker_thread('asyncio'):
             threadlocals.loop = self.loop
@@ -723,14 +733,16 @@ class WorkerThread(Thread):
 
                 func, args, future = item
                 if not future.cancelled():
+                    result = None
+                    exception: Optional[BaseException] = None
                     try:
                         result = func(*args)
                     except BaseException as exc:
-                        if not self.loop.is_closed() and not future.cancelled():
-                            self.loop.call_soon_threadsafe(future.set_exception, exc)
-                    else:
-                        if not self.loop.is_closed() and not future.cancelled():
-                            self.loop.call_soon_threadsafe(future.set_result, result)
+                        exception = exc
+
+                    if not self.loop.is_closed():
+                        self.loop.call_soon_threadsafe(
+                            self._report_result, future, result, exception)
 
                 self.queue.task_done()
 
@@ -785,12 +797,7 @@ async def run_sync_in_worker_thread(
                     worker.stop()
 
             worker.queue.put_nowait((func, args, future))
-            try:
-                return await future
-            finally:
-                assert worker.is_alive()
-                worker.idle_since = current_time()
-                idle_workers.append(worker)
+            return await future
 
 
 def run_sync_from_thread(func: Callable[..., T_Retval], *args: object,


Test patch:
diff --git a/tests/test_to_thread.py b/tests/test_to_thread.py
index 5880ab4..2e8f011 100644
--- a/tests/test_to_thread.py
+++ b/tests/test_to_thread.py
@@ -2,6 +2,7 @@ import asyncio
 import sys
 import threading
 import time
+from concurrent.futures import Future
 from functools import partial
 from typing import Any, List, NoReturn, Optional
 
@@ -114,6 +115,22 @@ async def test_cancel_worker_thread(cancellable: bool, expected_last_active: str
     assert last_active == expected_last_active
 
 
+async def test_cancel_wait_on_thread() -> None:
+    event = threading.Event()
+    future: Future[bool] = Future()
+
+    def wait_event() -> None:
+        future.set_result(event.wait(1))
+
+    async with create_task_group() as tg:
+        tg.start_soon(partial(to_thread.run_sync, cancellable=True), wait_event)
+        await wait_all_tasks_blocked()
+        tg.cancel_scope.cancel()
+
+    await to_thread.run_sync(event.set)
+    assert future.result(1)
+
+
 @pytest.mark.parametrize('anyio_backend', ['asyncio'])
 async def test_asyncio_cancel_native_task() -> None:
     task: "Optional[asyncio.Task[None]]" = None