Use native context parameter for create_task on Python 3.11+ (#2859)

## Summary

Both HTTP protocol implementations (`h11_impl.py` and
`httptools_impl.py`) use
`contextvars.Context().run(loop.create_task, ...)` to start ASGI tasks
with a
fresh context. Python 3.11 added a `context=` parameter to
`create_task()`,
which avoids the extra indirection through `Context.run()`.

This has been a known TODO in the codebase for a while. Under
high-concurrency
workloads, the `Context().run()` wrapper adds a small but measurable
overhead
per request compared to the native kwarg, since it has to set up and
tear down
the context activation around the call.

The change uses `sys.version_info` to branch at runtime — 3.11+ gets the
native
kwarg, older versions keep the existing behavior. Coverage pragmas
follow the
existing convention in `_types.py` (`py-lt-311` / `py-gte-311` on the
branch
lines).

---------

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

tests/benchmarks/http.py

@@ -131,7 +131,7 @@ class MockLoop:
         self._tasks: list[asyncio.Task[Any]] = []
         self._later: list[MockTimerHandle] = []
 
-    def create_task(self, coroutine: Any) -> Any:
+    def create_task(self, coroutine: Any, **kwargs: Any) -> Any:
         self._tasks.insert(0, coroutine)
         return MockTask()
 

tests/protocols/test_http.py

@@ -226,7 +226,7 @@ class MockLoop:
         self._tasks: list[asyncio.Task[Any]] = []
         self._later: list[MockTimerHandle] = []
 
-    def create_task(self, coroutine: Any) -> Any:
+    def create_task(self, coroutine: Any, **kwargs: Any) -> Any:
         self._tasks.insert(0, coroutine)
         return MockTask()
 

uvicorn/protocols/http/h11_impl.py

@@ -4,6 +4,7 @@ import asyncio
 import contextvars
 import http
 import logging
+import sys
 from collections.abc import Callable
 from typing import Any, Literal
 from urllib.parse import unquote
@@ -252,9 +253,10 @@ class H11Protocol(asyncio.Protocol):
                 # For the asyncio loop, we need to explicitly start with an empty context
                 # as it can be polluted from previous ASGI runs.
                 # See https://github.com/python/cpython/issues/140947 for details.
-                task = contextvars.Context().run(self.loop.create_task, self.cycle.run_asgi(app))
-                # TODO: Replace the line above with the line below for Python >= 3.11
-                # task = self.loop.create_task(self.cycle.run_asgi(app), context=contextvars.Context())
+                if sys.version_info >= (3, 11):  # pragma: py-lt-311
+                    task = self.loop.create_task(self.cycle.run_asgi(app), context=contextvars.Context())
+                else:  # pragma: py-gte-311
+                    task = contextvars.Context().run(self.loop.create_task, self.cycle.run_asgi(app))
                 task.add_done_callback(self.tasks.discard)
                 self.tasks.add(task)
 

uvicorn/protocols/http/httptools_impl.py

@@ -5,6 +5,7 @@ import contextvars
 import http
 import logging
 import re
+import sys
 import urllib
 from asyncio.events import TimerHandle
 from collections import deque
@@ -291,9 +292,10 @@ class HttpToolsProtocol(asyncio.Protocol):
             # For the asyncio loop, we need to explicitly start with an empty context
             # as it can be polluted from previous ASGI runs.
             # See https://github.com/python/cpython/issues/140947 for details.
-            task = contextvars.Context().run(self.loop.create_task, self.cycle.run_asgi(app))
-            # TODO: Replace the line above with the line below for Python >= 3.11
-            # task = self.loop.create_task(self.cycle.run_asgi(app), context=contextvars.Context())
+            if sys.version_info >= (3, 11):  # pragma: py-lt-311
+                task = self.loop.create_task(self.cycle.run_asgi(app), context=contextvars.Context())
+            else:  # pragma: py-gte-311
+                task = contextvars.Context().run(self.loop.create_task, self.cycle.run_asgi(app))
             task.add_done_callback(self.tasks.discard)
             self.tasks.add(task)
         else: