Repository: agronholm/anyio

Issue:
to_thread.run_sync() doesn't work with the tornado event loop
Hi there!

[jupyter_server](https://github.com/jupyter-server/jupyter_server)  uses the anyio v2 `run_sync_in_worker_thread` function , and I noticed a possible bug after upgrading to anyio v3. 

This is the error I see when using `to_thread.run_sync()`:
```python
HTTPServerRequest(protocol='http', host='127.0.0.1:38403', method='GET', uri='/a%40b/api/contents/Directory%20with%20spaces%20in', version='HTTP/1.1', remote_ip='127.0.0.1')
Traceback (most recent call last):
  File "/opt/hostedtoolcache/Python/3.7.10/x64/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 697, in run_sync_in_worker_thread
    work_queue = _threadpool_work_queue.get()
  File "/opt/hostedtoolcache/Python/3.7.10/x64/lib/python3.7/site-packages/anyio/lowlevel.py", line 120, in get
    raise LookupError(f'Run variable "{self._name}" has no value and no default set')
LookupError: Run variable "_threadpool_work_queue" has no value and no default set

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/hostedtoolcache/Python/3.7.10/x64/lib/python3.7/site-packages/tornado/web.py", line 1704, in _execute
    result = await result
  File "/home/runner/work/jupyter_server/jupyter_server/jupyter_server/services/contents/handlers.py", line 111, in get
    path=path, type=type, format=format, content=content,
  File "/home/runner/work/jupyter_server/jupyter_server/jupyter_server/utils.py", line 176, in ensure_async
    result = await obj
  File "/home/runner/work/jupyter_server/jupyter_server/jupyter_server/services/contents/filemanager.py", line 692, in get
    model = await self._dir_model(path, content=content)
  File "/home/runner/work/jupyter_server/jupyter_server/jupyter_server/services/contents/filemanager.py", line 574, in _dir_model
    dir_contents = await run_sync_in_worker_thread(os.listdir, os_dir)
  File "/opt/hostedtoolcache/Python/3.7.10/x64/lib/python3.7/site-packages/anyio/to_thread.py", line 37, in run_sync_in_worker_thread
    return await run_sync(func, *args, cancellable=cancellable, limiter=limiter)
  File "/opt/hostedtoolcache/Python/3.7.10/x64/lib/python3.7/site-packages/anyio/to_thread.py", line 29, in run_sync
    limiter=limiter)
  File "/opt/hostedtoolcache/Python/3.7.10/x64/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 707, in run_sync_in_worker_thread
    find_root_task().add_done_callback(_loop_shutdown_callback)
  File "/opt/hostedtoolcache/Python/3.7.10/x64/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 56, in find_root_task
Error:     raise RuntimeError('Cannot find root task for setting cleanup callback')
```
It looks like the `find_root_task` function is failing because the callback is a tornado.ioloop and not an uvloop.loop.

Jupyter server is using the tornado event loop.
For more context, this is the github issue: https://github.com/jupyter-server/jupyter_server/issues/487

**Possible Solution**

I was able to fix it by updating the following code, but I'm not sure if this is a proper fix.  
[src/anyio/_backends/_asyncio.py line 50](https://github.com/agronholm/anyio/pull/227/files#diff-202c3fc4551e5a287df488a9bd96a25dcc1405ec135d5c7aceb6ef4c02755341R50) 

```python
if cb is _run_until_complete_cb or cb.__module__ in ('uvloop.loop', 'tornado.ioloop')
```


Discussion:
mwakaba2: @agronholm The tests are passing after applying your PR. I attached my PR for reference. Thanks so much for the quick refactor! 🙂
agronholm: > @agronholm The tests are passing after applying your PR. I attached my PR for reference. Thanks so much for the quick refactor!

Good to hear. I'd still like to get one of the collaborators to review these changes since they're nontrivial. I almost introduced a subtle bug during the development of this PR which would have caused worker threads in some edge cases to be exempted from shutdown/cleanup.

Patch:
diff --git a/docs/synchronization.rst b/docs/synchronization.rst
index e5400df..b737b54 100644
--- a/docs/synchronization.rst
+++ b/docs/synchronization.rst
@@ -119,15 +119,15 @@ Example::
 
             await sleep(1)
             async with condition:
-                await condition.notify(1)
+                condition.notify(1)
 
             await sleep(1)
             async with condition:
-                await condition.notify(2)
+                condition.notify(2)
 
             await sleep(1)
             async with condition:
-                await condition.notify_all()
+                condition.notify_all()
 
     run(main)
 
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index e3eb2a1..95e4de2 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -3,6 +3,11 @@ Version history
 
 This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
+**UNRELEASED**
+
+- Fixed ``to_thread.run_sync()`` raising ``RuntimeError`` when no "root" task could be found for
+  setting up a cleanup callback
+
 **3.0.0**
 
 - Curio support has been dropped (see the :doc:`FAQ <faq>` as for why)
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 68fac2e..02053e4 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -17,8 +17,8 @@ from socket import AddressFamily, SocketKind, SocketType
 from threading import Thread
 from types import TracebackType
 from typing import (
-    Any, Awaitable, Callable, Collection, Coroutine, Deque, Dict, Generator, List, Optional,
-    Sequence, Set, Tuple, Type, TypeVar, Union, cast)
+    Any, Awaitable, Callable, Collection, Coroutine, Deque, Dict, Generator, Iterable, List,
+    Optional, Sequence, Set, Tuple, Type, TypeVar, Union, cast)
 from weakref import WeakKeyDictionary
 
 from .. import CapacityLimiterStatistics, EventStatistics, TaskInfo, abc
@@ -46,18 +46,14 @@ if sys.version_info >= (3, 7):
     from asyncio import all_tasks, create_task, current_task, get_running_loop
     from asyncio import run as native_run
 
-    def find_root_task() -> asyncio.Task:
-        for task in all_tasks():
-            if task._callbacks:
-                for cb, context in task._callbacks:  # type: ignore
-                    if cb is _run_until_complete_cb or cb.__module__ == 'uvloop.loop':
-                        return task
-
-        raise RuntimeError('Cannot find root task for setting cleanup callback')
+    def _get_task_callbacks(task: asyncio.Task) -> Iterable[Callable]:
+        return [cb for cb, context in task._callbacks]  # type: ignore
 else:
-
     _T = TypeVar('_T')
 
+    def _get_task_callbacks(task: asyncio.Task) -> Iterable[Callable]:
+        return task._callbacks  # type: ignore
+
     def native_run(main, *, debug=False):
         # Snatched from Python 3.7
         from asyncio import coroutines, events, tasks
@@ -129,20 +125,43 @@ else:
 
         return asyncio.Task.current_task(loop)
 
-    def find_root_task() -> asyncio.Task:
-        for task in all_tasks():
-            for cb in task._callbacks:
-                if cb is _run_until_complete_cb or cb.__module__ == 'uvloop.loop':
-                    return task
-
-        raise RuntimeError('Cannot find root task for setting cleanup callback')
-
 T_Retval = TypeVar('T_Retval')
 
 # Check whether there is native support for task names in asyncio (3.8+)
 _native_task_names = hasattr(asyncio.Task, 'get_name')
 
-WORKER_THREAD_MAX_IDLE_TIME = 10  # seconds
+
+_root_task: RunVar[Optional[asyncio.Task]] = RunVar('_root_task')
+
+
+def find_root_task() -> asyncio.Task:
+    try:
+        root_task = _root_task.get()
+    except LookupError:
+        for task in all_tasks():
+            if task._callbacks:
+                for cb in _get_task_callbacks(task):
+                    if cb is _run_until_complete_cb or cb.__module__ == 'uvloop.loop':
+                        _root_task.set(task)
+                        return task
+
+        _root_task.set(None)
+    else:
+        if root_task is not None:
+            return root_task
+
+    # Look up the topmost task in the AnyIO task tree, if possible
+    task = cast(asyncio.Task, current_task())
+    state = _task_states.get(task)
+    if state:
+        cancel_scope = state.cancel_scope
+        while cancel_scope and cancel_scope._parent_scope is not None:
+            cancel_scope = cancel_scope._parent_scope
+
+        if cancel_scope is not None:
+            return cast(asyncio.Task, cancel_scope._host_task)
+
+    return task
 
 
 def get_callable_name(func: Callable) -> str:
@@ -650,41 +669,51 @@ class TaskGroup(abc.TaskGroup):
 _Retval_Queue_Type = Tuple[Optional[T_Retval], Optional[BaseException]]
 
 
-def _thread_pool_worker(work_queue: Queue, loop: asyncio.AbstractEventLoop) -> None:
-    func: Callable
-    args: tuple
-    future: asyncio.Future
-    with claim_worker_thread('asyncio'):
-        loop = threadlocals.loop = loop
-        while True:
-            func, args, future = work_queue.get()
-            if func is None:
-                # Shutdown command received
-                return
+class WorkerThread(Thread):
+    __slots__ = 'root_task', 'loop', 'queue', 'idle_since'
 
-            if not future.cancelled():
-                try:
-                    result = func(*args)
-                except BaseException as exc:
-                    if not loop.is_closed() and not future.cancelled():
-                        loop.call_soon_threadsafe(future.set_exception, exc)
-                else:
-                    if not loop.is_closed() and not future.cancelled():
-                        loop.call_soon_threadsafe(future.set_result, result)
+    MAX_IDLE_TIME = 10  # seconds
+
+    def __init__(self, root_task: asyncio.Task):
+        super().__init__(name='AnyIO worker thread')
+        self.root_task = root_task
+        self.loop = root_task._loop
+        self.queue: Queue[Union[Tuple[Callable, tuple, asyncio.Future], None]] = Queue(2)
+        self.idle_since = current_time()
+
+    def run(self) -> None:
+        with claim_worker_thread('asyncio'):
+            threadlocals.loop = self.loop
+            while True:
+                item = self.queue.get()
+                if item is None:
+                    # Shutdown command received
+                    return
 
-            work_queue.task_done()
+                func, args, future = item
+                if not future.cancelled():
+                    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)
 
