Fixed from_thread.run(_sync)? not setting sniffio on asyncio (agron…

…holm#524)

Co-authored-by: Alex Grönholm <[email protected]>

docs/versionhistory.rst

@@ -50,6 +50,15 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   ``TLSStream.wrap()`` being inadvertently set on Python 3.11.3 and 3.10.11
 - Fixed ``CancelScope`` to properly handle asyncio task uncancellation on Python 3.11
   (PR by Nikolay Bryskin)
+- Fixed ``from_thread.run`` and ``from_thread.run_sync`` not setting sniffio on asyncio.
+  As a result:
+
+  - Fixed ``from_thread.run_sync`` failing when used to call sniffio-dependent functions
+    on asyncio
+  - Fixed ``from_thread.run`` failing when used to call sniffio-dependent functions on
+    asyncio from a thread running trio or curio
+  - Fixed deadlock when using ``from_thread.start_blocking_portal(backend="asyncio")``
+    in a thread running trio or curio (PR by Ganden Schaffner)
 
 **3.6.1**
 

src/anyio/_backends/_asyncio.py

@@ -2050,8 +2050,10 @@ def run_async_from_thread(
         token: object,
     ) -> T_Retval:
         loop = cast(AbstractEventLoop, token)
-        f: concurrent.futures.Future[T_Retval] = asyncio.run_coroutine_threadsafe(
-            func(*args), loop
+        context = copy_context()
+        context.run(sniffio.current_async_library_cvar.set, "asyncio")
+        f: concurrent.futures.Future[T_Retval] = context.run(
+            asyncio.run_coroutine_threadsafe, func(*args), loop
         )
         return f.result()
 
@@ -2062,6 +2064,7 @@ def run_sync_from_thread(
         @wraps(func)
         def wrapper() -> None:
             try:
+                sniffio.current_async_library_cvar.set("asyncio")
                 f.set_result(func(*args))
             except BaseException as exc:
                 f.set_exception(exc)

tests/test_from_thread.py

@@ -10,12 +10,14 @@
 from typing import Any, AsyncGenerator, NoReturn, TypeVar
 
 import pytest
+import sniffio
 from _pytest.logging import LogCaptureFixture
 
 from anyio import (
     Event,
     create_task_group,
     from_thread,
+    get_all_backends,
     get_cancelled_exc_class,
     get_current_task,
     run,
@@ -145,6 +147,16 @@ def worker() -> int:
 
         assert await to_thread.run_sync(worker) == 6
 
+    async def test_sniffio(self, anyio_backend_name: str) -> None:
+        async def async_func() -> str:
+            return sniffio.current_async_library()
+
+        def worker() -> str:
+            sniffio.current_async_library_cvar.set("something invalid for async_func")
+            return from_thread.run(async_func)
+
+        assert await to_thread.run_sync(worker) == anyio_backend_name
+
 
 class TestRunSyncFromThread:
     def test_run_sync_from_unclaimed_thread(self) -> None:
@@ -163,6 +175,13 @@ def worker() -> int:
 
         assert await to_thread.run_sync(worker) == 6
 
+    async def test_sniffio(self, anyio_backend_name: str) -> None:
+        def worker() -> str:
+            sniffio.current_async_library_cvar.set("something invalid for async_func")
+            return from_thread.run_sync(sniffio.current_async_library)
+
+        assert await to_thread.run_sync(worker) == anyio_backend_name
+
 
 class TestBlockingPortal:
     class AsyncCM:
@@ -524,3 +543,21 @@ async def raise_baseexception() -> None:
                     portal.call(raise_baseexception)
 
                 assert exc.value.__context__ is None
+
+    @pytest.mark.parametrize("portal_backend_name", get_all_backends())
+    async def test_from_async(
+        self, anyio_backend_name: str, portal_backend_name: str
+    ) -> None:
+        """
+        Test that portals don't deadlock when started/used from async code.
+
+        Note: This test will deadlock if there is a regression. A deadlock should be
+        treated as a failure. See also
+        https://github.com/agronholm/anyio/pull/524#discussion_r1183080886.
+
+        """
+        if anyio_backend_name == "trio" and portal_backend_name == "trio":
+            pytest.xfail("known bug (#525)")
+
+        with start_blocking_portal(portal_backend_name) as portal:
+            portal.call(checkpoint)