Repository: agronholm/anyio

Issue:
`anyio.to_thread.run_sync()` hangs IPython when using top-level await expressions
Not sure if this is a problem with IPython, `anyio` or something else.

If you launch a IPython shell, like so:
```bash
$ ipython
Python 3.10.0b2+ (heads/3.10:9c89d62, Jun  2 2021, 20:22:16) [GCC 10.3.0]
Type 'copyright', 'credits' or 'license' for more information
IPython 7.24.1 -- An enhanced Interactive Python. Type '?' for help.

In [1]: 
```

And paste this example in, it will run once:
```python3
from time import sleep
from anyio.to_thread import run_sync


def sync_func(time: float):
  sleep(time)
  print(f'Slept {time} seconds.')


await run_sync(sync_func, 0.5)
```

After it finishes executing, if you await another `anyio.to_thread.run_sync()` coroutine, it will hang the session:

```python3
In [6]: await run_sync(sync_func, 0.5)
```

Here's the stacktrace when you hit Ctrl-C:
```python3
In [6]: await run_sync(sync_func, 0.5)
^C---------------------------------------------------------------------------
KeyboardInterrupt                         Traceback (most recent call last)
~/.pyenv/versions/3.10-dev/lib/python3.10/site-packages/IPython/core/async_helpers.py in __call__(self, coro)
     26         import asyncio
     27 
---> 28         return asyncio.get_event_loop().run_until_complete(coro)
     29 
     30     def __str__(self):

~/.pyenv/versions/3.10-dev/lib/python3.10/asyncio/base_events.py in run_until_complete(self, future)
    626         future.add_done_callback(_run_until_complete_cb)
    627         try:
--> 628             self.run_forever()
    629         except:
    630             if new_task and future.done() and not future.cancelled():

~/.pyenv/versions/3.10-dev/lib/python3.10/asyncio/base_events.py in run_forever(self)
    593             events._set_running_loop(self)
    594             while True:
--> 595                 self._run_once()
    596                 if self._stopping:
    597                     break

~/.pyenv/versions/3.10-dev/lib/python3.10/asyncio/base_events.py in _run_once(self)
   1843             timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT)
   1844 
-> 1845         event_list = self._selector.select(timeout)
   1846         self._process_events(event_list)
   1847 

~/.pyenv/versions/3.10-dev/lib/python3.10/selectors.py in select(self, timeout)
    467             ready = []
    468             try:
--> 469                 fd_event_list = self._selector.poll(timeout, max_ev)
    470             except InterruptedError:
    471                 return ready

KeyboardInterrupt: 
```

If you use [`asyncio.to_thread()`](https://docs.python.org/3/library/asyncio-task.html#asyncio.to_thread), then awaiting coroutines in succession works without a problem:
```python3
In [7]: from asyncio import to_thread

In [14]: await to_thread(sync_func, 0.5)
Slept 0.5 seconds.

In [15]: await to_thread(sync_func, 0.5)
Slept 0.5 seconds.
```

Using top-level await expression when [launching Python via `python3 -m asyncio`](https://piccolo-orm.com/blog/top-level-await-in-python/) works, though.

It also happens on Python 3.9.



Discussion:

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 5fa0671..74435e1 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -5,6 +5,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- Fixed ``to_thread.run_sync()`` hanging on the second call on asyncio when used with
+  ``loop.run_until_complete()``
 - Fixed the type annotation of ``open_signal_receiver()`` as a synchronous context manager
 - Fixed the type annotation of ``DeprecatedAwaitable(|List|Float).__await__`` to match the ``typing.Awaitable`` protocol
 
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index d6c9b47..b86e568 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -136,21 +136,18 @@ _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 getattr(cb, '__module__', None) == 'uvloop.loop'):
-                        _root_task.set(task)
-                        return task
-
-        _root_task.set(None)
-    else:
-        if root_task is not None:
-            return root_task
+    root_task = _root_task.get(None)
+    if root_task is not None and not root_task.done():
+        return root_task
+
+    # Look for a task that has been started via run_until_complete()
+    for task in all_tasks():
+        if task._callbacks and not task.done():
+            for cb in _get_task_callbacks(task):
+                if (cb is _run_until_complete_cb
+                        or getattr(cb, '__module__', None) == 'uvloop.loop'):
+                    _root_task.set(task)
+                    return task
 
     # Look up the topmost task in the AnyIO task tree, if possible
     task = cast(asyncio.Task, current_task())
@@ -791,6 +788,7 @@ async def run_sync_in_worker_thread(
             try:
                 return await future
             finally:
+                assert worker.is_alive()
                 worker.idle_since = current_time()
                 idle_workers.append(worker)
 


Test patch:
diff --git a/tests/test_to_thread.py b/tests/test_to_thread.py
index 7530d8a..5880ab4 100644
--- a/tests/test_to_thread.py
+++ b/tests/test_to_thread.py
@@ -183,3 +183,15 @@ def test_asyncio_run_sync_no_asyncio_run(asyncio_event_loop: asyncio.AbstractEve
     asyncio_event_loop.set_exception_handler(exception_handler)
     asyncio_event_loop.run_until_complete(to_thread.run_sync(time.sleep, 0))
     assert not exceptions
+
+
+def test_asyncio_run_sync_multiple(asyncio_event_loop: asyncio.AbstractEventLoop) -> None:
+    """Regression test for #304."""
+    asyncio_event_loop.call_later(0.5, asyncio_event_loop.stop)
+    for _ in range(3):
+        asyncio_event_loop.run_until_complete(to_thread.run_sync(time.sleep, 0))
+
+    for t in threading.enumerate():
+        if t.name == 'AnyIO worker thread':
+            t.join(2)
+            assert not t.is_alive()