From dae997fa066829c0e374adce96c1ce5cf3a16028 Mon Sep 17 00:00:00 2001 From: Davide Rizzo Date: Mon, 11 Sep 2023 12:49:41 +0200 Subject: [PATCH 1/6] asyncio: remove outgoing buffer from sslproto --- Lib/asyncio/sslproto.py | 111 ++++------------------------------------ 1 file changed, 11 insertions(+), 100 deletions(-) diff --git a/Lib/asyncio/sslproto.py b/Lib/asyncio/sslproto.py index 84e82fc69cc5fd..d7d5f3dc001ecc 100644 --- a/Lib/asyncio/sslproto.py +++ b/Lib/asyncio/sslproto.py @@ -162,16 +162,14 @@ def set_write_buffer_limits(self, high=None, low=None): reduces opportunities for doing I/O and computation concurrently. """ - self._ssl_protocol._set_write_buffer_limits(high, low) - self._ssl_protocol._control_app_writing() + self._ssl_protocol._transport.set_write_buffer_limits(high, low) def get_write_buffer_limits(self): - return (self._ssl_protocol._outgoing_low_water, - self._ssl_protocol._outgoing_high_water) + return self._ssl_protocol._transport.get_write_buffer_limits() def get_write_buffer_size(self): """Return the current size of the write buffers.""" - return self._ssl_protocol._get_write_buffer_size() + return self._ssl_protocol._transport.get_write_buffer_size() def set_read_buffer_limits(self, high=None, low=None): """Set the high- and low-water limits for read flow control. @@ -203,11 +201,6 @@ def get_read_buffer_size(self): """Return the current size of the read buffer.""" return self._ssl_protocol._get_read_buffer_size() - @property - def _protocol_paused(self): - # Required for sendfile fallback pause_writing/resume_writing logic - return self._ssl_protocol._app_writing_paused - def write(self, data): """Write some data bytes to the transport. @@ -254,11 +247,6 @@ def _force_close(self, exc): if self._ssl_protocol is not None: self._ssl_protocol._abort(exc) - def _test__append_write_backlog(self, data): - # for test only - self._ssl_protocol._write_backlog.append(data) - self._ssl_protocol._write_buffer_size += len(data) - class SSLProtocol(protocols.BufferedProtocol): max_size = 256 * 1024 # Buffer size passed to read() @@ -305,10 +293,6 @@ def __init__(self, loop, app_protocol, sslcontext, waiter, # completes. self._extra = dict(sslcontext=sslcontext) - # App data write buffering - self._write_backlog = collections.deque() - self._write_buffer_size = 0 - self._waiter = waiter self._loop = loop self._set_app_protocol(app_protocol) @@ -334,8 +318,6 @@ def __init__(self, loop, app_protocol, sslcontext, waiter, # Flow Control - self._ssl_writing_paused = False - self._app_reading_paused = False self._ssl_reading_paused = False @@ -344,10 +326,6 @@ def __init__(self, loop, app_protocol, sslcontext, waiter, self._set_read_buffer_limits() self._eof_received = False - self._app_writing_paused = False - self._outgoing_high_water = 0 - self._outgoing_low_water = 0 - self._set_write_buffer_limits() self._get_app_transport() def _set_app_protocol(self, app_protocol): @@ -397,7 +375,6 @@ def connection_lost(self, exc): meaning a regular EOF is received or the connection was aborted or closed). """ - self._write_backlog.clear() self._outgoing.read() self._conn_lost += 1 @@ -474,7 +451,6 @@ def eof_received(self): self._do_flush() elif self._state == SSLProtocolState.FLUSHING: - self._do_write() self._set_state(SSLProtocolState.SHUTDOWN) self._do_shutdown() @@ -699,38 +675,19 @@ def _write_appdata(self, list_of_data): return for data in list_of_data: - self._write_backlog.append(data) - self._write_buffer_size += len(data) + self._sslobj.write(data) try: if self._state == SSLProtocolState.WRAPPED: - self._do_write() + self._process_outgoing() except Exception as ex: self._fatal_error(ex, 'Fatal error on SSL protocol') - def _do_write(self): - try: - while self._write_backlog: - data = self._write_backlog[0] - count = self._sslobj.write(data) - data_len = len(data) - if count < data_len: - self._write_backlog[0] = data[count:] - self._write_buffer_size -= count - else: - del self._write_backlog[0] - self._write_buffer_size -= data_len - except SSLAgainErrors: - pass - self._process_outgoing() - def _process_outgoing(self): - if not self._ssl_writing_paused: - data = self._outgoing.read() - if len(data): - self._transport.write(data) - self._control_app_writing() + data = self._outgoing.read() + if len(data): + self._transport.write(data) # Incoming flow @@ -748,10 +705,7 @@ def _do_read(self): self._do_read__buffered() else: self._do_read__copied() - if self._write_backlog: - self._do_write() - else: - self._process_outgoing() + self._process_outgoing() self._control_ssl_reading() except ssl.SSLError as ex: # gh-98078: send the fatal TLS alert left in the outgoing @@ -833,46 +787,6 @@ def _call_eof_received(self): except BaseException as ex: self._fatal_error(ex, 'Error calling eof_received()') - # Flow control for writes from APP socket - - def _control_app_writing(self): - size = self._get_write_buffer_size() - if size >= self._outgoing_high_water and not self._app_writing_paused: - self._app_writing_paused = True - try: - self._app_protocol.pause_writing() - except (KeyboardInterrupt, SystemExit): - raise - except BaseException as exc: - self._loop.call_exception_handler({ - 'message': 'protocol.pause_writing() failed', - 'exception': exc, - 'transport': self._app_transport, - 'protocol': self, - }) - elif size <= self._outgoing_low_water and self._app_writing_paused: - self._app_writing_paused = False - try: - self._app_protocol.resume_writing() - except (KeyboardInterrupt, SystemExit): - raise - except BaseException as exc: - self._loop.call_exception_handler({ - 'message': 'protocol.resume_writing() failed', - 'exception': exc, - 'transport': self._app_transport, - 'protocol': self, - }) - - def _get_write_buffer_size(self): - return self._outgoing.pending + self._write_buffer_size - - def _set_write_buffer_limits(self, high=None, low=None): - high, low = add_flowcontrol_defaults( - high, low, constants.FLOW_CONTROL_HIGH_WATER_SSL_WRITE) - self._outgoing_high_water = high - self._outgoing_low_water = low - # Flow control for reads to APP socket def _pause_reading(self): @@ -917,16 +831,13 @@ def pause_writing(self): """Called when the low-level transport's buffer goes over the high-water mark. """ - assert not self._ssl_writing_paused - self._ssl_writing_paused = True + self._app_protocol.pause_writing() def resume_writing(self): """Called when the low-level transport's buffer drains below the low-water mark. """ - assert self._ssl_writing_paused - self._ssl_writing_paused = False - self._process_outgoing() + self._app_protocol.resume_writing() def _fatal_error(self, exc, message='Fatal error on transport'): if self._transport: From 7e323d819701fa837342a57ea1b9a485b1fd3f3f Mon Sep 17 00:00:00 2001 From: Davide Rizzo Date: Tue, 19 Sep 2023 16:16:32 +0200 Subject: [PATCH 2/6] backport test_remote_shutdown_receives_trailing_data from uvloop --- Lib/test/test_asyncio/test_ssl.py | 54 ++++++++++++++++++++++--------- 1 file changed, 38 insertions(+), 16 deletions(-) diff --git a/Lib/test/test_asyncio/test_ssl.py b/Lib/test/test_asyncio/test_ssl.py index 902dc7e04e5809..5c1dd2d7e35957 100644 --- a/Lib/test/test_asyncio/test_ssl.py +++ b/Lib/test/test_asyncio/test_ssl.py @@ -1301,7 +1301,6 @@ async def client(addr): def test_remote_shutdown_receives_trailing_data(self): CHUNK = 1024 * 128 - SIZE = 32 sslctx = self._create_server_ssl_context( test_utils.ONLYCERT, @@ -1339,25 +1338,38 @@ def server(sock): sslobj.write(b'pong') sock.send(outgoing.read()) - time.sleep(0.2) # wait for the peer to fill its backlog + data_len = 0 + filled.wait() + # trigger peer's resume_writing() + incoming.write(sock.recv(65536 * 4)) + while True: + try: + chunk = len(sslobj.read(16384)) + data_len += chunk + except ssl.SSLWantReadError: + break # send close_notify but don't wait for response with self.assertRaises(ssl.SSLWantReadError): sslobj.unwrap() sock.send(outgoing.read()) + eof_received.wait() # should receive all data - data_len = 0 while True: try: chunk = len(sslobj.read(16384)) data_len += chunk except ssl.SSLWantReadError: incoming.write(sock.recv(16384)) + if not incoming.pending: + # EOF received + break except ssl.SSLZeroReturnError: break - self.assertEqual(data_len, CHUNK * SIZE) + assert count > 0 + self.assertEqual(data_len, CHUNK * count) # verify that close_notify is received sslobj.unwrap() @@ -1369,19 +1381,20 @@ def eof_server(sock): self.assertEqual(sock.recv_all(4), b'ping') sock.send(b'pong') - time.sleep(0.2) # wait for the peer to fill its backlog + filled.wait() # send EOF sock.shutdown(socket.SHUT_WR) # should receive all data - data = sock.recv_all(CHUNK * SIZE) - self.assertEqual(len(data), CHUNK * SIZE) + assert count > 0 + data = sock.recv_all(CHUNK * count) + self.assertEqual(len(data), CHUNK * count) sock.close() async def client(addr): - nonlocal future + nonlocal future, count future = self.loop.create_future() reader, writer = await asyncio.open_connection( @@ -1393,15 +1406,18 @@ async def client(addr): data = await reader.readexactly(4) self.assertEqual(data, b'pong') - # fill write backlog in a hacky way - renegotiation won't help - for _ in range(SIZE): - writer.transport._test__append_write_backlog(b'x' * CHUNK) + count = 0 + # Fill the buffer - with a hack to ensure the protocol is + # paused without actually waiting on drain + while not writer._protocol._paused: + writer.write(b'x' * CHUNK) + count += 1 + await asyncio.sleep(0) + filled.set() - try: - data = await reader.read() - self.assertEqual(data, b'') - except (BrokenPipeError, ConnectionResetError): - pass + data = await reader.read() + self.assertEqual(data, b'') + eof_received.set() await future @@ -1418,9 +1434,15 @@ def wrapper(sock): self.loop.call_soon_threadsafe(future.set_result, None) return wrapper + filled = threading.Event() + eof_received = threading.Event() + count = 0 with self.tcp_server(run(server)) as srv: self.loop.run_until_complete(client(srv.addr)) + filled = threading.Event() + eof_received = threading.Event() + count = 0 with self.tcp_server(run(eof_server)) as srv: self.loop.run_until_complete(client(srv.addr)) From b5a7b5556c91cf607c568969cb53b4d54c451f27 Mon Sep 17 00:00:00 2001 From: Davide Rizzo Date: Tue, 19 Sep 2023 17:06:12 +0200 Subject: [PATCH 3/6] test_flush_before_shutdown should not depend on sslproto internals --- Lib/test/test_asyncio/test_ssl.py | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/Lib/test/test_asyncio/test_ssl.py b/Lib/test/test_asyncio/test_ssl.py index 5c1dd2d7e35957..902b5fa42aff47 100644 --- a/Lib/test/test_asyncio/test_ssl.py +++ b/Lib/test/test_asyncio/test_ssl.py @@ -1248,14 +1248,16 @@ def test_flush_before_shutdown(self): client_sslctx = self._create_client_ssl_context() future = None + count = 0 + filled = threading.Event() def server(sock): sock.starttls(sslctx, server_side=True) self.assertEqual(sock.recv_all(4), b'ping') sock.send(b'pong') - time.sleep(0.5) # hopefully stuck the TCP buffer - data = sock.recv_all(CHUNK * SIZE) - self.assertEqual(len(data), CHUNK * SIZE) + filled.wait() + data = sock.recv_all(CHUNK * count) + self.assertEqual(len(data), CHUNK * count) sock.close() def run(meth): @@ -1269,30 +1271,38 @@ def wrapper(sock): return wrapper async def client(addr): - nonlocal future + nonlocal future, count future = self.loop.create_future() reader, writer = await asyncio.open_connection( *addr, ssl=client_sslctx, server_hostname='', ssl_handshake_timeout=HANDSHAKE_TIMEOUT) - sslprotocol = writer.transport._ssl_protocol writer.write(b'ping') data = await reader.readexactly(4) self.assertEqual(data, b'pong') - sslprotocol.pause_writing() - for _ in range(SIZE): + await writer.drain() + + # Fill the buffer - with a hack to ensure the protocol is + # paused without actually waiting on drain + count = 0 + while not writer._protocol._paused: writer.write(b'x' * CHUNK) + count += 1 + await asyncio.sleep(0) + # Add more data to ensure we overflow the buffer + writer.write(b'x' * CHUNK) + count += 1 + filled.set() writer.close() - sslprotocol.resume_writing() await self.wait_closed(writer) try: data = await reader.read() self.assertEqual(data, b'') - except ConnectionResetError: + except (ConnectionResetError, BrokenPipeError): pass await future From eef8eda5d2c434adcc6d24579899fac97dd951e7 Mon Sep 17 00:00:00 2001 From: Davide Rizzo Date: Tue, 19 Sep 2023 17:58:39 +0200 Subject: [PATCH 4/6] remove _SendfileFallbackProtocol requirement of _FlowControlMixin --- Lib/asyncio/base_events.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index 9eb34c5f29fb55..52b2f757258680 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -209,12 +209,14 @@ def _check_ssl_socket(sock): class _SendfileFallbackProtocol(protocols.Protocol): def __init__(self, transp): - if not isinstance(transp, transports._FlowControlMixin): - raise TypeError("transport should be _FlowControlMixin instance") self._transport = transp self._proto = transp.get_protocol() self._should_resume_reading = transp.is_reading() - self._should_resume_writing = transp._protocol_paused + + # should we expect a call to resume_writing? + _, high_water = transp.get_write_buffer_limits() + self._should_resume_writing = transp.get_write_buffer_size() > high_water + transp.pause_reading() transp.set_protocol(self) if self._should_resume_writing: From 08eb83c82cc1393e2291a66cb217b256078db68f Mon Sep 17 00:00:00 2001 From: Davide Rizzo Date: Tue, 19 Sep 2023 18:05:03 +0200 Subject: [PATCH 5/6] _SSLProtocolTransport does not inherit _FlowControlMixin --- Lib/asyncio/sslproto.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Lib/asyncio/sslproto.py b/Lib/asyncio/sslproto.py index d7d5f3dc001ecc..425ab29b15ba62 100644 --- a/Lib/asyncio/sslproto.py +++ b/Lib/asyncio/sslproto.py @@ -79,8 +79,7 @@ def add_flowcontrol_defaults(high, low, kb): return hi, lo -class _SSLProtocolTransport(transports._FlowControlMixin, - transports.Transport): +class _SSLProtocolTransport(transports.Transport): _start_tls_compatible = True _sendfile_compatible = constants._SendfileMode.FALLBACK From 8b26ebc56b7e98015edd9a671f52deac0f3dbb1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Langa?= Date: Wed, 21 Oct 2026 03:33:49 +0200 Subject: [PATCH 6/6] Fix CI: news entry, lint fixes, and test updates (gh-109051) - Remove unused imports flagged by ruff (asyncio.base_events, asyncio.sslproto). - Add Misc/NEWS.d entry for gh-109051 (bedevere/news). - test_ssl: rewrite test_remote_shutdown_receives_trailing_data_on_slow_socket to fill the transport buffer instead of the removed _test__append_write_backlog helper, mirroring the sibling test rewrite already in this PR. - test_sslproto: stop mocking sslobj.write() failures; without an outgoing write buffer those errors propagate instead of being buffered and retried. Co-authored-by: Muse Code (Meta Muse Spark) --- Lib/asyncio/base_events.py | 1 - Lib/asyncio/sslproto.py | 1 - Lib/test/test_asyncio/test_ssl.py | 47 ++++++++++++++----- Lib/test/test_asyncio/test_sslproto.py | 9 +++- ...-10-21-01-27-59.gh-issue-109051.q7ZmXk.rst | 4 ++ 5 files changed, 47 insertions(+), 15 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2026-10-21-01-27-59.gh-issue-109051.q7ZmXk.rst diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index 52b2f757258680..1c204e221b34f7 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -48,7 +48,6 @@ from . import staggered from . import tasks from . import timeouts -from . import transports from . import trsock from .log import logger diff --git a/Lib/asyncio/sslproto.py b/Lib/asyncio/sslproto.py index 425ab29b15ba62..8b30aa665ed484 100644 --- a/Lib/asyncio/sslproto.py +++ b/Lib/asyncio/sslproto.py @@ -2,7 +2,6 @@ # SPDX-License-Identifier: PSF-2.0 AND (MIT OR Apache-2.0) # SPDX-FileCopyrightText: Copyright (c) 2015-2021 MagicStack Inc. http://magic.io -import collections import enum import warnings try: diff --git a/Lib/test/test_asyncio/test_ssl.py b/Lib/test/test_asyncio/test_ssl.py index 902b5fa42aff47..0b5682b2348b56 100644 --- a/Lib/test/test_asyncio/test_ssl.py +++ b/Lib/test/test_asyncio/test_ssl.py @@ -1465,7 +1465,6 @@ def test_remote_shutdown_receives_trailing_data_on_slow_socket(self): # The slow path is triggered here by setting SO_SNDBUF, see code and comment below. CHUNK = 1024 * 128 - SIZE = 32 sslctx = self._create_server_ssl_context( test_utils.ONLYCERT, @@ -1503,25 +1502,38 @@ def server(sock): sslobj.write(b'pong') sock.send(outgoing.read()) - time.sleep(0.2) # wait for the peer to fill its backlog + data_len = 0 + filled.wait() + # trigger peer's resume_writing() + incoming.write(sock.recv(65536 * 4)) + while True: + try: + chunk = len(sslobj.read(16384)) + data_len += chunk + except ssl.SSLWantReadError: + break # send close_notify but don't wait for response with self.assertRaises(ssl.SSLWantReadError): sslobj.unwrap() sock.send(outgoing.read()) + eof_received.wait() # should receive all data - data_len = 0 while True: try: chunk = len(sslobj.read(16384)) data_len += chunk except ssl.SSLWantReadError: incoming.write(sock.recv(16384)) + if not incoming.pending: + # EOF received + break except ssl.SSLZeroReturnError: break - self.assertEqual(data_len, CHUNK * SIZE*2) + assert count > 0 + self.assertEqual(data_len, CHUNK * count) # verify that close_notify is received sslobj.unwrap() @@ -1533,19 +1545,20 @@ def eof_server(sock): self.assertEqual(sock.recv_all(4), b'ping') sock.send(b'pong') - time.sleep(0.2) # wait for the peer to fill its backlog + filled.wait() # send EOF sock.shutdown(socket.SHUT_WR) # should receive all data - data = sock.recv_all(CHUNK * SIZE) - self.assertEqual(len(data), CHUNK * SIZE) + assert count > 0 + data = sock.recv_all(CHUNK * count) + self.assertEqual(len(data), CHUNK * count) sock.close() async def client(addr): - nonlocal future + nonlocal future, count future = self.loop.create_future() reader, writer = await asyncio.open_connection( @@ -1557,15 +1570,21 @@ async def client(addr): data = await reader.readexactly(4) self.assertEqual(data, b'pong') - # fill write backlog in a hacky way - renegotiation won't help - for _ in range(SIZE*2): - writer.transport._test__append_write_backlog(b'x' * CHUNK) + count = 0 + # Fill the buffer - with a hack to ensure the protocol is + # paused without actually waiting on drain + while not writer._protocol._paused: + writer.write(b'x' * CHUNK) + count += 1 + await asyncio.sleep(0) + filled.set() try: data = await reader.read() self.assertEqual(data, b'') except (BrokenPipeError, ConnectionResetError): pass + eof_received.set() # Make sure _SelectorSocketTransport enters the delayed write # path in its `write` method by wrapping socket in a fake class @@ -1614,9 +1633,15 @@ def wrapper(sock): self.loop.call_soon_threadsafe(future.set_result, None) return wrapper + filled = threading.Event() + eof_received = threading.Event() + count = 0 with self.tcp_server(run(server)) as srv: self.loop.run_until_complete(client(srv.addr)) + filled = threading.Event() + eof_received = threading.Event() + count = 0 with self.tcp_server(run(eof_server)) as srv: self.loop.run_until_complete(client(srv.addr)) diff --git a/Lib/test/test_asyncio/test_sslproto.py b/Lib/test/test_asyncio/test_sslproto.py index acb80c19c0bdd0..9afc7330df0e05 100644 --- a/Lib/test/test_asyncio/test_sslproto.py +++ b/Lib/test/test_asyncio/test_sslproto.py @@ -47,7 +47,10 @@ def connection_made(self, ssl_proto, *, do_handshake=None): sslobj = mock.Mock() # emulate reading decompressed data sslobj.read.side_effect = ssl.SSLWantReadError - sslobj.write.side_effect = ssl.SSLWantReadError + # NOTE: sslobj.write is left as a plain Mock (writes succeed). + # There is no outgoing write buffer anymore, so an SSLWantReadError + # from write() would propagate to the caller instead of being + # buffered and retried. if do_handshake is not None: sslobj.do_handshake = do_handshake ssl_proto._sslobj = sslobj @@ -169,7 +172,9 @@ def test_connection_lost_when_busy(self): # emulate reading decompressed data sslobj = mock.Mock() sslobj.read.side_effect = ssl.SSLWantReadError - sslobj.write.side_effect = ssl.SSLWantReadError + # NOTE: writes succeed (plain Mock). There is no outgoing write + # buffer anymore, so an SSLWantReadError from write() would + # propagate to the caller instead of being buffered and retried. ssl_proto._sslobj = sslobj # emulate outgoing data diff --git a/Misc/NEWS.d/next/Library/2026-10-21-01-27-59.gh-issue-109051.q7ZmXk.rst b/Misc/NEWS.d/next/Library/2026-10-21-01-27-59.gh-issue-109051.q7ZmXk.rst new file mode 100644 index 00000000000000..6c0f49f5bd50d3 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-10-21-01-27-59.gh-issue-109051.q7ZmXk.rst @@ -0,0 +1,4 @@ +Remove the redundant outgoing write buffer from :mod:`asyncio` +TLS transports. Encrypted data is now written directly to the underlying +transport, and write flow control is delegated to it. This also fixes an +:exc:`AssertionError` when the buffer size was checked inconsistently.