From 2bce46d50ef68f9de55cc0ca2cdc4d826396f79d Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Tue, 4 Aug 2026 14:14:24 +0530 Subject: [PATCH 1/3] Count the bytes a peer has not read yet A stream transport now charges every byte it writes and is credited when the receiving end hands those bytes to its protocol, so what is on the wire, held by a partition or parked behind a paused reader all weigh on the writer. Crossing the high mark pauses the protocol and falling back to the low one resumes it, both synchronously, the way the standard library's own transports do it. None of it applies until net.set_flow_control() arms it. Libraries set write-buffer limits uninvited, so arming on their call would change runs nobody touched; unarmed, nothing is charged and the reported buffer size stays zero. --- src/simloop/_net.py | 38 ++++++++- src/simloop/_transports.py | 157 +++++++++++++++++++++++++++++++++---- 2 files changed, 178 insertions(+), 17 deletions(-) diff --git a/src/simloop/_net.py b/src/simloop/_net.py index cca06fc..37bdde2 100644 --- a/src/simloop/_net.py +++ b/src/simloop/_net.py @@ -18,7 +18,11 @@ import asyncio -from simloop._transports import _SimDatagramTransport, _SimStreamTransport +from simloop._transports import ( + _check_limits, + _SimDatagramTransport, + _SimStreamTransport, +) if TYPE_CHECKING: from simloop._loop import SimLoop @@ -408,6 +412,8 @@ def __init__(self, loop: SimLoop) -> None: self._default_latency: tuple[float, float] = (0.0, 0.0) self._default_drop = 0.0 self._default_duplicate = 0.0 + self._flow_control = False + self._flow_defaults: tuple[int, int] = (16 * 1024, 64 * 1024) self._links: dict[tuple[str, str], _Link] = {} self._cuts: set[frozenset[str]] = set() self._held: list[_Packet] = [] @@ -598,6 +604,36 @@ def set_disk( ) self.host(name).disk._configure(buffered=buffered, torn=torn) + def set_flow_control( + self, *, enabled: bool = True, high: int | None = None, low: int | None = None + ) -> None: + """Make stream writes push back when the peer is not keeping up. + + Armed, a transport's write buffer counts every byte it has written + that the peer's protocol has not received — still on the wire, held + by a partition, or parked because the peer paused reading. Crossing + ``high`` calls ``pause_writing`` on the protocol, so ``drain()`` + really waits; dropping back to ``low`` calls ``resume_writing``. Both + happen at once, without a scheduling step of their own. Watermarks + default to the standard library's 64 KiB and 16 KiB and can be set + here for the whole network or per transport with + ``set_write_buffer_limits``. + + Off unless asked for, and a transport's own limits are recorded but + inert until then, because libraries set them uninvited: a run that + never calls this decides exactly what it decided without it. + Turning it off releases whatever is paused. + + The buffer drains on the peer *application's* read, with no + read-ahead, so this pushes back sooner than a real socket does. + """ + if high is not None or low is not None: + self._flow_defaults = _check_limits(high, low) + self._flow_control = enabled + if not enabled: + for transport in list(self._streams.values()): + transport._release_flow_control() + def clock_offset(self, name: str) -> float: self._require_host(name) return self._clock_offsets.get(name, 0.0) diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py index e6cd418..90dc657 100644 --- a/src/simloop/_transports.py +++ b/src/simloop/_transports.py @@ -26,6 +26,23 @@ def _check_bytes(data: object) -> bytes: return bytes(data) +def _check_limits(high: int | None, low: int | None) -> tuple[int, int]: + """Derive a (low, high) watermark pair the way the stdlib derives it. + + The rules and the error text match + ``asyncio.transports._FlowControlMixin._set_write_buffer_limits``, so a + protocol that already knows what its own transport accepts learns nothing + new here. Shared with the network default so both are checked once. + """ + if high is None: + high = 64 * 1024 if low is None else 4 * low + if low is None: + low = high // 4 + if not high >= low >= 0: + raise ValueError(f"high ({high!r}) must be >= low ({low!r}) must be >= 0") + return (low, high) + + class _SimDatagramTransport(asyncio.DatagramTransport): def __init__(self, net: SimNetwork, local: _Addr, remote: _Addr | None) -> None: super().__init__() @@ -168,9 +185,14 @@ class _SimStreamTransport(asyncio.Transport): Reliability comes from per-direction sequence numbers dispatched in order by the network, not from retransmission: stream packets are never - dropped, only delayed or held. Flow control is not simulated — writes - leave immediately, so the reported write-buffer size is always zero and - the peer can never pause this side. + dropped, only delayed or held. + + The write buffer holds every byte written that the peer's protocol has + not received yet — in flight, held by a partition, waiting on an earlier + sequence number, or parked because the peer paused reading. Bytes are + charged in ``write`` and credited when the receiving end hands them up, + and crossing a watermark pauses or resumes the protocol synchronously. + None of it applies until ``net.set_flow_control()`` arms it. """ def __init__( @@ -189,7 +211,11 @@ def __init__( self._read_paused = False self._backlog: list[bytes] = [] self._eof_pending = False - self._limits = (16 * 1024, 64 * 1024) # (low, high): recorded, inert + self._write_buffer = 0 + self._protocol_paused = False + # None means this transport never set its own, so the network default + # applies — including a change to it made after this transport existed. + self._limits: tuple[int, int] | None = None self._extra_socket: _SimSocket | None = None self._peer_closed = False # the peer's FIN or RST has arrived @@ -224,6 +250,7 @@ def write(self, data: Any) -> None: raise RuntimeError("Cannot write to closing transport") if payload: self._send("data", payload) + self._charge(len(payload)) def writelines(self, list_of_data: Any) -> None: for data in list_of_data: @@ -259,6 +286,12 @@ def _finish(self, exc: Exception | None) -> None: return self._closed = True self._closing = True + # Nothing can be owed on a connection that no longer exists. The + # protocol is not resumed here: connection_lost is what wakes a writer + # waiting in drain(), and resume_writing on a torn-down protocol would + # be a second wakeup the stdlib never sends. + self._write_buffer = 0 + self._protocol_paused = False self._net._drop_stream(self._conn, self._local[0], self._local[1]) if self._extra_socket is not None: self._extra_socket._dispose() @@ -266,6 +299,94 @@ def _finish(self, exc: Exception | None) -> None: if protocol is not None: protocol.connection_lost(exc) + # ------------------------------------------------------------------ + # Write flow control (bytes the peer's protocol has not received yet) + # ------------------------------------------------------------------ + + def _effective_limits(self) -> tuple[int, int]: + if self._limits is not None: + return self._limits + return self._net._flow_defaults + + def _charge(self, count: int) -> None: + if not self._net._flow_control: + return + self._write_buffer += count + self._maybe_pause_protocol() + + def _credit(self, count: int) -> None: + if not self._net._flow_control or self._closed: + return + # Arming mid-run cannot un-send what is already on the wire, so a + # credit for bytes that were never charged stops at zero. + self._write_buffer = max(0, self._write_buffer - count) + self._maybe_resume_protocol() + + def _consumed(self, count: int) -> None: + """Release the sender of bytes this end has just handed to its protocol.""" + if not self._net._flow_control: + return + # The remote endpoint's own key: on a self-connection the two ends + # share a host, and the port is what addresses the one that wrote. + peer = self._net._streams.get( + (self._conn, self._remote[0], self._remote[1]) + ) + if peer is not None: + peer._credit(count) + + def _maybe_pause_protocol(self) -> None: + _low, high = self._effective_limits() + if self._write_buffer <= high or self._protocol_paused: + return + if self._protocol is None: + return + self._protocol_paused = True + try: + self._protocol.pause_writing() + except (SystemExit, KeyboardInterrupt): + raise + except BaseException as exc: + self._report_failure("protocol.pause_writing() failed", exc) + + def _maybe_resume_protocol(self) -> None: + low, _high = self._effective_limits() + if not self._protocol_paused or self._write_buffer > low: + return + self._protocol_paused = False + if self._protocol is None: + return + try: + self._protocol.resume_writing() + except (SystemExit, KeyboardInterrupt): + raise + except BaseException as exc: + self._report_failure("protocol.resume_writing() failed", exc) + + def _release_flow_control(self) -> None: + """Let go of a paused writer because the switch just went off.""" + self._write_buffer = 0 + if not self._protocol_paused: + return + self._protocol_paused = False + if self._protocol is None: + return + try: + self._protocol.resume_writing() + except (SystemExit, KeyboardInterrupt): + raise + except BaseException as exc: + self._report_failure("protocol.resume_writing() failed", exc) + + def _report_failure(self, message: str, exc: BaseException) -> None: + self._net._loop.call_exception_handler( + { + "message": message, + "exception": exc, + "transport": self, + "protocol": self._protocol, + } + ) + # ------------------------------------------------------------------ # Inbound (called by the network, already in seq order) # ------------------------------------------------------------------ @@ -277,6 +398,9 @@ def _data_arrived(self, data: bytes) -> None: self._backlog.append(data) return self._protocol.data_received(data) + # Credited after the call, not before: the sender is released only once + # the receiving protocol has finished with the bytes. + self._consumed(len(data)) def _eof_arrived(self) -> None: self._peer_closed = True @@ -315,7 +439,11 @@ def resume_reading(self) -> None: return self._read_paused = False while self._backlog and not self._read_paused and not self._closed: - self._protocol.data_received(self._backlog.pop(0)) + chunk = self._backlog.pop(0) + self._protocol.data_received(chunk) + # Per chunk, so a protocol that pauses again mid-drain leaves the + # rest of the backlog charged to the sender. + self._consumed(len(chunk)) if self._eof_pending and not self._read_paused and not self._closed: self._eof_pending = False self._eof_arrived() @@ -355,18 +483,15 @@ def get_protocol(self) -> Any: def set_write_buffer_limits( self, high: int | None = None, low: int | None = None ) -> None: - if high is None: - high = 64 * 1024 if low is None else 4 * low - if low is None: - low = high // 4 - if not high >= low >= 0: - raise ValueError( - f"high ({high!r}) must be >= low ({low!r}) must be >= 0" - ) - self._limits = (low, high) + self._limits = _check_limits(high, low) + # Only the pause side, matching the stdlib: lowering the marks under a + # full buffer pauses at once, raising them waits for a read to resume. + self._maybe_pause_protocol() def get_write_buffer_limits(self) -> tuple[int, int]: - return self._limits + return self._effective_limits() def get_write_buffer_size(self) -> int: - return 0 + # With flow control off nothing is charged and writes leave + # immediately, so nothing is buffered to report. + return self._write_buffer if self._net._flow_control else 0 From 8bac53db91584eed6d675c273fd7bce25110c773 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Tue, 4 Aug 2026 14:14:34 +0530 Subject: [PATCH 2/3] Hold the unarmed runs to their recorded digests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reference network workload never asks for flow control, so its two digests for seeds 0, 1 and 2 are pinned from before the feature existed: a user's recorded seed has to keep replaying. The rest covers the armed side — watermark crossings, a drain that waits for the peer's read, teardown and fault edges, and both deadlock shapes. The searched one answers after a short read window rather than a fixed byte count, so how much of the body the server took before its own response filled its buffer follows the latency draws, and seed 2 is where that leaves both ends waiting on each other. --- tests/test_flow_control.py | 830 ++++++++++++++++++++++++++++++++++++ tests/test_net_hardening.py | 26 ++ 2 files changed, 856 insertions(+) create mode 100644 tests/test_flow_control.py diff --git a/tests/test_flow_control.py b/tests/test_flow_control.py new file mode 100644 index 0000000..d2c3d0d --- /dev/null +++ b/tests/test_flow_control.py @@ -0,0 +1,830 @@ +"""Simulated write-side flow control: accounting, watermarks, and the switch.""" + +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest + +from simloop import SimLoop, SimulationDeadlockError +from simloop._explore import explore +from simloop._run import finish + + +def _network(seed: int = 0) -> SimLoop: + loop = SimLoop(seed=seed) + loop.net.host("server") + loop.net.host("client") + return loop + + +class _Sink(asyncio.Protocol): + """Server end that records what it was given and can refuse to read.""" + + pause_on_connect = False + + def __init__(self) -> None: + self.transport: Any = None + self.data = bytearray() + self.eof = False + self.lost: list[BaseException | None] = [] + + def connection_made(self, transport: Any) -> None: + self.transport = transport + if self.pause_on_connect: + transport.pause_reading() + + def data_received(self, data: bytes) -> None: + self.data += data + + def eof_received(self) -> bool: + self.eof = True + return True + + def connection_lost(self, exc: BaseException | None) -> None: + self.lost.append(exc) + + +class _HeldSink(_Sink): + pause_on_connect = True + + +class _WatchedStreams(asyncio.StreamReaderProtocol): + """A stream protocol that notes the teardown its transport delivered.""" + + def __init__( + self, reader: asyncio.StreamReader, lost: list[Exception | None] + ) -> None: + super().__init__(reader) + self._lost = lost + + def connection_lost(self, exc: Exception | None) -> None: + self._lost.append(exc) + super().connection_lost(exc) + + +class _Writer(asyncio.Protocol): + """Client end that notes the flow-control callbacks it was sent.""" + + def __init__(self) -> None: + self.transport: Any = None + self.events: list[str] = [] + self.data = bytearray() + + def connection_made(self, transport: Any) -> None: + self.transport = transport + + def data_received(self, data: bytes) -> None: + self.data += data + + def pause_writing(self) -> None: + self.events.append("pause") + + def resume_writing(self) -> None: + self.events.append("resume") + + +async def _serve(factory: Any, port: int = 9000) -> None: + async def start() -> None: + await asyncio.get_running_loop().create_server(factory, "0.0.0.0", port) + + loop = asyncio.get_running_loop() + assert isinstance(loop, SimLoop) + await loop.net.host("server").create_task(start()) + + +async def _pair(port: int = 9000) -> tuple[_HeldSink, Any, _Writer]: + """A server that will not read and a client protocol pinned to its own host.""" + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory, port) + loop = asyncio.get_running_loop() + assert isinstance(loop, SimLoop) + + async def connect() -> tuple[Any, _Writer]: + transport, protocol = await asyncio.get_running_loop().create_connection( + _Writer, "server", port + ) + return transport, protocol + + transport, client = await loop.net.host("client").create_task(connect()) + return sinks[0], transport, client + + +def test_a_run_that_asks_for_nothing_reports_no_write_buffer() -> None: + loop = _network() + + async def main() -> tuple[int, list[str]]: + sink, transport, client = await _pair() + for _ in range(64): + transport.write(b"x" * 1024) + size = transport.get_write_buffer_size() + await asyncio.sleep(0.01) + return size, client.events + + try: + size, events = loop.run_until_complete(main()) + finally: + loop.close() + assert size == 0 + assert events == [] + + +def test_watermark_crossing_pauses_and_resumes_exactly_once() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> list[str]: + sink, transport, client = await _pair() + for _ in range(5): + transport.write(b"x" * 1024) + assert client.events == ["pause"] + await asyncio.sleep(0.01) + sink.transport.resume_reading() + return client.events + + try: + events = loop.run_until_complete(main()) + finally: + loop.close() + assert events == ["pause", "resume"] + + +def test_the_buffer_counts_bytes_the_peer_has_not_received() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[int, int, int]: + sink, transport, client = await _pair() + for _ in range(5): + transport.write(b"x" * 1024) + written = transport.get_write_buffer_size() + await asyncio.sleep(0.01) + in_flight = transport.get_write_buffer_size() # still owed: peer is paused + sink.transport.resume_reading() + return written, in_flight, transport.get_write_buffer_size() + + try: + written, in_flight, drained = loop.run_until_complete(main()) + finally: + loop.close() + assert written == 5 * 1024 + assert in_flight == 5 * 1024 + assert drained == 0 + + +def test_drain_blocks_until_the_peer_reads() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[float, float]: + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory) + running = asyncio.get_running_loop() + + async def request() -> float: + _reader, writer = await asyncio.open_connection("server", 9000) + writer.write(b"x" * 5120) + await writer.drain() + return running.time() + + task = loop.net.host("client").create_task(request()) + await asyncio.sleep(0.01) + running.call_later(1.0, sinks[0].transport.resume_reading) + expected = running.time() + 1.0 + return await task, expected + + try: + released, expected = loop.run_until_complete(main()) + finally: + loop.close() + # The drain returns in the very step the peer's read credited it back. + assert released == pytest.approx(expected) + + +def test_writes_are_allowed_while_paused_and_grow_the_buffer() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[int, list[str]]: + sink, transport, client = await _pair() + for _ in range(5): + transport.write(b"x" * 1024) + assert client.events == ["pause"] + for _ in range(5): + transport.write(b"x" * 1024) + return transport.get_write_buffer_size(), client.events + + try: + size, events = loop.run_until_complete(main()) + finally: + loop.close() + assert size == 10 * 1024 + assert events == ["pause"] + + +def test_writelines_crosses_the_watermark_once() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[int, list[str]]: + sink, transport, client = await _pair() + transport.writelines([b"x" * 1024] * 5) + return transport.get_write_buffer_size(), client.events + + try: + size, events = loop.run_until_complete(main()) + finally: + loop.close() + assert size == 5 * 1024 + assert events == ["pause"] + + +def test_limits_set_on_a_transport_override_the_network_default() -> None: + loop = _network() + loop.net.set_flow_control() + + async def main() -> tuple[tuple[int, int], tuple[int, int], list[str]]: + sink, transport, client = await _pair() + network_default = transport.get_write_buffer_limits() + transport.set_write_buffer_limits(high=2048, low=512) + own = transport.get_write_buffer_limits() + for _ in range(5): + transport.write(b"x" * 1024) + return network_default, own, client.events + + try: + network_default, own, events = loop.run_until_complete(main()) + finally: + loop.close() + assert network_default == (16 * 1024, 64 * 1024) + assert own == (512, 2048) + # 5 KiB is under the network default and over this transport's own mark. + assert events == ["pause"] + + +def test_lowering_the_limits_pauses_immediately() -> None: + loop = _network() + loop.net.set_flow_control() + + async def main() -> tuple[list[str], list[str]]: + sink, transport, client = await _pair() + for _ in range(5): + transport.write(b"x" * 1024) + under_default = list(client.events) + transport.set_write_buffer_limits(high=1024, low=256) + return under_default, client.events + + try: + under_default, events = loop.run_until_complete(main()) + finally: + loop.close() + assert under_default == [] + assert events == ["pause"] + + +def test_bad_limits_are_rejected() -> None: + loop = _network() + + async def main() -> None: + sink, transport, client = await _pair() + with pytest.raises(ValueError): + transport.set_write_buffer_limits(high=1, low=2) + + try: + with pytest.raises(ValueError): + loop.net.set_flow_control(high=1, low=2) + loop.run_until_complete(main()) + finally: + loop.close() + + +def test_turning_flow_control_off_releases_a_paused_writer() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[float, float, list[str]]: + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory) + running = asyncio.get_running_loop() + seen: list[str] = [] + + async def request() -> float: + _reader, writer = await asyncio.open_connection("server", 9000) + writer.write(b"x" * 5120) + seen.append("writing") + await writer.drain() + seen.append("drained") + assert writer.transport.get_write_buffer_size() == 0 + return running.time() + + task = loop.net.host("client").create_task(request()) + await asyncio.sleep(0.01) + assert seen == ["writing"] + running.call_later(1.0, lambda: loop.net.set_flow_control(enabled=False)) + expected = running.time() + 1.0 + return await task, expected, seen + + try: + released, expected, seen = loop.run_until_complete(main()) + finally: + loop.close() + assert released == pytest.approx(expected) + assert seen == ["writing", "drained"] + + +def test_datagram_transports_have_no_write_buffer() -> None: + # Write-side flow control is a stream feature: a datagram endpoint queues + # nothing on anyone's behalf, so it does not answer for a write buffer. + loop = _network() + + async def main() -> Any: + transport, _protocol = await asyncio.get_running_loop().create_datagram_endpoint( + asyncio.DatagramProtocol, local_addr=("0.0.0.0", 7000) + ) + return transport + + try: + transport = loop.run_until_complete(main()) + finally: + loop.close() + assert not hasattr(transport, "get_write_buffer_size") + assert not hasattr(transport, "set_write_buffer_limits") + + +def test_write_eof_while_paused_still_reaches_the_peer() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[list[str], int, bool]: + sink, transport, client = await _pair() + for _ in range(5): + transport.write(b"x" * 1024) + transport.write_eof() + assert client.events == ["pause"] + await asyncio.sleep(0.01) + assert not sink.eof + sink.transport.resume_reading() + return client.events, len(sink.data), sink.eof + + try: + events, received, eof = loop.run_until_complete(main()) + finally: + loop.close() + assert events == ["pause", "resume"] + assert received == 5 * 1024 + assert eof is True + + +def test_close_while_a_drain_waits_wakes_it() -> None: + # FlowControlMixin.connection_lost(None) resolves every waiter, because the + # protocol is paused when the close lands. + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> Any: + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory) + + async def request() -> Any: + _reader, writer = await asyncio.open_connection("server", 9000) + writer.write(b"x" * 5120) + drain = asyncio.ensure_future(writer.drain()) + await asyncio.sleep(0.01) + assert not drain.done() + writer.transport.close() + return await drain + + return await loop.net.host("client").create_task(request()) + + try: + outcome = loop.run_until_complete(main()) + finally: + loop.close() + assert outcome is None + + +def test_drain_started_after_close_raises_connection_reset() -> None: + # The other stdlib branch: drain() on a transport whose connection_lost has + # already run reports the loss instead of waiting. Which branch a drain + # started in the same step as close() takes is a scheduling decision, so + # this waits for the teardown to land rather than racing it. + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> None: + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory) + + async def request() -> None: + _reader, writer = await asyncio.open_connection("server", 9000) + writer.write(b"x" * 5120) + writer.transport.close() + await asyncio.sleep(0.01) + with pytest.raises(ConnectionResetError): + await writer.drain() + + await loop.net.host("client").create_task(request()) + + try: + loop.run_until_complete(main()) + finally: + loop.close() + + +def test_a_peer_reset_fails_a_pending_drain() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> None: + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory) + + async def request() -> None: + _reader, writer = await asyncio.open_connection("server", 9000) + writer.write(b"x" * 5120) + with pytest.raises(ConnectionResetError): + await writer.drain() + + task = loop.net.host("client").create_task(request()) + await asyncio.sleep(0.01) + sinks[0].transport.abort() + await task + + try: + loop.run_until_complete(main()) + finally: + loop.close() + + +def test_a_partition_applies_backpressure_and_a_heal_lifts_it() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[float, int]: + sinks: list[_Sink] = [] + + def server_factory() -> _Sink: + sink = _Sink() + sinks.append(sink) + return sink + + await _serve(server_factory) + running = asyncio.get_running_loop() + + async def request() -> float: + _reader, writer = await asyncio.open_connection("server", 9000) + loop.net.partition({"server"}, {"client"}) + writer.write(b"x" * 5120) + with pytest.raises(TimeoutError): + async with asyncio.timeout(0.5): + await writer.drain() + running.call_later(0.5, loop.net.heal) + await writer.drain() + return running.time() + + released = await loop.net.host("client").create_task(request()) + return released, len(sinks[0].data) + + try: + released, received = loop.run_until_complete(main()) + finally: + loop.close() + assert released == pytest.approx(1.0) + assert received == 5120 + + +def test_a_crashed_peer_leaves_the_writer_paused_until_its_own_timeout() -> None: + # A crashed host sends no reset, so the survivor's transport is untouched + # by design and only its own clock ever tells it anything. + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> None: + sinks: list[_Sink] = [] + + def server_factory() -> _Sink: + sink = _Sink() + sinks.append(sink) + return sink + + await _serve(server_factory) + + async def request() -> None: + _reader, writer = await asyncio.open_connection("server", 9000) + writer.write(b"x" * 5120) + loop.net.crash("server") + with pytest.raises(TimeoutError): + async with asyncio.timeout(1.0): + await writer.drain() + + await loop.net.host("client").create_task(request()) + + try: + loop.run_until_complete(main()) + finally: + loop.close() + + +def test_a_crashing_writer_releases_its_own_drain() -> None: + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> tuple[bool, list[Exception | None]]: + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory) + lost: list[Exception | None] = [] + + async def request() -> None: + # open_connection's own assembly, so the stream protocol can be + # watched: everything else about the connection is unchanged. + running = asyncio.get_running_loop() + reader = asyncio.StreamReader(loop=running) + protocol = _WatchedStreams(reader, lost) + transport, _ = await running.create_connection( + lambda: protocol, "server", 9000 + ) + writer = asyncio.StreamWriter(transport, protocol, reader, running) + writer.write(b"x" * 5120) + await writer.drain() + + task = loop.net.host("client").create_task(request()) + await asyncio.sleep(0.01) + assert not task.done() + loop.net.crash("client") + await asyncio.sleep(0.01) + return task.done(), lost + + try: + done, lost = loop.run_until_complete(main()) + finally: + loop.close() + assert done is True + assert lost == [None] + + +def _crossing(seed: int, armed: bool) -> str: + """A run whose writer really crosses a watermark, traced end to end.""" + loop = SimLoop(seed=seed) + loop.net.host("server") + loop.net.host("client") + loop.net.set_defaults(latency=(0.001, 0.02)) + if armed: + loop.net.set_flow_control(high=2048, low=512) + + async def main() -> None: + sinks: list[_HeldSink] = [] + + def server_factory() -> _HeldSink: + sink = _HeldSink() + sinks.append(sink) + return sink + + await _serve(server_factory) + running = asyncio.get_running_loop() + + async def request() -> None: + _reader, writer = await asyncio.open_connection("server", 9000) + for _ in range(4): + writer.write(b"x" * 4096) + await writer.drain() + + task = loop.net.host("client").create_task(request()) + for _ in range(8): + await asyncio.sleep(0.05) + if sinks[0].transport.is_reading(): + sinks[0].transport.pause_reading() + else: + sinks[0].transport.resume_reading() + await task + assert running.time() > 0.0 + + try: + loop.run_until_complete(main()) + return loop.trace_hash() + finally: + loop.close() + + +def test_the_same_seed_traces_identically_under_flow_control() -> None: + assert _crossing(3, armed=True) == _crossing(3, armed=True) + # A crossing is visible in the trace: the writer's wakeup is scheduled by + # the read that released it, so an armed run cannot hash like an idle one. + assert _crossing(3, armed=True) != _crossing(3, armed=False) + + +def _small_exchange(armed: bool) -> str: + loop = SimLoop(seed=1) + loop.net.host("server") + loop.net.host("client") + if armed: + loop.net.set_flow_control() + + async def main() -> None: + sinks: list[_Sink] = [] + + def server_factory() -> _Sink: + sink = _Sink() + sinks.append(sink) + return sink + + await _serve(server_factory) + + async def request() -> None: + _reader, writer = await asyncio.open_connection("server", 9000) + writer.write(b"hello") + await writer.drain() + await asyncio.sleep(0.1) + writer.close() + + await loop.net.host("client").create_task(request()) + await asyncio.sleep(0.01) + assert bytes(sinks[0].data) == b"hello" + + try: + loop.run_until_complete(main()) + return loop.trace_hash() + finally: + loop.close() + + +def test_a_workload_that_never_crosses_a_watermark_traces_the_same_either_way() -> None: + assert _small_exchange(False) == _small_exchange(True) + + +# Both ends answer before they consume, so neither ever credits the other. +# Sizes are chosen against the stream readers' own 1 KiB buffers: the first +# few chunks are taken and credited, the reader then stops reading, and what +# is left unread is more than the 2 KiB high mark. +_HEADER = b"h" * 32 + b"\n" +_CHUNK = b"x" * 1024 +_CHUNKS = 8 +_READER_LIMIT = 1024 + + +def test_a_mutual_drain_is_a_deadlock() -> None: + # No timer is pending while both ends sit in drain(), so the run can never + # make progress — the backpressure form of the missing-timeout bug. The + # stall leaves the server's handler task pending, so teardown goes through + # finish(), which cancels it instead of letting its collection print. + loop = _network() + loop.net.set_flow_control(high=2048, low=512) + + async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await reader.readline() + for _ in range(_CHUNKS): + writer.write(_CHUNK) + await writer.drain() + await reader.readexactly(_CHUNKS * len(_CHUNK)) + + async def main() -> None: + async def start() -> None: + await asyncio.start_server(handle, "0.0.0.0", 9000, limit=_READER_LIMIT) + + await loop.net.host("server").create_task(start()) + reader, writer = await asyncio.open_connection( + "server", 9000, limit=_READER_LIMIT + ) + writer.write(_HEADER) + for _ in range(_CHUNKS): + writer.write(_CHUNK) + await writer.drain() + await reader.readexactly(_CHUNKS * len(_CHUNK)) + + try: + with pytest.raises(SimulationDeadlockError): + loop.run_until_complete(main()) + finally: + finish(loop) + + +# The same shape as the deadlock above, but the server answers after a short +# read window instead of after a fixed number of bytes, and every run is +# bounded so it terminates either way. How much of the body the server took +# before its own response filled its buffer then depends on the latency draws: +# take enough and the credit lifts the client's pause, take too little and +# both ends sit in drain() with neither reading. The sizes sit near the +# watermark on purpose — far above it every seed hangs, far below it none do, +# and neither teaches anything about the seed that matters. +_BODY_CHUNK = 1536 +_BODY_CHUNKS = 8 +_STREAM_LIMIT = 1024 +_READ_WINDOW = 0.012 +_RUN_BUDGET = 5.0 +# Observed by running the search below; seeds 0 and 1 finish, seed 2 does not. +_FOUND_SEED = 2 + + +async def _answer_before_reading() -> None: + loop = asyncio.get_running_loop() + assert isinstance(loop, SimLoop) + net = loop.net + net.host("server") + net.host("client") + net.set_defaults(latency=(0.001, 0.02)) + net.set_flow_control(high=2048, low=512) + + total = _BODY_CHUNK * _BODY_CHUNKS + header = b"h" * 32 + b"\n" + body = b"x" * _BODY_CHUNK + + async def read_up_to(reader: asyncio.StreamReader, want: int) -> int: + # Incremental, so the reader's own buffer limit is never what stalls a + # run: anything waiting here is waiting on the write side. + got = 0 + while got < want: + piece = await reader.read(_BODY_CHUNK) + if not piece: + break + got += len(piece) + return got + + async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await reader.readline() + # Counted as it goes: what the window cut short still has to be + # subtracted from what is left to read after the response. + taken = [0] + try: + async with asyncio.timeout(_READ_WINDOW): + while taken[0] < total: + piece = await reader.read(_BODY_CHUNK) + if not piece: + break + taken[0] += len(piece) + except TimeoutError: + pass + consumed = taken[0] + for _ in range(_BODY_CHUNKS): + writer.write(body) + await writer.drain() + await read_up_to(reader, total - consumed) + + async def start() -> None: + await asyncio.start_server( + handle, "0.0.0.0", 9000, limit=_STREAM_LIMIT + ) + + await net.host("server").create_task(start()) + + async def request() -> None: + reader, writer = await asyncio.open_connection( + "server", 9000, limit=_STREAM_LIMIT + ) + writer.write(header) + for _ in range(_BODY_CHUNKS): + writer.write(body) + await writer.drain() + await read_up_to(reader, total) + + async with asyncio.timeout(_RUN_BUDGET): + await net.host("client").create_task(request()) + + +def test_a_seed_finds_the_backpressure_deadlock() -> None: + report = explore(_answer_before_reading, range(32)) + assert report is not None + assert report.seed == _FOUND_SEED + assert report.seeds_passed == _FOUND_SEED + assert isinstance(report.exception, TimeoutError) diff --git a/tests/test_net_hardening.py b/tests/test_net_hardening.py index 4d504d2..9b4a301 100644 --- a/tests/test_net_hardening.py +++ b/tests/test_net_hardening.py @@ -39,6 +39,32 @@ def _run_child(seed: int, hashseed: str | None) -> str: return result.stdout.strip() +# Captured before write-side flow control existed. The reference workload +# never arms it, and a run that does not ask for it has to decide exactly what +# it decided before — including every fault draw, so a recorded seed a user +# already has still replays. +_RECORDED = { + 0: ( + "d9b64b9f0908ec4ccd605340c35bfe0a140fbdb1c4e5a70cf4e0bf2631b7fd4d " + "1354346b843e86108cfbf6729db26e95458f1fbae99964c85952f04ed4e44bf8" + ), + 1: ( + "e5c891d7618f7a15cb571c6dc567f2b2505dfdfb70e8bed9dc99978a28827f1b " + "67f19e5ed9bf9cc797540e040d43b0c9c92848deb2ba226dc7866a7e4052da4e" + ), + 2: ( + "6784369a11bfa7dc6f998ff3b606a0b56b0d1d3d55005acabdea44c6e7980b9e " + "6a9362b769406c2ddc58bb5dcadac8ddf0b5e86da72fee78edf9e15f67adb748" + ), +} + + +def test_flow_control_off_reproduces_the_recorded_runs() -> None: + workload = _load_workload() + for seed, recorded in _RECORDED.items(): + assert workload.run(seed) == recorded, f"seed {seed} no longer replays" + + def test_same_seed_replays_identically() -> None: workload = _load_workload() for seed in range(3): From d4dff92ef096a0db0d9188a31e8af8317741bcdd Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Tue, 4 Aug 2026 14:14:44 +0530 Subject: [PATCH 3/3] Say when a write pushes back and what it costs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flow control moves off the cut list and into the network table, with the model stated plainly: what the buffer holds, that pause and resume are synchronous and add no scheduling event of their own, and that the switch is what keeps library-set limits inert. The divergence is stated rather than glossed. The buffer drains when the peer's application receives the bytes, with no read-ahead, so simulated backpressure is tighter than the real thing — which is the point, and is why the standard library's numbers ship inside an opt-in mode. --- CHANGELOG.md | 27 +++++++++++++++++++++++++++ README.md | 6 +++++- docs/design.md | 25 +++++++++++++++++++++++-- docs/supported-api.md | 14 ++++++++------ 4 files changed, 63 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2b4c192..965767f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,33 @@ ## 0.2.0 (unreleased) +- `drain()` can finally block. `loop.net.set_flow_control()` gives stream + transports a write buffer that holds every byte written but not yet + received by the peer's protocol — still in flight, held by a partition, + queued behind an earlier sequence number, or parked because the peer called + `pause_reading()`. A slow reader, a cut link and a dead peer therefore all + push back, and a writer paused against a crashed peer keeps waiting until + its own timeout fires, because a crashed host sends no reset and a real + sender would wait too. Crossing the high mark calls `pause_writing()` and + falling back to the low one calls `resume_writing()`, both synchronously, + which is what turns a backpressure deadlock or an unhandled pause into + something a seed can find. Off unless you ask for it, and the watermarks + once armed are the standard library's own `(low=16 KiB, high=64 KiB)`, + overridable network-wide or per transport with `set_write_buffer_limits`. + The switch is what makes the defaults safe: `set_write_buffer_limits` on + its own records numbers without enforcing them, because libraries call it + uninvited — anyio sets limits on every stream, websockets on every + connection — and arming on their call would change, or deadlock, workloads + nobody touched. On hashes: the feature adds no packets and no scheduling + events of its own, so a run that never arms it is byte-identical to the run + it was before, checked against digests pinned from before the feature + existed; an armed run that actually crosses a mark hashes anew, because the + writer it wakes is a real scheduling decision. One honest divergence from + TCP: the buffer drains when the peer's *application* receives the bytes, + with no read-ahead, so simulated backpressure is strictly tighter than the + real thing — deliberately, since that is what makes a slow consumer + visibly slow. With the switch off `get_write_buffer_size()` reports `0`, + which is the truth: nothing is charged and writes leave immediately. - Traces now say *where*, and that changes every hash. Each scheduling event carries the host it belongs to — the machine that asked for a callback on `schedule`, the machine that owns it on `run` and `cancel`, and nothing at diff --git a/README.md b/README.md index c61b1ee..76fe4fb 100644 --- a/README.md +++ b/README.md @@ -296,7 +296,11 @@ simulated, so a client that connects a socket and hands it to Name resolution stays inside the simulation: `getaddrinfo` resolves sim host names to stable synthetic addresses and raises `socket.gaierror` for anything else — no real DNS, ever. -Write-side flow control is not simulated. +Write-side flow control is simulated on request: `net.set_flow_control()` +makes `drain()` really wait while the peer has not read, so backpressure +deadlocks and missing pause/resume handling become findable. It is off by +default, so a run that does not ask for it decides exactly what it decided +before. The full contract is in [docs/supported-api.md](https://github.com/dhruvl/simloop/blob/main/docs/supported-api.md). What that contract costs real libraries — what aiohttp, anyio, websockets and httpx actually do under simulation, measured rather than promised — is in diff --git a/docs/design.md b/docs/design.md index 8d2a473..9f78761 100644 --- a/docs/design.md +++ b/docs/design.md @@ -222,6 +222,28 @@ Decisions inside that model, each doing real work: application can actually defend against — write, sync, and only then act on it. Skipping that sync is a bug the simulation can now find, which is the whole reason the model exists. +- **A write buffer counts what the peer has not read.** A stream transport + owes every byte it wrote that the peer's protocol has not received — on the + wire, held by a partition, or parked behind the peer's `pause_reading()`. + That definition is the only one under which a slow reader, a cut link and a + dead peer all push back, and it needs no wire traffic: the simulation is one + process, so the receiving transport can credit the sending one directly. The + alternative — a real receive window with credit packets — was rejected on + price: every credit would be another packet, another uid, another latency + draw, moving the trace hash of every stream workload in the repository and + invalidating recorded seeds users already hold, all to buy fidelity a + single-process simulation does not need. Pause and resume fire + synchronously, from the write and from the peer's read, as the stdlib's own + transports do, so the feature adds no scheduling event of its own; the only + new event is the woken writer's, which is a real scheduling decision. It is + off unless asked for, for the same reason the disk is: a run that never + calls `set_flow_control` has to decide exactly what it decided before, and + libraries set write-buffer limits uninvited, so arming on their call would + change — or deadlock — workloads nobody touched. The honest divergence is + that the buffer drains on the peer *application's* read with no read-ahead, + making simulated backpressure strictly tighter than real backpressure. That + is deliberate: it is what makes a slow consumer visibly slow, and it is why + the stdlib's numbers ship inside an opt-in mode rather than by default. - **The accept is sequence 0.** The server builds its transport and sends `accept` before its protocol's `connection_made` can write; the client transport is built when the accept is *dispatched*, not when the connector @@ -234,8 +256,7 @@ Decisions inside that model, each doing real work: `datagram_received` belongs to the receiving machine, and `crash` knows exactly which tasks to kill. -What was cut, deliberately: write-side flow control (`drain()` never blocks -— buffers are unbounded), retransmission and congestion modeling, IP +What was cut, deliberately: retransmission and congestion modeling, IP addresses, TLS. Each would deepen the simulation without widening the class of bugs it can catch; the supported-subset contract beats chasing 100% of the asyncio surface. diff --git a/docs/supported-api.md b/docs/supported-api.md index 2d8f56a..b385088 100644 --- a/docs/supported-api.md +++ b/docs/supported-api.md @@ -46,6 +46,7 @@ a host belong to an implicit `driver` host. | `host.disk` | Storage that survives the crash: a `MutableMapping` per host, where state a real process would fsync belongs. By default a write is durable the moment it is made. Values are stored as given, so mutating a stored object afterwards is the caller's own aliasing, exactly as with a cache in front of a real disk. `disk.sync()` exists on every disk and does nothing on one that does not buffer, so the code under test is written the same way either way | | `loop.net.set_disk` | Makes a host's disk lie about when a write lands. `buffered=True` queues writes and deletes in order and only makes them durable on `sync()`; reads on that host see the queue merged over what is durable, in a fixed order — durable keys where the durable state has them, then the keys the queue invented, in write order, with a queued delete hiding a key. A crash throws the queue away and the reboot finds what was synced; with `torn=True` a seeded prefix of the queue survives instead, which is the state a machine that lost power part-way through a batch comes back with. A prefix is the whole model: writes never land out of order, and no value is ever half-written. Tearing needs a buffer to tear (`torn=True` alone is a `ValueError`), and reconfiguring a disk flushes whatever it was holding. The prefix is drawn from a seed-derived stream of its own, so a torn run makes exactly the network draws it would have made untorn, and a run that never calls `set_disk` draws nothing and records nothing — storage is not a scheduling event and has no trace events at all | | `loop.net.set_clock` / `clock_offset` | Per-host clock skew, in seconds. The offset changes what that host's tasks *read*: `loop.time()` (and `sim.time()` with it) returns true time plus the offset, and a deadline handed to `call_at` is interpreted on the calling task's clock. Durations are immune — `asyncio.sleep`, `asyncio.timeout`, `wait_for` and `call_later` cost the same everywhere, which is exactly what a wrong wall clock does to a real machine. By default the driver and unconfigured hosts read true time; the driver can be given an offset too. Trace timestamps stay on the true clock, so skew never perturbs scheduling and traces from skewed runs stay comparable | +| `loop.net.set_flow_control` / `transport.set_write_buffer_limits` | Makes writes push back when the peer is not keeping up. A stream transport's write buffer holds every byte it has written that the peer's protocol has not received yet — still in flight, held by a partition, waiting behind an earlier sequence number, or parked because the peer called `pause_reading()`. So a slow reader, a cut link and a dead peer all apply backpressure. Crossing `high` calls `pause_writing()` on the protocol and dropping back to `low` calls `resume_writing()`; both happen synchronously, from the write and from the peer's read, so `drain()` really waits and no scheduling event of its own is added. Marks default to the standard library's `(low=16 KiB, high=64 KiB)`, settable network-wide here and per transport with `set_write_buffer_limits(high, low)`, which derives and validates them exactly as the stdlib does (a `high` below `low` is a `ValueError`). **Off until `loop.net.set_flow_control()` says otherwise**: a transport's own `set_write_buffer_limits` records numbers without enforcing them until then, because libraries set them uninvited — anyio sets limits on every stream, websockets on every connection — and upgrading should not deadlock a workload nobody changed. With the switch off `get_write_buffer_size()` reports `0`, which is honest: nothing is charged and writes leave immediately. Honest divergence from TCP: the buffer drains when the peer's *application* receives the bytes, with no read-ahead, so simulated backpressure is strictly tighter than real backpressure — which is what makes a slow consumer visibly slow. A crashed peer sends no reset, so a writer paused against one stays paused until its own timeout fires, exactly as a real sender does. Datagram endpoints have no write buffer at all | | `transport.abort()` | Peer gets `connection_lost(ConnectionResetError)` | | `loop.getaddrinfo` | Resolves against the host table, never DNS: a registered host name, its synthetic address, or a loopback-shaped name (`None`, `""`, `localhost`, `127.0.0.1`, `0.0.0.0`) meaning the calling task's own host. Returns stdlib-shaped rows — `(AF_INET, SOCK_STREAM, IPPROTO_TCP, "", (address, port))` and the `SOCK_DGRAM` / `IPPROTO_UDP` row — filtered by `family`, `type` and `proto`. Ports are numeric (`int`, a digit string, or `None` for 0); resolver `flags` have nothing to vary | | `loop.getnameinfo` | Reverse lookup: a synthetic address maps back to its host name, and a host name (what `get_extra_info("peername")` reports) maps to itself. `NI_NUMERICHOST` returns the address instead; services are always numeric | @@ -71,11 +72,10 @@ by construction — in `examples/jobqueue/` only the broker reads a clock, so skewing a worker changes nothing the cluster decides. Clock faults reach only code that compares timestamps taken on different machines. -Limitations, stated honestly: write-side flow control is not simulated -(`drain()` never blocks, write buffers are unbounded, the peer cannot pause -your writes); there is no retransmission or congestion model — streams are -reliable by construction; and addressing is IPv4-only and entirely synthetic -— there are no routes, no netmasks, and no service-name database. +Limitations, stated honestly: there is no retransmission or congestion model +— streams are reliable by construction; and addressing is IPv4-only and +entirely synthetic — there are no routes, no netmasks, and no service-name +database. ## Fenced @@ -121,7 +121,9 @@ host)`: A `schedule` event names the host that *asked* for the callback, while `run` and `cancel` name the host the callback belongs to. The difference is the point: a wakeup that crosses machines is a `schedule` on one host and a `run` -on another. An empty host means the event belongs to the simulation rather +on another. Flow control is that shape exactly: it adds no packets and no +scheduling events of its own, and what does appear is the wakeup of a writer +that was waiting in `drain()`, scheduled by the host whose read released it. An empty host means the event belongs to the simulation rather than to any machine — a clock advance, which is global; the network's own delivery step, which happens on the wire between two machines rather than on either of them; and every `net` event, whose label already says which