Skip to content

Commit 5710879

Browse files
committed
Replace HTTP/3 stream wait futures with custom wakers
WaitingStreams previously stored Abortable<WaitForStream> values in a FuturesUnordered with per-direction abort maps. Each wait existed only to register a channel waker and return readiness to H3Driver. Channel endpoints now remain in StreamCtx with one shared StreamWake per stream. RecvState owns RecvDelivery in Open and FinPending, pairing the app-bound sender with its capacity-wait flag. mark_fin_pending moves the whole payload, preserving waits. Closed drops both without manual wait cancellation. Capacity polling retains Pending waits and releases its readiness permit before application reads. Closed delivery cannot wait. The outbound receiver remains independent because queued FIN can drop it before transport writes finish. An outbound waiting flag identifies the receiver to re-poll, and stale ready IDs are ignored after cleanup. ReadyStreams encapsulates coalescing: the queue owns Arc<StreamWake> notifications, and dequeue clears their AtomicBool before returning a stream ID. Channel polling no longer participates in queue acknowledgment. Direct polls preserve queued notifications; dequeue rearms wakes even without a subsequent channel poll. Wake metadata owns no stream endpoints. STOP is probed before H3 frame validation, except for queued PeerStreamError. STOP drains ordinary buffered frames, preserves PeerStreamError, and redispatches it to close both directions. Errors sent later through outstanding permits are rejected when the receiver is dropped. App-bound readiness is serviced before outbound data. Ready-queue draining retains Tokio cooperative budgeting. Regressions verify FIN wait preservation, automatic wait removal on closure, stale wakes, cooperative yields, wake coalescing, and deferred cleanup. Remaining lifecycle invariant checks have runtime paths and debug assertions. The cooperative-yield regression bounds its post-yield receive with a deadline. A lost readiness notification fails both channel-direction cases instead of leaving the test suite waiting indefinitely. Peer-reset handling remains idempotent when the body reader and HTTP/3 polling both report a reset. Unit coverage checks repeated notifications after open or FIN-pending delivery and verifies that capacity waits are dropped without closing the response channel. Strict cancellation checks apply only to local reset and STOP operations.
1 parent 3753036 commit 5710879

7 files changed

Lines changed: 1010 additions & 629 deletions

File tree

‎tokio-quiche/src/http3/driver/AGENTS.md‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,7 @@ Async HTTP/3 driver bridging `quiche::h3::Connection` to Tokio tasks via channel
1212
| `hooks.rs` | `DriverHooks` trait (sealed). Defines `headers_received`, `stream_recv_closed`, `stream_closed`, `conn_established`, `conn_command`, `wait_for_action` |
1313
| `client.rs` | `ClientHooks` impl, `ClientH3Driver`/`ClientH3Controller` aliases, `ClientH3Event`/`ClientH3Command` |
1414
| `server.rs` | `ServerHooks` impl, `ServerH3Driver`/`ServerH3Controller` aliases, `ServerH3Event`/`ServerH3Command` |
15-
| `streams.rs` | `StreamCtx`, `FlowCtx`, `WaitForStream` future, capacity/readiness signals |
16-
| `waiting_streams.rs` | `WaitingStreams` keyed tracker for pending upstream-capacity and downstream-data waits |
15+
| `streams.rs` | `StreamCtx`, `FlowCtx`, shared per-stream waker and channel waiting flags |
1716
| `datagram.rs` | DATAGRAM/CONNECT-UDP flow handling |
1817
| `connection.rs` | `H3Conn` wrapper exposing `h3::Connection` operations |
1918
| `test_utils.rs` | `DriverTestHelper<H>` -- wraps `Pipe` + `H3Driver` for unit tests |
@@ -23,18 +22,22 @@ Async HTTP/3 driver bridging `quiche::h3::Connection` to Tokio tasks via channel
2322

