Repository: django/asgiref

Issue:
sync_to_async does not find root thread inside a task
When you use `create_task` (or any way of making a coroutine not through asgiref), and the root thread is synchronous, then `sync_to_async` in thread-sensitive mode will use the root thread outside of tasks and a single, specific new thread inside of tasks, ruining the thread-sensitive guarantee.

Instead, we should make `sync_to_async` always first look for an executor on the root thread and, if it finds it, use that to run code.

Discussion:

Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 8a05077..0379a31 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -6,7 +6,7 @@ repos:
         args: ["--py37-plus"]
 
   - repo: https://github.com/psf/black
-    rev: 20.8b1
+    rev: 22.3.0
     hooks:
       - id: black
         args: ["--target-version=py37"]
diff --git a/asgiref/sync.py b/asgiref/sync.py
index b71b379..a70dac1 100644
--- a/asgiref/sync.py
+++ b/asgiref/sync.py
@@ -1,3 +1,4 @@
+import asyncio
 import asyncio.coroutines
 import contextvars
 import functools
@@ -101,6 +102,10 @@ class AsyncToSync:
     # Local, not a threadlocal, so that tasks can work out what their parent used.
     executors = Local()
 
+    # When we can't find a CurrentThreadExecutor from the context, such as
+    # inside create_task, we'll look it up here from the running event loop.
+    loop_thread_executors: "Dict[asyncio.AbstractEventLoop, CurrentThreadExecutor]" = {}
+
     def __init__(self, awaitable, force_new_loop=False):
         if not callable(awaitable) or not _iscoroutinefunction_or_partial(awaitable):
             # Python does not have very reliable detection of async functions
@@ -164,6 +169,7 @@ class AsyncToSync:
             old_current_executor = None
         current_executor = CurrentThreadExecutor()
         self.executors.current = current_executor
+        loop = None
         # Use call_soon_threadsafe to schedule a synchronous callback on the
         # main event loop's thread if it's there, otherwise make a new loop
         # in this thread.
@@ -175,6 +181,7 @@ class AsyncToSync:
             if not (self.main_event_loop and self.main_event_loop.is_running()):
                 # Make our own event loop - in a new thread - and run inside that.
                 loop = asyncio.new_event_loop()
+                self.loop_thread_executors[loop] = current_executor
                 loop_executor = ThreadPoolExecutor(max_workers=1)
                 loop_future = loop_executor.submit(
                     self._run_event_loop, loop, awaitable
@@ -194,6 +201,8 @@ class AsyncToSync:
                     current_executor.run_until_future(call_result)
         finally:
             # Clean up any executor we were running
+            if loop is not None:
+                del self.loop_thread_executors[loop]
             if hasattr(self.executors, "current"):
                 del self.executors.current
             if old_current_executor:
@@ -378,6 +387,9 @@ class SyncToAsync:
                     # Create new thread executor in current context
                     executor = ThreadPoolExecutor(max_workers=1)
                     self.context_to_thread_executor[thread_sensitive_context] = executor
+            elif loop in AsyncToSync.loop_thread_executors:
+                # Re-use thread executor for running loop
+                executor = AsyncToSync.loop_thread_executors[loop]
             elif self.deadlock_context and self.deadlock_context.get(False):
                 raise RuntimeError(
                     "Single thread executor already being used, would deadlock"


Test patch:
diff --git a/tests/test_server.py b/tests/test_server.py
index 8f86096..616ccf2 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -21,7 +21,8 @@ class Server(StatelessServer):
             application,
             max_applications=max_applications,
         )
-        self._sock = sock.socket(sock.AF_INET, sock.SOCK_DGRAM | sock.SOCK_NONBLOCK)
+        self._sock = sock.socket(sock.AF_INET, sock.SOCK_DGRAM)
+        self._sock.setblocking(False)
         self._sock.bind(("127.0.0.1", 0))
 
     @property
@@ -54,7 +55,8 @@ class Server(StatelessServer):
 
 class Client:
     def __init__(self, name):
-        self._sock = sock.socket(sock.AF_INET, sock.SOCK_DGRAM | sock.SOCK_NONBLOCK)
+        self._sock = sock.socket(sock.AF_INET, sock.SOCK_DGRAM)
+        self._sock.setblocking(False)
         self.name = name
 
     async def register(self, server_addr, name=None):
diff --git a/tests/test_sync.py b/tests/test_sync.py
index 8f563d9..2837423 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -397,15 +397,21 @@ def test_thread_sensitive_outside_sync():
     @async_to_sync
     async def middle():
         await inner()
+        await asyncio.create_task(inner_task())
 
-    # Inner sync function
+    # Inner sync functions
     @sync_to_async
     def inner():
         result["thread"] = threading.current_thread()
 
+    @sync_to_async
+    def inner_task():
+        result["thread2"] = threading.current_thread()
+
     # Run it
     middle()
     assert result["thread"] == threading.current_thread()
+    assert result["thread2"] == threading.current_thread()
 
 
 @pytest.mark.asyncio