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 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 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):