Fixed #32889 -- Allowed per-request sync_to_async context in ASGIHandler .

By using a asgiref's ThreadSensitiveContext context manager, requests
will be able to execute independently of other requests when sync work
is involved.

Prior to this commit, a single global thread was used to execute any
sync work independent of the request from which that work was scheduled.
This could result in contention for the global sync thread in the case
of a slow sync function.

Requests are now isolated to their own sync thread.

2 files changed, 49 insertions, 7 deletions

diff --git a/tests/asgi/tests.py b/tests/asgi/tests.py
index 3509bb0aa7..7eb35724df 100644
--- a/tests/asgi/tests.py
+++ b/tests/asgi/tests.py
@@ -4,7 +4,6 @@ import threading
 from pathlib import Path
 from unittest import skipIf
 
-from asgiref.sync import SyncToAsync
 from asgiref.testing import ApplicationCommunicator
 
 from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
@@ -16,7 +15,7 @@ from django.test import (
 )
 from django.utils.http import http_date
 
-from .urls import test_filename
+from .urls import sync_waiter, test_filename
 
 TEST_STATIC_ROOT = Path(__file__).parent / 'project' / 'static'
 
@@ -235,11 +234,39 @@ class ASGITest(SimpleTestCase):
         # Give response.close() time to finish.
         await communicator.wait()
 
-        # At this point, AsyncToSync does not have a current executor. Thus
-        # SyncToAsync falls-back to .single_thread_executor.
-        target_thread = next(iter(SyncToAsync.single_thread_executor._threads))
+        # AsyncToSync should have executed the signals in the same thread.
         request_started_thread, request_finished_thread = signal_handler.threads
-        self.assertEqual(request_started_thread, target_thread)
-        self.assertEqual(request_finished_thread, target_thread)
+        self.assertEqual(request_started_thread, request_finished_thread)
         request_started.disconnect(signal_handler)
         request_finished.disconnect(signal_handler)
+
+    async def test_concurrent_async_uses_multiple_thread_pools(self):
+        sync_waiter.active_threads.clear()
+
+        # Send 2 requests concurrently
+        application = get_asgi_application()
+        scope = self.async_request_factory._base_scope(path='/wait/')
+        communicators = []
+        for _ in range(2):
+            communicators.append(ApplicationCommunicator(application, scope))
+            await communicators[-1].send_input({'type': 'http.request'})
+
+        # Each request must complete with a status code of 200
+        # If requests aren't scheduled concurrently, the barrier in the
+        # sync_wait view will time out, resulting in a 500 status code.
+        for communicator in communicators:
+            response_start = await communicator.receive_output()
+            self.assertEqual(response_start['type'], 'http.response.start')
+            self.assertEqual(response_start['status'], 200)
+            response_body = await communicator.receive_output()
+            self.assertEqual(response_body['type'], 'http.response.body')
+            self.assertEqual(response_body['body'], b'Hello World!')
+            # Give response.close() time to finish.
+            await communicator.wait()
+
+        # The requests should have scheduled on different threads. Note
+        # active_threads is a set (a thread can only appear once), therefore
+        # length is a sufficient check.
+        self.assertEqual(len(sync_waiter.active_threads), 2)
+
+        sync_waiter.active_threads.clear()
diff --git a/tests/asgi/urls.py b/tests/asgi/urls.py
index ff8d21ea7c..22d85604d1 100644
--- a/tests/asgi/urls.py
+++ b/tests/asgi/urls.py
@@ -1,3 +1,5 @@
+import threading
+
 from django.http import FileResponse, HttpResponse
 from django.urls import path
 
@@ -14,6 +16,18 @@ def hello_meta(request):
     )
 
 
+def sync_waiter(request):
+    with sync_waiter.lock:
+        sync_waiter.active_threads.add(threading.current_thread())
+    sync_waiter.barrier.wait(timeout=0.5)
+    return hello(request)
+
+
+sync_waiter.active_threads = set()
+sync_waiter.lock = threading.Lock()
+sync_waiter.barrier = threading.Barrier(2)
+
+
 test_filename = __file__
 
 
@@ -21,4 +35,5 @@ urlpatterns = [
     path('', hello),
     path('file/', lambda x: FileResponse(open(test_filename, 'rb'))),
     path('meta/', hello_meta),
+    path('wait/', sync_waiter),
 ]