Repository: django/asgiref
Pull request #367 (2023-01-11 15:31:10)

Allow nesting sync_to_async via asyncio.wait_for
Change the order of fallbacks used by SyncToAsync to find the appropriate executor for sync code, so it prefers to use AsyncToSync.loop_thread_executors rather than thread_sensitive_context. Add test case to demonstrate problem.

Resolves django/asgiref#348

carltongibson: Hi @jthorniley 

> Resolves https://github.com/django/asgiref/issues/348

I'm still looking at this. There's two issues it looks like: even with the patch here, @brownan's example  from #348 still hits a deadlock. 
carltongibson: The test-case doesn't quite match the behaviour of the example running under Daphne. 

In the test the executors selected in SyncToAsync.__call__ are: 

```
<concurrent.futures.thread.ThreadPoolExecutor object at 0x1064a4730> # in middleware
<asgiref.current_thread_executor.CurrentThreadExecutor object at 0x1064a0850> # in async view
```

Under Daphne we get the same `concurrent.futures.thread.ThreadPoolExecutor` selected twice (which causes the timeout/deadlock, depending on which option is selected.) 
jthorniley: Hi @carltongibson thanks for your comments, sorry I haven't had time to look at this for a while.

I've tried an alternative approach which I think makes more sense. Instead of changing how we look up the executor, set the reference to the current thread executor using the contextvars lib - since Python 3.7 is the minimum supported version this seems like its probably an option when it wasn't before.

The asgiref local has to be maintained across sync_to_async/async_to_sync conversions using the `launch_map` lookups, but these lookup are based on asyncio Task ID, which changes when you use asyncio.gather and friends, whereas contextvars is generally "just works" in asyncio code. Also, we already make the effort to transfer context to the new thread in AsyncToSync so there isn't really much extra logic to take care of.

I've tried this implementation with the example ASGI code in issue #348 - and it seems to work (GET request completes and the page renders "Hello world"). I've tried daphne and uvicorn and they both seem ok with this.

One problem is that one of the deadlock unit tests now fails, I think because there isn't a deadlock anymore! (It just actually works). I think its probably safe to just remove the test?


andrewgodwin: The intention was always to implement Local via contextvars once we hit the minimum version to support it, so I wouldn't be against doing that.
jthorniley: @andrewgodwin 
> The intention was always to implement Local via contextvars once we hit the minimum version to support it, so I wouldn't be against doing that.

That makes sense, contextvars is python 3.7 which seems to be now the min supported version, I did try to actually do that first and I have an implementation that seems to work, but I wasnt too happy with the way it handles deleting keys, see here: 

https://github.com/django/asgiref/compare/main...jthorniley:asgiref:local-contextvar-draft?expand=1

