Repository: django/asgiref

Issue:
Proposal: introduce `http.response.pathsend` extension
Some ASGI servers might not be able to implement `http.response.zerocopysend` extension due to specific implementation constraints, for instance Rust implemented servers like [Granian](https://github.com/emmett-framework/granian) won't play well with file descriptors.

Nevertheless, such servers might still be able to directly serve files with "zero copy" techniques when the source path is known in advance. For instance, Granian already provide such feature within the [RSGI protocol](https://github.com/emmett-framework/granian/blob/master/docs/spec/RSGI.md#http-protocol-interface), offering ~28x performance increase over standard file serving in ASGI.

Given this dramatic performance gain, it would be nice to introduce a similar approach also in ASGI, applying the same *offloading* technique for such kind of contents, where the server itself is the sole responsible of processing the files. And this is the aim of this proposal: introduce an extension quite similar to `http.response.zerocopysend` but based on paths rather than file descriptiors.

The idea is to add a `pathsend` extension which the server can list in the appropriate key of the scope:

```python
"scope": {
    ...
    "extensions": {
        "http.response.pathsend": {},
    },
}
```

With the above extra definition, a new event can be produced by ASGI applications, with the following keys:

- `type` (Unicode string): "http.response.pathsend"
- `pathsend` (Unicode string): The string representation of the file path to serve (platform specific)

which will instruct the server to serve the file directly from the relevant path.

A couple of details worth to mention:
- in this design, the ASGI application is still responsible to send the `http.response.start` event with the relevant headers (eg: `content-type`, `content-length`, etc.)
- the path should be absolute, or when relative would be relative to the current working directory the server was started

Once we discussed this properly, and if it makes sense, I'm up to open a PR with the relevant changes.
Reference of the Granian issue: https://github.com/emmett-framework/granian/issues/82

Discussion:
Kludex: Not related: From this PR I noticed that Zero Copy Send related type is missing.

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
diff --git a/asgiref/typing.py b/asgiref/typing.py
index d77d1c9..46cb164 100644
--- a/asgiref/typing.py
+++ b/asgiref/typing.py
@@ -32,6 +32,7 @@ __all__ = (
     "HTTPResponseStartEvent",
     "HTTPResponseBodyEvent",
     "HTTPResponseTrailersEvent",
+    "HTTPResponsePathsendEvent",
     "HTTPServerPushEvent",
     "HTTPDisconnectEvent",
     "WebSocketConnectEvent",
@@ -138,6 +139,11 @@ class HTTPResponseTrailersEvent(TypedDict):
     more_trailers: bool
 
 
+class HTTPResponsePathsendEvent(TypedDict):
+    type: Literal["http.response.pathsend"]
+    path: str
+
+
 class HTTPServerPushEvent(TypedDict):
     type: Literal["http.response.push"]
     path: str
diff --git a/docs/extensions.rst b/docs/extensions.rst
index 1714750..8fe0416 100644
--- a/docs/extensions.rst
+++ b/docs/extensions.rst
@@ -129,6 +129,41 @@ After calling this extension to respond, the ASGI application itself should
 actively close the used file descriptor - ASGI servers are not responsible for
 closing descriptors.
 
+Path Send
+---------
+
+Path Send allows you to send the contents of a file path to the
+HTTP client without handling file descriptors, offloading the operation
+directly to the server.
+
+ASGI servers that implement this extension will provide
+``http.response.pathsend`` in the extensions part of the scope::
+
+    "scope": {
+        ...
+        "extensions": {
+            "http.response.pathsend": {},
+        },
+    }
+
+The ASGI framework can initiate a path-send by sending a message with
+the following keys. This message can be sent at any time after the
+*Response Start* message, and cannot be mixed with ``http.response.body``.
+It can be called just one time in one response.
+Except for the characteristics of path-send, it should behave the same
+as ordinary ``http.response.body``.
+
+Keys:
+
+* ``type`` (*Unicode string*): ``"http.response.pathsend"``
+
+* ``path`` (*Unicode string*): The string representation of the absolute
+file path to be sent by the server, platform specific.
+
+The ASGI application itself is responsible to send the relevant headers
+in the *Response Start* message, like the ``Content-Type`` and
+``Content-Length`` headers for the file to be sent.
+
 TLS
 ---
 


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