diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index 77c70aaa7b986e..7a6837546d930f 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -14,6 +14,7 @@ """ import collections +import contextvars import collections.abc import concurrent.futures import errno @@ -290,6 +291,7 @@ def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog, self._ssl_shutdown_timeout = ssl_shutdown_timeout self._serving = False self._serving_forever_fut = None + self._context = contextvars.copy_context() def __repr__(self): return f'<{self.__class__.__name__} sockets={self.sockets!r}>' @@ -319,7 +321,7 @@ def _start_serving(self): self._loop._start_serving( self._protocol_factory, sock, self._ssl_context, self, self._backlog, self._ssl_handshake_timeout, - self._ssl_shutdown_timeout) + self._ssl_shutdown_timeout, context=self._context) def get_loop(self): return self._loop @@ -509,7 +511,8 @@ def _make_ssl_transport( extra=None, server=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, - call_connection_made=True): + call_connection_made=True, + context=None): """Create SSL transport.""" raise NotImplementedError @@ -1213,9 +1216,10 @@ async def _create_connection_transport( self, sock, protocol_factory, ssl, server_hostname, server_side=False, ssl_handshake_timeout=None, - ssl_shutdown_timeout=None): + ssl_shutdown_timeout=None, context=None): sock.setblocking(False) + context = context if context is not None else contextvars.copy_context() protocol = protocol_factory() waiter = self.create_future() @@ -1225,9 +1229,10 @@ async def _create_connection_transport( sock, protocol, sslcontext, waiter, server_side=server_side, server_hostname=server_hostname, ssl_handshake_timeout=ssl_handshake_timeout, - ssl_shutdown_timeout=ssl_shutdown_timeout) + ssl_shutdown_timeout=ssl_shutdown_timeout, + context=context) else: - transport = self._make_socket_transport(sock, protocol, waiter) + transport = self._make_socket_transport(sock, protocol, waiter, context=context) try: await waiter diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py index 3fa93b14a6787f..2dc1569d780791 100644 --- a/Lib/asyncio/proactor_events.py +++ b/Lib/asyncio/proactor_events.py @@ -642,7 +642,7 @@ def __init__(self, proactor): signal.set_wakeup_fd(self._csock.fileno()) def _make_socket_transport(self, sock, protocol, waiter=None, - extra=None, server=None): + extra=None, server=None, context=None): return _ProactorSocketTransport(self, sock, protocol, waiter, extra, server) @@ -651,7 +651,7 @@ def _make_ssl_transport( *, server_side=False, server_hostname=None, extra=None, server=None, ssl_handshake_timeout=None, - ssl_shutdown_timeout=None): + ssl_shutdown_timeout=None, context=None): ssl_protocol = sslproto.SSLProtocol( self, protocol, sslcontext, waiter, server_side, server_hostname, @@ -837,7 +837,7 @@ def _write_to_self(self): def _start_serving(self, protocol_factory, sock, sslcontext=None, server=None, backlog=100, ssl_handshake_timeout=None, - ssl_shutdown_timeout=None): + ssl_shutdown_timeout=None, context=None): def loop(f=None): try: diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py index ff7e16df3c6273..9685e7fc05d241 100644 --- a/Lib/asyncio/selector_events.py +++ b/Lib/asyncio/selector_events.py @@ -67,10 +67,10 @@ def __init__(self, selector=None): self._transports = weakref.WeakValueDictionary() def _make_socket_transport(self, sock, protocol, waiter=None, *, - extra=None, server=None): + extra=None, server=None, context=None): self._ensure_fd_no_transport(sock) return _SelectorSocketTransport(self, sock, protocol, waiter, - extra, server) + extra, server, context=context) def _make_ssl_transport( self, rawsock, protocol, sslcontext, waiter=None, @@ -78,16 +78,17 @@ def _make_ssl_transport( extra=None, server=None, ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT, ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT, + context=None, ): self._ensure_fd_no_transport(rawsock) ssl_protocol = sslproto.SSLProtocol( self, protocol, sslcontext, waiter, server_side, server_hostname, ssl_handshake_timeout=ssl_handshake_timeout, - ssl_shutdown_timeout=ssl_shutdown_timeout + ssl_shutdown_timeout=ssl_shutdown_timeout, ) _SelectorSocketTransport(self, rawsock, ssl_protocol, - extra=extra, server=server) + extra=extra, server=server, context=context) return ssl_protocol._app_transport def _make_datagram_transport(self, sock, protocol, @@ -159,16 +160,16 @@ def _write_to_self(self): def _start_serving(self, protocol_factory, sock, sslcontext=None, server=None, backlog=100, ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT, - ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT): + ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT, context=None): self._add_reader(sock.fileno(), self._accept_connection, protocol_factory, sock, sslcontext, server, backlog, - ssl_handshake_timeout, ssl_shutdown_timeout) + ssl_handshake_timeout, ssl_shutdown_timeout, context) def _accept_connection( self, protocol_factory, sock, sslcontext=None, server=None, backlog=100, ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT, - ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT): + ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT, context=None): # This method is only called once for each event loop tick where the # listening socket has triggered an EVENT_READ. There may be multiple # connections waiting for an .accept() so it is called in a loop. @@ -204,21 +205,22 @@ def _accept_connection( self._start_serving, protocol_factory, sock, sslcontext, server, backlog, ssl_handshake_timeout, - ssl_shutdown_timeout) + ssl_shutdown_timeout, context) else: raise # The event loop will catch, log and ignore it. else: extra = {'peername': addr} + conn_context = context.copy() if context is not None else None accept = self._accept_connection2( protocol_factory, conn, extra, sslcontext, server, - ssl_handshake_timeout, ssl_shutdown_timeout) - self.create_task(accept) + ssl_handshake_timeout, ssl_shutdown_timeout, context=conn_context) + self.create_task(accept, context=conn_context) async def _accept_connection2( self, protocol_factory, conn, extra, sslcontext=None, server=None, ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT, - ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT): + ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT, context=None): protocol = None transport = None try: @@ -229,11 +231,12 @@ async def _accept_connection2( conn, protocol, sslcontext, waiter=waiter, server_side=True, extra=extra, server=server, ssl_handshake_timeout=ssl_handshake_timeout, - ssl_shutdown_timeout=ssl_shutdown_timeout) + ssl_shutdown_timeout=ssl_shutdown_timeout, + context=context) else: transport = self._make_socket_transport( conn, protocol, waiter=waiter, extra=extra, - server=server) + server=server, context=context) try: await waiter @@ -275,9 +278,9 @@ def _ensure_fd_no_transport(self, fd): f'File descriptor {fd!r} is used by transport ' f'{transport!r}') - def _add_reader(self, fd, callback, *args): + def _add_reader(self, fd, callback, *args, context=None): self._check_closed() - handle = events.Handle(callback, args, self, None) + handle = events.Handle(callback, args, self, context=context) key = self._selector.get_map().get(fd) if key is None: self._selector.register(fd, selectors.EVENT_READ, @@ -309,9 +312,9 @@ def _remove_reader(self, fd): else: return False - def _add_writer(self, fd, callback, *args): + def _add_writer(self, fd, callback, *args, context=None): self._check_closed() - handle = events.Handle(callback, args, self, None) + handle = events.Handle(callback, args, self, context=context) key = self._selector.get_map().get(fd) if key is None: self._selector.register(fd, selectors.EVENT_WRITE, @@ -770,7 +773,7 @@ class _SelectorTransport(transports._FlowControlMixin, # exception) _sock = None - def __init__(self, loop, sock, protocol, extra=None, server=None): + def __init__(self, loop, sock, protocol, extra=None, server=None, context=None): super().__init__(extra, loop) self._extra['socket'] = trsock.TransportSocket(sock) try: @@ -784,7 +787,7 @@ def __init__(self, loop, sock, protocol, extra=None, server=None): self._extra['peername'] = None self._sock = sock self._sock_fd = sock.fileno() - + self._context = context self._protocol_connected = False self.set_protocol(protocol) @@ -866,7 +869,7 @@ def close(self): if not self._buffer: self._conn_lost += 1 self._loop._remove_writer(self._sock_fd) - self._loop.call_soon(self._call_connection_lost, None) + self._call_soon(self._call_connection_lost, None) def __del__(self, _warn=warnings.warn): if self._sock is not None: @@ -899,7 +902,7 @@ def _force_close(self, exc): self._closing = True self._loop._remove_reader(self._sock_fd) self._conn_lost += 1 - self._loop.call_soon(self._call_connection_lost, exc) + self._call_soon(self._call_connection_lost, exc) def _call_connection_lost(self, exc): try: @@ -921,8 +924,13 @@ def get_write_buffer_size(self): def _add_reader(self, fd, callback, *args): if not self.is_reading(): return - self._loop._add_reader(fd, callback, *args) + self._loop._add_reader(fd, callback, *args, context=self._context) + def _add_writer(self, fd, callback, *args): + self._loop._add_writer(fd, callback, *args, context=self._context) + + def _call_soon(self, callback, *args): + self._loop.call_soon(callback, *args, context=self._context) class _SelectorSocketTransport(_SelectorTransport): @@ -930,10 +938,9 @@ class _SelectorSocketTransport(_SelectorTransport): _sendfile_compatible = constants._SendfileMode.TRY_NATIVE def __init__(self, loop, sock, protocol, waiter=None, - extra=None, server=None): - + extra=None, server=None, context=None): self._read_ready_cb = None - super().__init__(loop, sock, protocol, extra, server) + super().__init__(loop, sock, protocol, extra, server, context) self._eof = False self._empty_waiter = None if _HAS_SENDMSG: @@ -945,14 +952,12 @@ def __init__(self, loop, sock, protocol, waiter=None, # decreases the latency (in some cases significantly.) base_events._set_nodelay(self._sock) - self._loop.call_soon(self._protocol.connection_made, self) + self._call_soon(self._protocol.connection_made, self) # only start reading when connection_made() has been called - self._loop.call_soon(self._add_reader, - self._sock_fd, self._read_ready) + self._call_soon(self._add_reader, self._sock_fd, self._read_ready) if waiter is not None: # only wake up the waiter when connection_made() has been called - self._loop.call_soon(futures._set_result_unless_cancelled, - waiter, None) + self._call_soon(futures._set_result_unless_cancelled, waiter, None) def set_protocol(self, protocol): if isinstance(protocol, protocols.BufferedProtocol): @@ -1081,7 +1086,7 @@ def write(self, data): if not data: return # Not all was written; register write handler. - self._loop._add_writer(self._sock_fd, self._write_ready) + self._add_writer(self._sock_fd, self._write_ready) # Add it to the buffer. self._buffer.append(data) @@ -1185,7 +1190,7 @@ def writelines(self, list_of_data): self._write_ready() # If the entire buffer couldn't be written, register a write handler if self._buffer: - self._loop._add_writer(self._sock_fd, self._write_ready) + self._add_writer(self._sock_fd, self._write_ready) self._maybe_pause_protocol() def can_write_eof(self): @@ -1226,14 +1231,12 @@ def __init__(self, loop, sock, protocol, address=None, super().__init__(loop, sock, protocol, extra) self._address = address self._buffer_size = 0 - self._loop.call_soon(self._protocol.connection_made, self) + self._call_soon(self._protocol.connection_made, self) # only start reading when connection_made() has been called - self._loop.call_soon(self._add_reader, - self._sock_fd, self._read_ready) + self._call_soon(self._add_reader, self._sock_fd, self._read_ready) if waiter is not None: # only wake up the waiter when connection_made() has been called - self._loop.call_soon(futures._set_result_unless_cancelled, - waiter, None) + self._call_soon(futures._set_result_unless_cancelled, waiter, None) def get_write_buffer_size(self): return self._buffer_size @@ -1280,7 +1283,7 @@ def sendto(self, data, addr=None): self._sock.sendto(data, addr) return except (BlockingIOError, InterruptedError): - self._loop._add_writer(self._sock_fd, self._sendto_ready) + self._add_writer(self._sock_fd, self._sendto_ready) except OSError as exc: self._protocol.error_received(exc) return diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py index 8c02de77c24740..e59bc25668b4cb 100644 --- a/Lib/test/test_asyncio/test_base_events.py +++ b/Lib/test/test_asyncio/test_base_events.py @@ -1696,7 +1696,8 @@ def mock_make_ssl_transport(sock, protocol, sslcontext, waiter, server_side=False, server_hostname='python.org', ssl_handshake_timeout=handshake_timeout, - ssl_shutdown_timeout=shutdown_timeout) + ssl_shutdown_timeout=shutdown_timeout, + context=ANY) # Next try an explicit server_hostname. self.loop._make_ssl_transport.reset_mock() coro = self.loop.create_connection( @@ -1711,7 +1712,8 @@ def mock_make_ssl_transport(sock, protocol, sslcontext, waiter, server_side=False, server_hostname='perl.com', ssl_handshake_timeout=handshake_timeout, - ssl_shutdown_timeout=shutdown_timeout) + ssl_shutdown_timeout=shutdown_timeout, + context=ANY) # Finally try an explicit empty server_hostname. self.loop._make_ssl_transport.reset_mock() coro = self.loop.create_connection( @@ -1726,7 +1728,8 @@ def mock_make_ssl_transport(sock, protocol, sslcontext, waiter, server_side=False, server_hostname='', ssl_handshake_timeout=handshake_timeout, - ssl_shutdown_timeout=shutdown_timeout) + ssl_shutdown_timeout=shutdown_timeout, + context=ANY) def test_create_connection_no_ssl_server_hostname_errors(self): # When not using ssl, server_hostname must be None. @@ -2104,7 +2107,7 @@ def test_accept_connection_exception(self, m_log): constants.ACCEPT_RETRY_DELAY, # self.loop._start_serving mock.ANY, - MyProto, sock, None, None, mock.ANY, mock.ANY, mock.ANY) + MyProto, sock, None, None, mock.ANY, mock.ANY, mock.ANY, mock.ANY) def test_call_coroutine(self): async def simple_coroutine(): diff --git a/Lib/test/test_asyncio/test_server_context.py b/Lib/test/test_asyncio/test_server_context.py new file mode 100644 index 00000000000000..3f15654a1af467 --- /dev/null +++ b/Lib/test/test_asyncio/test_server_context.py @@ -0,0 +1,314 @@ +import asyncio +import contextvars +import unittest +import sys + +from unittest import TestCase + +try: + import ssl +except ImportError: + ssl = None + +from test.test_asyncio import utils as test_utils + +def tearDownModule(): + asyncio.events._set_event_loop_policy(None) + +class ServerContextvarsTestCase: + loop_factory = None # To be defined in subclasses + server_ssl_context = None # To be defined in subclasses for SSL tests + client_ssl_context = None # To be defined in subclasses for SSL tests + + def run_coro(self, coro): + return asyncio.run(coro, loop_factory=self.loop_factory) + + def test_start_server1(self): + # Test that asyncio.start_server captures the context at the time of server creation + async def test(): + var = contextvars.ContextVar("var", default="default") + + async def handle_client(reader, writer): + value = var.get() + writer.write(value.encode()) + await writer.drain() + writer.close() + + server = await asyncio.start_server(handle_client, '127.0.0.1', 0, + ssl=self.server_ssl_context) + # change the value + var.set("after_server") + + async def client(addr): + reader, writer = await asyncio.open_connection(*addr, + ssl=self.client_ssl_context) + data = await reader.read(100) + writer.close() + await writer.wait_closed() + return data.decode() + + async with server: + addr = server.sockets[0].getsockname() + self.assertEqual(await client(addr), "default") + + self.assertEqual(var.get(), "after_server") + + self.run_coro(test()) + + def test_start_server2(self): + # Test that mutations to the context in one handler don't affect other handlers or the server's context + async def test(): + var = contextvars.ContextVar("var", default="default") + + async def handle_client(reader, writer): + value = var.get() + writer.write(value.encode()) + var.set("in_handler") + await writer.drain() + writer.close() + + server = await asyncio.start_server(handle_client, '127.0.0.1', 0, + ssl=self.server_ssl_context) + var.set("after_server") + + async def client(addr): + reader, writer = await asyncio.open_connection(*addr, + ssl=self.client_ssl_context) + data = await reader.read(100) + writer.close() + await writer.wait_closed() + return data.decode() + + async with server: + addr = server.sockets[0].getsockname() + self.assertEqual(await client(addr), "default") + self.assertEqual(await client(addr), "default") + self.assertEqual(await client(addr), "default") + + self.assertEqual(var.get(), "after_server") + + self.run_coro(test()) + + def test_start_server3(self): + # Test that mutations to context in concurrent handlers don't affect each other or the server's context + async def test(): + var = contextvars.ContextVar("var", default="default") + var.set("before_server") + + async def handle_client(reader, writer): + writer.write(var.get().encode()) + await writer.drain() + writer.close() + + server = await asyncio.start_server(handle_client, '127.0.0.1', 0, + ssl=self.server_ssl_context) + var.set("after_server") + + async def client(addr): + reader, writer = await asyncio.open_connection(*addr, + ssl=self.client_ssl_context) + data = await reader.read(100) + self.assertEqual(data.decode(), "before_server") + writer.close() + await writer.wait_closed() + + async with server: + addr = server.sockets[0].getsockname() + async with asyncio.TaskGroup() as tg: + for _ in range(100): + tg.create_task(client(addr)) + + self.assertEqual(var.get(), "after_server") + + self.run_coro(test()) + + def test_create_server1(self): + # Test that loop.create_server captures the context at the time of server creation + # and that mutations to the context in protocol callbacks don't affect the server's context + async def test(): + var = contextvars.ContextVar("var", default="default") + + class EchoProtocol(asyncio.Protocol): + def connection_made(self, transport): + self.transport = transport + value = var.get() + var.set("in_handler") + self.transport.write(value.encode()) + self.transport.close() + + server = await asyncio.get_running_loop().create_server( + lambda: EchoProtocol(), '127.0.0.1', 0, + ssl=self.server_ssl_context) + var.set("after_server") + + async def client(addr): + reader, writer = await asyncio.open_connection(*addr, + ssl=self.client_ssl_context) + data = await reader.read(100) + self.assertEqual(data.decode(), "default") + writer.close() + await writer.wait_closed() + + async with server: + addr = server.sockets[0].getsockname() + await client(addr) + + self.assertEqual(var.get(), "after_server") + + self.run_coro(test()) + + def test_create_server2(self): + # Test that mutations to context in one protocol instance don't affect other instances or the server's context + async def test(): + var = contextvars.ContextVar("var", default="default") + + class EchoProtocol(asyncio.Protocol): + def __init__(self): + super().__init__() + assert var.get() == "default", var.get() + def connection_made(self, transport): + self.transport = transport + value = var.get() + var.set("in_handler") + self.transport.write(value.encode()) + self.transport.close() + + server = await asyncio.get_running_loop().create_server( + lambda: EchoProtocol(), '127.0.0.1', 0, + ssl=self.server_ssl_context) + + var.set("after_server") + + async def client(addr, expected): + reader, writer = await asyncio.open_connection(*addr, + ssl=self.client_ssl_context) + data = await reader.read(100) + self.assertEqual(data.decode(), expected) + writer.close() + await writer.wait_closed() + + async with server: + addr = server.sockets[0].getsockname() + await client(addr, "default") + await client(addr, "default") + + self.assertEqual(var.get(), "after_server") + + self.run_coro(test()) + + def test_gh140947(self): + # See https://github.com/python/cpython/issues/140947 + + cvar1 = contextvars.ContextVar("cvar1") + cvar2 = contextvars.ContextVar("cvar2") + cvar3 = contextvars.ContextVar("cvar3") + results = {} + is_ssl = self.server_ssl_context is not None + + def capture_context(meth): + result = [] + for k,v in contextvars.copy_context().items(): + if k.name.startswith("cvar"): + result.append((k.name, v)) + results[meth] = sorted(result) + + class DemoProtocol(asyncio.Protocol): + def __init__(self, on_conn_lost): + self.transport = None + self.on_conn_lost = on_conn_lost + self.tasks = set() + + def connection_made(self, transport): + capture_context("connection_made") + self.transport = transport + + def data_received(self, data): + capture_context("data_received") + + task = asyncio.create_task(self.asgi()) + self.tasks.add(task) + task.add_done_callback(self.tasks.discard) + + self.transport.pause_reading() + + def connection_lost(self, exc): + capture_context("connection_lost") + if not self.on_conn_lost.done(): + self.on_conn_lost.set_result(True) + + async def asgi(self): + capture_context("asgi start") + cvar1.set(True) + # make sure that we only resume after the pause + # otherwise the resume does nothing + if is_ssl: + while not self.transport._ssl_protocol._app_reading_paused: + await asyncio.sleep(0.01) + else: + while not self.transport._paused: + await asyncio.sleep(0.01) + cvar2.set(True) + self.transport.resume_reading() + cvar3.set(True) + capture_context("asgi end") + + async def main(): + loop = asyncio.get_running_loop() + on_conn_lost = loop.create_future() + + server = await loop.create_server( + lambda: DemoProtocol(on_conn_lost), '127.0.0.1', 0, + ssl=self.server_ssl_context) + async with server: + addr = server.sockets[0].getsockname() + reader, writer = await asyncio.open_connection(*addr, + ssl=self.client_ssl_context) + writer.write(b"anything") + await writer.drain() + writer.close() + await writer.wait_closed() + await on_conn_lost + + self.run_coro(main()) + self.assertDictEqual(results, { + "connection_made": [], + "data_received": [], + "asgi start": [], + "asgi end": [("cvar1", True), ("cvar2", True), ("cvar3", True)], + "connection_lost": [], + }) + + +class AsyncioEventLoopTests(TestCase, ServerContextvarsTestCase): + loop_factory = staticmethod(asyncio.new_event_loop) + +@unittest.skipUnless(ssl, "SSL not available") +class AsyncioEventLoopSSLTests(AsyncioEventLoopTests): + def setUp(self): + super().setUp() + self.server_ssl_context = test_utils.simple_server_sslcontext() + self.client_ssl_context = test_utils.simple_client_sslcontext() + +if sys.platform == "win32": + class AsyncioProactorEventLoopTests(TestCase, ServerContextvarsTestCase): + loop_factory = asyncio.ProactorEventLoop + + class AsyncioSelectorEventLoopTests(TestCase, ServerContextvarsTestCase): + loop_factory = asyncio.SelectorEventLoop + + @unittest.skipUnless(ssl, "SSL not available") + class AsyncioProactorEventLoopSSLTests(AsyncioProactorEventLoopTests): + def setUp(self): + super().setUp() + self.server_ssl_context = test_utils.simple_server_sslcontext() + self.client_ssl_context = test_utils.simple_client_sslcontext() + + @unittest.skipUnless(ssl, "SSL not available") + class AsyncioSelectorEventLoopSSLTests(AsyncioSelectorEventLoopTests): + def setUp(self): + super().setUp() + self.server_ssl_context = test_utils.simple_server_sslcontext() + self.client_ssl_context = test_utils.simple_client_sslcontext() + +if __name__ == "__main__": + unittest.main() diff --git a/Lib/test/test_asyncio/utils.py b/Lib/test/test_asyncio/utils.py index a480e16e81bb91..62cfcf8ceb5f2a 100644 --- a/Lib/test/test_asyncio/utils.py +++ b/Lib/test/test_asyncio/utils.py @@ -388,8 +388,8 @@ def close(self): else: # pragma: no cover raise AssertionError("Time generator is not finished") - def _add_reader(self, fd, callback, *args): - self.readers[fd] = events.Handle(callback, args, self, None) + def _add_reader(self, fd, callback, *args, context=None): + self.readers[fd] = events.Handle(callback, args, self, context) def _remove_reader(self, fd): self.remove_reader_count[fd] += 1 @@ -414,8 +414,8 @@ def assert_no_reader(self, fd): if fd in self.readers: raise AssertionError(f'fd {fd} is registered') - def _add_writer(self, fd, callback, *args): - self.writers[fd] = events.Handle(callback, args, self, None) + def _add_writer(self, fd, callback, *args, context=None): + self.writers[fd] = events.Handle(callback, args, self, context) def _remove_writer(self, fd): self.remove_writer_count[fd] += 1 diff --git a/Misc/NEWS.d/next/Library/2026-03-21-08-23-26.gh-issue-140947.owZ4r_.rst b/Misc/NEWS.d/next/Library/2026-03-21-08-23-26.gh-issue-140947.owZ4r_.rst new file mode 100644 index 00000000000000..88e787e8549a1d --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-03-21-08-23-26.gh-issue-140947.owZ4r_.rst @@ -0,0 +1 @@ +Fix incorrect contextvars handling in server tasks created by :mod:`asyncio`. Patch by Kumar Aditya.