(Comments in the diff, I won't repeat but tldr I'm worried I've made a memory leak) 


carltongibson: Hi @jthorniley — thanks for swinging back to this. 

For [life reasons](https://noumenal.es/posts/2023-Q1-status-update/mbB/), it's not likely I'll have capacity to look at this closely soon, so let me just comment quickly not to block: 

> One problem is that one of the deadlock unit tests now fails, I think because there isn't a deadlock anymore! (It just actually works). I think it's probably safe to just remove the test?

Sounds right 😅

>> The intention was always to implement Local via contextvars once we hit the minimum version to support it, so I wouldn't be against doing that.
>
>That makes sense, contextvars is python 3.7 which seems to be now the min supported version...

This would be great IMO! I briefly glanced at this back last year, but didn't get to progress it. If you could push it through, super! 🎁

There's even a comment about this: 

https://github.com/django/asgiref/blob/03571587bf0f61ddff77f1f6fe3593439cb92f48/asgiref/local.py#L29-L30
jthorniley: Thanks both for the comments. I've updated the PR to use contextvars as an implementation for Local, and reverted the logic that keeps track of the CurrentThreadExecutor to use that new Local implementation rather than a plain contextvar, as I found in testing that plain contextvars are not thread safe when you are passing them between threads as is the case here.

Anyway, I think this implementation is fairly solid from what testing I've managed to do.
andrewgodwin: Talking of tests... those and the lint failures _will_ need fixing first!
jthorniley: The test failure was a good catch - the state was not cleaned up properly so it only occurred on the full test run rather than running that test individually.

Lint issue also fixed.

Whether this is ready to merge... as you say its hard to assess this sort of thing beyond the tests. I'm tempted to wait until I've rolled out this branch at work on Monday just to give some extra confidence from a "real-world" test (I can't see what would go wrong, and its passed all our CI). I will update then.
andrewgodwin: Alright, sounds like a good plan!
andrewgodwin: @jthorniley Did you manage to get any testing of this in this week?
jthorniley: @andrewgodwin yes and no I'm afraid... We rolled it out but had to roll back due to (unrelated to this change) infrastructure issues and haven't managed to redo this. So yes, this still looks fine still as far as I can see, but didn't get a proper test really.
andrewgodwin: Ah, no problem - let's wait a bit longer to see if you can give it some more burn-in time. Hopefully the other issues are easy to fix!
jthorniley: Ok, we've tried rolling out again and actually I think there are some issues with this branch so unfortunately it wouldn't be good to merge now. 

Using this change (with a Django/ASGI app) we end up with too many database connections getting created, I think because they are held in a local by Django and the behaviour has changed in some unfortunate way (hard to track down exactly).

Additionally (trying to track down something more concrete) I found that running the current Django tests with this asgiref branch results in a failure:

```
./runtests.py asgi
Testing against Django installed in '/Users/james/repos/django/.venv/lib/python3.10/site-packages/django' with up to 8 processes
Found 16 test(s).
System check identified no issues (0 silenced).
..F.............
======================================================================
FAIL: test_concurrent_async_uses_multiple_thread_pools (asgi.tests.ASGITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/james/repos/asgiref/asgiref/sync.py", line 235, in __call__
    return call_result.result()
  File "/Users/james/.pyenv/versions/3.10.7/lib/python3.10/concurrent/futures/_base.py", line 451, in result
    return self.__get_result()
  File "/Users/james/.pyenv/versions/3.10.7/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/Users/james/repos/asgiref/asgiref/sync.py", line 297, in main_wrap
    result = await self.awaitable(*args, **kwargs)
  File "/Users/james/repos/django/tests/asgi/tests.py", line 339, in test_concurrent_async_uses_multiple_thread_pools
    self.assertEqual(response_start["status"], 200)
AssertionError: 500 != 200

----------------------------------------------------------------------
Ran 16 tests in 6.927s

FAILED (failures=1)
```

I still think this approach (changing to `contextvars` for the implementation of `Local`) makes sense, but these issues need to be resolved.
andrewgodwin: Hmm, well that's annoying, but not entirely unexpected. I can only presume that the behaviour around synchronous contexts is different enough to make it break - contextvars inherit in a different way to what Local currently does.
jthorniley: I think this is now a viable solution.

I'll just note that there's a separate issue #399 which is also directly affected by this (i.e. this essentially does that).

So this went stale as I got a bit stuck but I've had time to work on it, so I've rebased and addressed a few issues:

## Local(thread_critical=True)

The behaviour of Local(thread_critical=True) has a critical subtlety which I've added a test for in `test_local` `test_thread_critical_local_not_context_dependent_in_sync_thread()`

Essentially, when `Local(thread_critical=True)` behaves more or less identically to `threading.local()` i.e. "global within the current thread", rather than simulating the contextvar behaviour of being local to a specific call stack. This is important because django uses a `Local(thread_critical=True)` to store the database connection reference at the point at which it is created. 

In an Django ASGIHandler server, there will be one sync thread per request, and it will get a DB connection initialised local to that thread (typically) somewhere inside the async view code, wrapped in a `sync_to_async`. However, the connection is only closed using a callback attached to the `request_finished` signal. As a result, its important that the callbacks from the `request_finished` signal see the same `Local()` data as the inner `sync_to_async` code (provided its running in the same thread), otherwise DB connections never get closed.

Its easy to confirm that `test_thread_critical_local_not_context_dependent_in_sync_thread()` is just testing the existing behaviour of `Local` - i.e. it passes if copied onto current main branch.

## ApplicationCommunicator

The change in behaviour of `Local` means that a `Local` is now preserved even when using `asyncio.create_task`- in the case of `ApplicationCommunicator` it was actually a benefit that it wasn't - this was the cause of the django test failure that I had above (now resolved).

When the django test code run an `async def test_x` function, it wraps it in `async_to_sync` and as a result any `sync_to_async` wrapped code inside that function will run in the original sync test thread.

However that means when using `ApplicationCommunicator` any `sync_to_async` wrapped code inside the ASGI handler will end up running in the single top-level test thread, which is really not what we want - `ApplicationCommunicator` should simulate the behaviour if you'd started the ASGI server straight from a "normal" async thread - i.e. the first `sync_to_async` inside an ASGI request should create its own Thread to run in.

The solution was to create a new `contextvars.Context()` inside `ApplicationCommunicator` so that the ASGI server doesn't try to run code in the top-level test thread.

andrewgodwin: I really appreciate your work on this!

Looks like the tests are failing in CI, at least - different environment issue to you locally, or something else, you think?
jthorniley: @andrewgodwin  thanks for running it - yes just taking a quick look:

I think CI is running mypy which I didn't run locally, but I'm sure I can tidy that up.

Additionally used py3.11 locally - and the tests do work there in CI, i'm guessing some API changes between 3.10 - 3.11 that I didn't account for, I'll investigate that today.
andrewgodwin: Thanks - looks great now. I'll merge it in, and hopefully when we make a release with this in we won't get any immediate screaming! (The test suite looks good and pretty decently covers this, but there's always the chance of uncaught side effects)

Patch:
diff --git a/asgiref/local.py b/asgiref/local.py
index 17314d4..a8b9459 100644
--- a/asgiref/local.py
+++ b/asgiref/local.py
@@ -1,120 +1,128 @@
-import random
-import string
-import sys
+import asyncio
+import contextlib
+import contextvars
 import threading
-import weakref
+from typing import Any, Dict, Union
+
+
+class _CVar:
+    """Storage utility for Local."""
+
+    def __init__(self) -> None:
+        self._data: "contextvars.ContextVar[Dict[str, Any]]" = contextvars.ContextVar(
+            "asgiref.local"
+        )
+
+    def __getattr__(self, key):
+        storage_object = self._data.get({})
+        try:
+            return storage_object[key]
+        except KeyError:
+            raise AttributeError(f"{self!r} object has no attribute {key!r}")
+
+    def __setattr__(self, key: str, value: Any) -> None:
+        if key == "_data":
+            return super().__setattr__(key, value)
+
+        storage_object = self._data.get({})
+        storage_object[key] = value
+        self._data.set(storage_object)
+
+    def __delattr__(self, key: str) -> None:
+        storage_object = self._data.get({})
+        if key in storage_object:
+            del storage_object[key]
+            self._data.set(storage_object)
+        else:
+            raise AttributeError(f"{self!r} object has no attribute {key!r}")
 
 
 class Local:
-    """
-    A drop-in replacement for threading.locals that also works with asyncio
-    Tasks (via the current_task asyncio method), and passes locals through
-    sync_to_async and async_to_sync.
-
-    Specifically:
-     - Locals work per-coroutine on any thread not spawned using asgiref
-     - Locals work per-thread on any thread not spawned using asgiref
-     - Locals are shared with the parent coroutine when using sync_to_async
-     - Locals are shared with the parent thread when using async_to_sync
-       (and if that thread was launched using sync_to_async, with its parent
-       coroutine as well, with this working for indefinite levels of nesting)
-
-    Set thread_critical to True to not allow locals to pass from an async Task
-    to a thread it spawns. This is needed for code that truly needs
-    thread-safety, as opposed to things used for helpful context (e.g. sqlite
-    does not like being called from a different thread to the one it is from).
-    Thread-critical code will still be differentiated per-Task within a thread
-    as it is expected it does not like concurrent access.
-
-    This doesn't use contextvars as it needs to support 3.6. Once it can support
-    3.7 only, we can then reimplement the storage more nicely.
+    """Local storage for async tasks.
+
+    This is a namespace object (similar to `threading.local`) where data is
+    also local to the current async task (if there is one).
+
+    In async threads, local means in the same sense as the `contextvars`
+    module - i.e. a value set in an async frame will be visible:
+
+    - to other async code `await`-ed from this frame.
+    - to tasks spawned using `asyncio` utilities (`create_task`, `wait_for`,
+      `gather` and probably others).
+    - to code scheduled in a sync thread using `sync_to_async`
+
+    In "sync" threads (a thread with no async event loop running), the
+    data is thread-local, but additionally shared with async code executed
+    via the `async_to_sync` utility, which schedules async code in a new thread
+    and copies context across to that thread.
+
+    If `thread_critical` is True, then the local will only be visible per-thread,
+    behaving exactly like `threading.local` if the thread is sync, and as
+    `contextvars` if the thread is async. This allows genuinely thread-sensitive
+    code (such as DB handles) to be kept stricly to their initial thread and
+    disable the sharing across `sync_to_async` and `async_to_sync` wrapped calls.
+
+    Unlike plain `contextvars` objects, this utility is threadsafe.
     """
 
     def __init__(self, thread_critical: bool = False) -> None:
         self._thread_critical = thread_critical
         self._thread_lock = threading.RLock()
-        self._context_refs: "weakref.WeakSet[object]" = weakref.WeakSet()
-        # Random suffixes stop accidental reuse between different Locals,
-        # though we try to force deletion as well.
-        self._attr_name = "_asgiref_local_impl_{}_{}".format(
-            id(self),
-            "".join(random.choice(string.ascii_letters) for i in range(8)),
-        )
 
-    def _get_context_id(self):
-        """
-        Get the ID we should use for looking up variables
-        """
-        # Prevent a circular reference
-        from .sync import AsyncToSync, SyncToAsync
-
-        # First, pull the current task if we can
-        context_id = SyncToAsync.get_current_task()
-        context_is_async = True
-        # OK, let's try for a thread ID
-        if context_id is None:
-            context_id = threading.current_thread()
-            context_is_async = False
-        # If we're thread-critical, we stop here, as we can't share contexts.
+        self._storage: "Union[threading.local, _CVar]"
+
+        if thread_critical:
+            # Thread-local storage
+            self._storage = threading.local()
+        else:
+            # Contextvar storage
+            self._storage = _CVar()
+
+    @contextlib.contextmanager
+    def _lock_storage(self):
+        # Thread safe access to storage
         if self._thread_critical:
-            return context_id
-        # Now, take those and see if we can resolve them through the launch maps
-        for i in range(sys.getrecursionlimit()):
             try:
-                if context_is_async:
-                    # Tasks have a source thread in AsyncToSync
-                    context_id = AsyncToSync.launch_map[context_id]
-                    context_is_async = False
-                else:
-                    # Threads have a source task in SyncToAsync
-                    context_id = SyncToAsync.launch_map[context_id]
-                    context_is_async = True
-            except KeyError:
-                break
+                # this is a test for are we in a async or sync
+                # thread - will raise RuntimeError if there is
+                # no current loop
+                asyncio.get_running_loop()
+            except RuntimeError:
+                # We are in a sync thread, the storage is
+                # just the plain thread local (i.e, "global within
+                # this thread" - it doesn't matter where you are
+                # in a call stack you see the same storage)
+                yield self._storage
+            else:
+                # We are in an async thread - storage is still
+                # local to this thread, but additionally should
+                # behave like a context var (is only visible with
+                # the same async call stack)
+
+                # Ensure context exists in the current thread
+                if not hasattr(self._storage, "cvar"):
+                    self._storage.cvar = _CVar()
+
+                # self._storage is a thread local, so the members
+                # can't be accessed in another thread (we don't
+                # need any locks)
+                yield self._storage.cvar
         else:
-            # Catch infinite loops (they happen if you are screwing around
-            # with AsyncToSync implementations)
-            raise RuntimeError("Infinite launch_map loops")
-        return context_id
-
-    def _get_storage(self):
-        context_obj = self._get_context_id()
-        if not hasattr(context_obj, self._attr_name):
-            setattr(context_obj, self._attr_name, {})
-            self._context_refs.add(context_obj)
-        return getattr(context_obj, self._attr_name)
-
-    def __del__(self):
-        try:
-            for context_obj in self._context_refs:
-                try:
-                    delattr(context_obj, self._attr_name)
-                except AttributeError:
-                    pass
-        except TypeError:
-            # WeakSet.__iter__ can crash when interpreter is shutting down due
-            # to _IterationGuard being None.
-            pass
+            # Lock for thread_critical=False as other threads
+            # can access the exact same storage object
+            with self._thread_lock:
+                yield self._storage
 
     def __getattr__(self, key):
-        with self._thread_lock:
-            storage = self._get_storage()
-            if key in storage:
-                return storage[key]
-            else:
-                raise AttributeError(f"{self!r} object has no attribute {key!r}")
+        with self._lock_storage() as storage:
+            return getattr(storage, key)
 
     def __setattr__(self, key, value):
-        if key in ("_context_refs", "_thread_critical", "_thread_lock", "_attr_name"):
+        if key in ("_local", "_storage", "_thread_critical", "_thread_lock"):
             return super().__setattr__(key, value)
-        with self._thread_lock:
-            storage = self._get_storage()
-            storage[key] = value
+        with self._lock_storage() as storage:
+            setattr(storage, key, value)
 
     def __delattr__(self, key):
-        with self._thread_lock:
-            storage = self._get_storage()
-            if key in storage:
-                del storage[key]
-            else:
-                raise AttributeError(f"{self!r} object has no attribute {key!r}")
+        with self._lock_storage() as storage:
+            delattr(storage, key)
diff --git a/asgiref/sync.py b/asgiref/sync.py
index 4d93c6b..2a819a5 100644
--- a/asgiref/sync.py
+++ b/asgiref/sync.py
@@ -140,12 +140,9 @@ class AsyncToSync(Generic[_P, _R]):
     finally exiting once the async task returns.
     """
 
-    # Maps launched Tasks to the threads that launched them (for locals impl)
-    launch_map: "Dict[asyncio.Task[object], threading.Thread]" = {}
-
-    # Keeps track of which CurrentThreadExecutor to use. This uses an asgiref
-    # Local, not a threadlocal, so that tasks can work out what their parent used.
-    executors = Local()
+    # Keeps a reference to the CurrentThreadExecutor in local context, so that
+    # any sync_to_async inside the wrapped code can find it.
+    executors: "Local" = 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.
@@ -211,23 +208,20 @@ class AsyncToSync(Generic[_P, _R]):
                     "just await the async function directly."
                 )
 
-        # Wrapping context in list so it can be reassigned from within
-        # `main_wrap`.
-        context = [contextvars.copy_context()]
-
         # Make a future for the return information
         call_result: "Future[_R]" = Future()
-        # Get the source thread
-        source_thread = threading.current_thread()
+
         # Make a CurrentThreadExecutor we'll use to idle in this thread - we
         # need one for every sync frame, even if there's one above us in the
         # same thread.
-        if hasattr(self.executors, "current"):
-            old_current_executor = self.executors.current
-        else:
-            old_current_executor = None
+        old_executor = getattr(self.executors, "current", None)
         current_executor = CurrentThreadExecutor()
         self.executors.current = current_executor
+
+        # Wrapping context in list so it can be reassigned from within
+        # `main_wrap`.
+        context = [contextvars.copy_context()]
+
         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
@@ -235,7 +229,6 @@ class AsyncToSync(Generic[_P, _R]):
         try:
             awaitable = self.main_wrap(
                 call_result,
-                source_thread,
                 sys.exc_info(),
                 context,
                 *args,
@@ -267,11 +260,9 @@ class AsyncToSync(Generic[_P, _R]):
             # 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:
-                self.executors.current = old_current_executor
             _restore_context(context[0])
+            # Restore old current thread executor state
+            self.executors.current = old_executor
 
         # Wait for results from the future.
         return call_result.result()
@@ -322,7 +313,6 @@ class AsyncToSync(Generic[_P, _R]):
     async def main_wrap(
         self,
         call_result: "Future[_R]",
-        source_thread: threading.Thread,
         exc_info: "OptExcInfo",
         context: List[contextvars.Context],
         *args: _P.args,
@@ -338,9 +328,6 @@ class AsyncToSync(Generic[_P, _R]):
         if context is not None:
             _restore_context(context[0])
 
-        current_task = SyncToAsync.get_current_task()
-        assert current_task is not None
-        self.launch_map[current_task] = source_thread
         try:
             # If we have an exception, run the function inside the except block
             # after raising it so exc_info is correctly populated.
@@ -356,8 +343,6 @@ class AsyncToSync(Generic[_P, _R]):
         else:
             call_result.set_result(result)
         finally:
-            del self.launch_map[current_task]
-
             context[0] = contextvars.copy_context()
 
 
@@ -383,9 +368,6 @@ class SyncToAsync(Generic[_P, _R]):
     a TypeError will be raised.
     """
 
-    # Maps launched threads to the coroutines that spawned them
-    launch_map: "Dict[threading.Thread, asyncio.Task[object]]" = {}
-
     # Storage for main event loop references
     threadlocal = threading.local()
 
@@ -440,9 +422,10 @@ class SyncToAsync(Generic[_P, _R]):
 
         # Work out what thread to run the code in
         if self._thread_sensitive:
-            if hasattr(AsyncToSync.executors, "current"):
+            current_thread_executor = getattr(AsyncToSync.executors, "current", None)
+            if current_thread_executor:
                 # If we have a parent sync thread above somewhere, use that
-                executor = AsyncToSync.executors.current
+                executor = current_thread_executor
             elif self.thread_sensitive_context.get(None):
                 # If we have a way of retrieving the current context, attempt
                 # to use a per-context thread pool executor
@@ -481,7 +464,6 @@ class SyncToAsync(Generic[_P, _R]):
                 functools.partial(
                     self.thread_handler,
                     loop,
-                    self.get_current_task(),
                     sys.exc_info(),
                     func,
                     child,
@@ -503,7 +485,7 @@ class SyncToAsync(Generic[_P, _R]):
         func = functools.partial(self.__call__, parent)
         return functools.update_wrapper(func, self.func)
 
-    def thread_handler(self, loop, source_task, exc_info, func, *args, **kwargs):
+    def thread_handler(self, loop, exc_info, func, *args, **kwargs):
         """
         Wraps the sync application with exception handling.
         """
@@ -513,45 +495,17 @@ class SyncToAsync(Generic[_P, _R]):
         # Set the threadlocal for AsyncToSync
         self.threadlocal.main_event_loop = loop
         self.threadlocal.main_event_loop_pid = os.getpid()
-        # Set the task mapping (used for the locals module)
-        current_thread = threading.current_thread()
-        if AsyncToSync.launch_map.get(source_task) == current_thread:
-            # Our parent task was launched from this same thread, so don't make
-            # a launch map entry - let it shortcut over us! (and stop infinite loops)
-            parent_set = False
-        else:
-            self.launch_map[current_thread] = source_task
-            parent_set = True
-        source_task = (
-            None  # allow the task to be garbage-collected in case of exceptions
-        )
+
         # Run the function
-        try:
-            # If we have an exception, run the function inside the except block
-            # after raising it so exc_info is correctly populated.
-            if exc_info[1]:
-                try:
-                    raise exc_info[1]
-                except BaseException:
-                    return func(*args, **kwargs)
-            else:
+        # If we have an exception, run the function inside the except block
+        # after raising it so exc_info is correctly populated.
+        if exc_info[1]:
+            try:
+                raise exc_info[1]
+            except BaseException:
                 return func(*args, **kwargs)
-        finally:
-            # Only delete the launch_map parent if we set it, otherwise it is
-            # from someone else.
-            if parent_set:
-                del self.launch_map[current_thread]
-
-    @staticmethod
-    def get_current_task() -> Optional["asyncio.Task[Any]"]:
-        """
-        Implementation of asyncio.current_task()
-        that returns None if there is no task.
-        """
-        try:
-            return asyncio.current_task()
-        except RuntimeError:
-            return None
+        else:
+            return func(*args, **kwargs)
 
 
 @overload


Test patch:
diff --git a/asgiref/testing.py b/asgiref/testing.py
index 6624317..aa7cff1 100644
--- a/asgiref/testing.py
+++ b/asgiref/testing.py
@@ -1,4 +1,5 @@
 import asyncio
+import contextvars
 import time
 
 from .compatibility import guarantee_single_callable
@@ -16,8 +17,13 @@ class ApplicationCommunicator:
         self.scope = scope
         self.input_queue = asyncio.Queue()
         self.output_queue = asyncio.Queue()
-        self.future = asyncio.ensure_future(
-            self.application(scope, self.input_queue.get, self.output_queue.put)
+        # Clear context - this ensures that context vars set in the testing scope
+        # are not "leaked" into the application which would normally begin with
+        # an empty context. In Python >= 3.11 this could also be written as:
+        # asyncio.create_task(..., context=contextvars.Context())
+        self.future = contextvars.Context().run(
+            asyncio.create_task,
+            self.application(scope, self.input_queue.get, self.output_queue.put),
         )
 
     async def wait(self, timeout=1):
diff --git a/tests/test_local.py b/tests/test_local.py
index 035c99f..d50cba2 100644
--- a/tests/test_local.py
+++ b/tests/test_local.py
@@ -1,3 +1,4 @@
+import asyncio
 import gc
 import threading
 
@@ -307,18 +308,33 @@ async def test_local_threads_and_tasks():
     assert test_local.counter == 6
 
 
-def test_local_del_swallows_type_error(monkeypatch):
-    test_local = Local()
-
-    blow_up_calls = 0
-
-    def blow_up(self):
-        nonlocal blow_up_calls
-        blow_up_calls += 1
-        raise TypeError()
-
-    monkeypatch.setattr("weakref.WeakSet.__iter__", blow_up)
+def test_thread_critical_local_not_context_dependent_in_sync_thread():
+    # Test function is sync, thread critical local should
+    # be visible everywhere in the sync thread, even if set
+    # from inside a sync_to_async/async_to_sync stack (so
+    # long as it was set in sync code)
+    test_local_tc = Local(thread_critical=True)
+    test_local_not_tc = Local(thread_critical=False)
+    test_thread = threading.current_thread()
+
+    @sync_to_async
+    def inner_sync_function():
+        # sync_to_async should run this code inside the original
+        # sync thread, confirm this here
+        assert test_thread == threading.current_thread()
+        test_local_tc.test_value = "_123_"
+        test_local_not_tc.test_value = "_456_"
+
+    @async_to_sync
+    async def async_function():
+        await asyncio.create_task(inner_sync_function())
 
-    test_local.__del__()
+    async_function()
 
-    assert blow_up_calls == 1
+    # assert: the inner_sync_function should have set a value
+    # visible here
+    assert test_local_tc.test_value == "_123_"
+    # however, if the local was non-thread-critical, then the
+    # inner value was set inside a new async context, meaning that
+    # we do not see it, as context vars don't propagate up the stack
+    assert not hasattr(test_local_not_tc, "test_value")
diff --git a/tests/test_sync.py b/tests/test_sync.py
index c46e9be..7391c08 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -721,29 +721,6 @@ async def test_sync_to_async_uses_executor():
     )
 
 
-def test_sync_to_async_deadlock_raises():
-    def db_write():
-        pass
-
-    async def io_task():
-        await sync_to_async(db_write)()
-
-    async def do_io_tasks():
-        t = asyncio.create_task(io_task())
-        await t
-        # await asyncio.gather(io_task()) # Also deadlocks
-        # await io_task() # Works
-
-    def view():
-        async_to_sync(do_io_tasks)()
-
-    async def server_entry():
-        await sync_to_async(view)()
-
-    with pytest.raises(RuntimeError):
-        asyncio.run(server_entry())
-
-
 def test_sync_to_async_deadlock_ignored_with_exception():
     """
     Ensures that throwing an exception from inside a deadlock-protected block
@@ -836,3 +813,39 @@ async def test_sync_to_async_with_blocker_non_thread_sensitive():
         raise
     finally:
         await trigger_task
+
+
+@pytest.mark.asyncio
+async def test_sync_to_async_within_create_task():
+    """
+    Test a stack of sync_to_async/async_to_sync/sync_to_async works even when last
+    sync_to_async is wrapped in asyncio.wait_for.
+    """
+    main_thread = threading.current_thread()
+    sync_thread = None
+
+    # Hypothetical Django scenario - middleware function is sync and will run
+    # in a new thread created by sync_to_async
+    def sync_middleware():
+        nonlocal sync_thread
+        sync_thread = threading.current_thread()
+        assert sync_thread != main_thread
+        # View is async and wrapped with async_to_sync.
+        async_to_sync(async_view)()
+
+    async def async_view():
+        # Call a sync function using sync_to_async, but asyncio.wait_for it
+        # rather than directly await it.
+        await asyncio.wait_for(sync_to_async(sync_task)(), timeout=1)
+
+    task_executed = False
+
+    def sync_task():
+        nonlocal task_executed, sync_thread
+        assert sync_thread == threading.current_thread()
+        task_executed = True
+
+    async with ThreadSensitiveContext():
+        await sync_to_async(sync_middleware)()
+
+    assert task_executed