2423
| Task | Start at |
2524
|------|----------|
26-
| Channel architecture | `mod.rs:332` (`H3Driver` struct fields: `h3_event_sender`, `cmd_recv`, `stream_map`, `waiting_streams`) |
25+
| Channel architecture | `mod.rs` (`H3Driver` struct fields: `h3_event_sender`, `cmd_recv`, `stream_map`, `ready_streams`) |
2726
| `select!` loop / priority ordering | `mod.rs` `wait_for_data` impl -- uses `biased` select! |
2827
| Stream lifecycle | `cleanup_stream`, `shutdown_stream`, `process_h3_fin`, `process_h3_data` in `mod.rs` |
29-
| Per-stream backpressure | `streams.rs` defines wait futures; `waiting_streams.rs` tracks and cancels them by stream ID |
28+
| Per-stream backpressure | `streams.rs` keeps endpoints in `StreamCtx`; one generic waker queues each stream ID once, and the driver re-polls both waiting directions |
3029
| Adding endpoint-specific behavior | `hooks.rs` -- add method to `DriverHooks`, impl in `client.rs`/`server.rs` |
3130
| Writing tests | `test_utils.rs` for `DriverTestHelper`, `tests.rs` for examples |
3231

3332
## ANTI-PATTERNS
3433

3534
- Store receive and send lifecycle states explicitly; channel absence does not
36-
imply completion, especially when a waiting future owns the endpoint.
37-
Received FIN and app-channel FIN delivery are distinct receive states.
35+
imply completion. Received FIN and app-channel FIN delivery are distinct
36+
receive states. Open and pending receive states own an `InboundDelivery` payload
37+
with its sender and capacity-wait flag; closing the state drops both. FIN
38+
preserves the payload, including an existing wait. The outbound receiver is
39+
independent: queued FIN can drop it
40+
while the transport still has body data to send.
3841
- Do not add overlapping boolean completion methods for these states. A single
3942
derived helper can centralize the cleanup decision without conflating pending
4043
FIN with closed delivery. Match lifecycle enums exhaustively where a new
@@ -45,7 +48,15 @@ Async HTTP/3 driver bridging `quiche::h3::Connection` to Tokio tasks via channel
4548
- **`process_write_frame` uses `Error::Done` as success** -- non-obvious control flow, don't replicate elsewhere.
4649
- **`DriverHooks` is sealed** -- `mod hooks` is `pub(crate)`, trait has `#[allow(private_interfaces)]`. Do not expose.
4750
- **Stream cleanup is distributed** across 4+ functions (`cleanup_stream`, `shutdown_stream`, `process_h3_fin`, `process_h3_data`). Understand all paths before modifying.
48-
- **Waiting-stream cancellation must be keyed** -- cancelling waiting streams must not linearly walk all pending futures for performance reasons.
51+
- **Waiting-stream cancellation must be keyed** -- clear flags and drop endpoints
52+
through `stream_map`; do not linearly scan the ready queue.
53+
- `ReadyStreams` owns queue acknowledgment: dequeueing clears the queued wake
54+
flag before returning the stream ID. Channel polling never changes that flag.
55+
Keep waiting flags set on `Pending`, including spurious wakes and
56+
cooperative-budget yields.
57+
- Stream cleanup drops channel endpoints immediately; stale ready IDs are ignored.
58+
- STOP_SENDING discards ordinary outbound frames, but a buffered `PeerStreamError`
59+
must still shut down both directions.
4960
- **`STREAM_CAPACITY`** is 1 in test/debug, 16 in release. Tests exercise backpressure differently from prod.
5061

5162
## NOTES

‎tokio-quiche/src/http3/driver/client.rs‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -211,8 +211,11 @@ impl ClientHooks {
211211
};
212212

213213
// log::info!("sent h3 request"; "stream_id" => stream_id);
214-
let (mut stream_ctx, send, recv) =
215-
StreamCtx::new(stream_id, STREAM_CAPACITY);
214+
let (mut stream_ctx, send, recv) = StreamCtx::new(
215+
stream_id,
216+
STREAM_CAPACITY,
217+
driver.ready_stream_sender.clone(),
218+
);
216219

217220
if body_finished {
218221
// `send_request()` already sent FIN for bodyless requests, so
@@ -235,9 +238,7 @@ impl ClientHooks {
235238

236239
if let Some(body_writer) = request.body_writer {
237240
let _ = body_writer.send(send.clone());
238-
driver
239-
.waiting_streams
240-
.push(stream_ctx.wait_for_recv(stream_id))?;
241+
stream_ctx.wait_for_recv();
241242
}
242243

243244
driver.insert_stream(stream_id, stream_ctx);

0 commit comments

Comments
 (0)