+                self.queue.task_done()
 
-_threadpool_work_queue: RunVar[Queue] = RunVar('_threadpool_work_queue')
-_threadpool_idle_workers: RunVar[Deque[Tuple[Thread, float]]] = RunVar(
-    '_threadpool_idle_workers')
-_threadpool_workers: RunVar[Set[Thread]] = RunVar('_threadpool_workers')
+    def stop(self, f: Optional[asyncio.Task] = None) -> None:
+        self.queue.put_nowait(None)
+        _threadpool_workers.get().discard(self)
+        try:
+            _threadpool_idle_workers.get().remove(self)
+        except ValueError:
+            pass
 
 
-def _loop_shutdown_callback(f: asyncio.Future) -> None:
-    """This is called when the root task has finished."""
-    for _ in range(len(_threadpool_workers.get())):
-        _threadpool_work_queue.get().put_nowait((None, None, None))
+_threadpool_idle_workers: RunVar[Deque[WorkerThread]] = RunVar('_threadpool_idle_workers')
+_threadpool_workers: RunVar[Set[WorkerThread]] = RunVar('_threadpool_workers')
 
 
 async def run_sync_in_worker_thread(
@@ -694,45 +723,42 @@ async def run_sync_in_worker_thread(
 
     # If this is the first run in this event loop thread, set up the necessary variables
     try:
-        work_queue = _threadpool_work_queue.get()
         idle_workers = _threadpool_idle_workers.get()
         workers = _threadpool_workers.get()
     except LookupError:
-        work_queue = Queue()
         idle_workers = deque()
         workers = set()
-        _threadpool_work_queue.set(work_queue)
         _threadpool_idle_workers.set(idle_workers)
         _threadpool_workers.set(workers)
-        find_root_task().add_done_callback(_loop_shutdown_callback)
 
     async with (limiter or current_default_thread_limiter()):
         with CancelScope(shield=not cancellable):
             future: asyncio.Future = asyncio.Future()
-            work_queue.put_nowait((func, args, future))
+            root_task = find_root_task()
             if not idle_workers:
-                thread = Thread(target=_thread_pool_worker, args=(work_queue, get_running_loop()),
-                                name='AnyIO worker thread')
-                workers.add(thread)
-                thread.start()
+                worker = WorkerThread(root_task)
+                worker.start()
+                workers.add(worker)
+                root_task.add_done_callback(worker.stop)
             else:
-                thread, idle_since = idle_workers.pop()
+                worker = idle_workers.pop()
 
-                # Prune any other workers that have been idle for WORKER_MAX_IDLE_TIME seconds or
-                # longer
+                # Prune any other workers that have been idle for MAX_IDLE_TIME seconds or longer
                 now = current_time()
                 while idle_workers:
-                    if now - idle_workers[0][1] < WORKER_THREAD_MAX_IDLE_TIME:
+                    if now - idle_workers[0].idle_since < WorkerThread.MAX_IDLE_TIME:
                         break
 
-                    idle_workers.popleft()
-                    work_queue.put_nowait(None)
-                    workers.remove(thread)
+                    worker = idle_workers.popleft()
+                    worker.root_task.remove_done_callback(worker.stop)
+                    worker.stop()
 
+            worker.queue.put_nowait((func, args, future))
             try:
                 return await future
             finally:
-                idle_workers.append((thread, current_time()))
+                worker.idle_since = current_time()
+                idle_workers.append(worker)
 
 
 def run_sync_from_thread(func: Callable[..., T_Retval], *args,


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 92fc808..faf5c10 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -49,3 +49,12 @@ def client_context(ca):
     client_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
     ca.configure_trust(client_context)
     return client_context
+
+
+@pytest.fixture
+def asyncio_event_loop():
+    loop = asyncio.DefaultEventLoopPolicy().new_event_loop()
+    asyncio.set_event_loop(loop)
+    yield loop
+    asyncio.set_event_loop(None)
+    loop.close()
diff --git a/tests/test_debugging.py b/tests/test_debugging.py
index 19a9179..80af499 100644
--- a/tests/test_debugging.py
+++ b/tests/test_debugging.py
@@ -53,9 +53,7 @@ async def test_get_running_tasks():
 
 
 @pytest.mark.filterwarnings('ignore:"@coroutine" decorator is deprecated:DeprecationWarning')
-def test_wait_generator_based_task_blocked():
-    from asyncio import DefaultEventLoopPolicy, Event, coroutine, set_event_loop
-
+def test_wait_generator_based_task_blocked(asyncio_event_loop):
     async def native_coro_part():
         await wait_all_tasks_blocked()
         assert not gen_task._coro.gi_running
@@ -66,19 +64,13 @@ def test_wait_generator_based_task_blocked():
 
         event.set()
 
-    @coroutine
+    @asyncio.coroutine
     def generator_part():
         yield from event.wait()
 
-    loop = DefaultEventLoopPolicy().new_event_loop()
-    try:
-        set_event_loop(loop)
-        event = Event()
-        gen_task = loop.create_task(generator_part())
-        loop.run_until_complete(native_coro_part())
-    finally:
-        set_event_loop(None)
-        loop.close()
+    event = asyncio.Event()
+    gen_task = asyncio_event_loop.create_task(generator_part())
+    asyncio_event_loop.run_until_complete(native_coro_part())
 
 
 @pytest.mark.parametrize('anyio_backend', ['asyncio'])
diff --git a/tests/test_from_thread.py b/tests/test_from_thread.py
index d169b80..c596cb1 100644
--- a/tests/test_from_thread.py
+++ b/tests/test_from_thread.py
@@ -1,4 +1,5 @@
 import threading
+import time
 from concurrent.futures import CancelledError
 from contextlib import suppress
 
@@ -53,7 +54,14 @@ class TestRunAsyncFromThread:
         # The thread should not exist after the event loop has been closed
         initial_count = threading.active_count()
         run(main, backend='asyncio')
-        assert threading.active_count() == initial_count
+
+        for _ in range(10):
+            if threading.active_count() == initial_count:
+                return
+
+            time.sleep(0.1)
+
+        pytest.fail('Worker thread did not exit within 1 second')
 
     async def test_run_async_from_thread_exception(self):
         async def add(a, b):
diff --git a/tests/test_to_thread.py b/tests/test_to_thread.py
index c538b95..e39c052 100644
--- a/tests/test_to_thread.py
+++ b/tests/test_to_thread.py
@@ -84,6 +84,7 @@ async def test_cancel_worker_thread(cancellable, expected_last_active):
 
     """
     def thread_worker():
+        print('thread worker:', threading.current_thread(), flush=True)
         nonlocal last_active
         from_thread.run_sync(sleep_event.set)
         time.sleep(0.2)
@@ -110,14 +111,40 @@ async def test_cancel_worker_thread(cancellable, expected_last_active):
 
 
 @pytest.mark.parametrize('anyio_backend', ['asyncio'])
-async def test_cancel_asyncio_native_task():
+async def test_asyncio_cancel_native_task():
     async def run_in_thread():
         nonlocal task
         task = current_task()
-        await to_thread.run_sync(time.sleep, 1, cancellable=True)
+        await to_thread.run_sync(time.sleep, 0.2, cancellable=True)
 
     task = None
     async with create_task_group() as tg:
         tg.start_soon(run_in_thread)
         await wait_all_tasks_blocked()
         task.cancel()
+
+
+def test_asyncio_no_root_task(asyncio_event_loop):
+    """
+    Regression test for #264.
+
+    Ensures that to_thread.run_sync() does not raise an error when there is no root task, but
+    instead tries to find the top most parent task by traversing the cancel scope tree, or failing
+    that, uses the current task to set up a shutdown callback.
+
+    """
+    async def run_in_thread():
+        try:
+            await to_thread.run_sync(time.sleep, 0)
+        finally:
+            asyncio_event_loop.call_soon(asyncio_event_loop.stop)
+
+    task = asyncio_event_loop.create_task(run_in_thread())
+    asyncio_event_loop.run_forever()
+    task.result()
+
+    # Wait for worker threads to exit
+    for t in threading.enumerate():
+        if t.name == 'AnyIO worker thread':
+            t.join(2)
+            assert not t.is_alive()