Skip to content

Commit b9ffdcb

Browse files
committed
WebTransport: add anticipatedConcurrentIncomingUni/BidirectionalStreams support
The spec lets an application supply, at session construction, how many incoming streams of each type it expects the peer to open concurrently, so the transport can grant that much headroom up front rather than only after the peer bumps into the default limit and waits a round trip for MAX_STREAMS. webtransport_set_anticipated_incoming_{uni,bidi}_streams stores the per-session value and recomputes the connection-wide total across every not-yet-closed WebTransport session, clamped to MAX_ANTICIPATED_INCOMING_STREAMS so an application cannot ratchet the advertised limit arbitrarily high (it is monotonic and never comes back down as sessions close). The unidirectional total also adds HTTP3_UNI_CONTROL_STREAMS, since HTTP/3's own control and QPACK streams draw from the same limit. Closed sessions are excluded from the sum; sessions still negotiating are included, since that is exactly when the API supplies this value. Client-only: the anticipated-streams attributes are on the WebTransport interface, which has no server-side equivalent.
1 parent c7a0cd2 commit b9ffdcb

6 files changed

Lines changed: 600 additions & 1 deletion

File tree

‎neqo-http3/src/connection.rs‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,19 @@ use crate::{
5151
stream_type_reader::NewStreamHeadReader,
5252
};
5353

54+
/// Number of unidirectional streams HTTP/3 opens for its own use (control, QPACK encoder, QPACK
55+
/// decoder). Streams of reserved types (RFC 9114, Section 6.2.3), which a peer may also open,
56+
/// are deliberately not counted: how many a given peer opens is a guess.
57+
pub const HTTP3_UNI_CONTROL_STREAMS: u64 = 3;
58+
59+
/// Upper bound on the connection-wide incoming stream limit derived from anticipated
60+
/// `WebTransport` streams.
61+
///
62+
/// Without a cap, an application-controlled value summed across an unbounded number of sessions
63+
/// could ratchet the advertised `MAX_STREAMS` arbitrarily high; the limit is monotonic, so it
64+
/// never comes back down as sessions close.
65+
pub const MAX_ANTICIPATED_INCOMING_STREAMS: u64 = 10_000;
66+
5467
pub struct RequestDescription<'b, T: RequestTarget> {
5568
pub method: &'b str,
5669
pub connect_type: Option<ConnectType>,
@@ -1437,6 +1450,52 @@ impl Http3Connection {
14371450
Ok(stats)
14381451
}
14391452

1453+
/// Recompute the connection's incoming limit for `stream_type` as the sum over open
1454+
/// `WebTransport` sessions; see
1455+
/// [`crate::webtransport::ClientSession::webtransport_set_anticipated_incoming_uni_streams`].
1456+
/// Walks every receive stream: call per session, not per stream.
1457+
///
1458+
/// # Errors
1459+
/// Returns `InvalidStreamId` if the session does not exist, is not a `WebTransport`
1460+
/// session, or is already closing.
1461+
pub(crate) fn webtransport_set_anticipated_incoming_streams(
1462+
&self,
1463+
conn: &mut Connection,
1464+
session_id: StreamId,
1465+
stream_type: StreamType,
1466+
value: u16,
1467+
) -> Res<()> {
1468+
let session = self.webtransport_session(session_id)?;
1469+
if session.borrow().is_closing() {
1470+
return Err(Error::InvalidStreamId);
1471+
}
1472+
session
1473+
.borrow_mut()
1474+
.set_anticipated_incoming(stream_type, value);
1475+
1476+
let total: u64 = self
1477+
.recv_streams
1478+
.values()
1479+
.filter_map(|s| {
1480+
let session = s.extended_connect_session()?;
1481+
let session = session.borrow();
1482+
(!session.is_closing()
1483+
&& session.connect_type() == ExtendedConnectType::WebTransport)
1484+
.then(|| u64::from(session.anticipated_incoming(stream_type)))
1485+
})
1486+
.sum::<u64>()
1487+
.min(MAX_ANTICIPATED_INCOMING_STREAMS);
1488+
1489+
let total_including_h3 = match stream_type {
1490+
// HTTP/3 opens its own unidirectional streams (control and the two QPACK
1491+
// streams) out of the same limit, so they have to be added on top.
1492+
StreamType::UniDi => total + HTTP3_UNI_CONTROL_STREAMS,
1493+
StreamType::BiDi => total,
1494+
};
1495+
conn.set_remote_max_streams(stream_type, total_including_h3);
1496+
Ok(())
1497+
}
1498+
14401499
pub(crate) fn extended_connect_close_session(
14411500
&mut self,
14421501
conn: &mut Connection,

‎neqo-http3/src/features/extended_connect/session.rs‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ use std::{
1616
use neqo_common::{Bytes, Encoder, Header, MessageType, Role, qdebug, qtrace, to_u64};
1717
use neqo_transport::{
1818
AppError, Connection, DatagramQueueCapacity, DatagramQueueOutcome, DatagramTracking, StreamId,
19+
StreamType,
1920
streams::{SendGroupId, SendOrder},
2021
};
2122
use rustc_hash::FxHashSet as HashSet;
@@ -68,6 +69,8 @@ pub(crate) struct Session {
6869
/// Set when a datagram capsule was refused with `FlowControlLimit`, so a
6970
/// resume event fires once the control stream is writable again.
7071
datagram_capsule_blocked: bool,
72+
anticipated_incoming_uni: u16,
73+
anticipated_incoming_bidi: u16,
7174
}
7275

7376
#[derive(Debug, PartialEq, Clone, Copy)]
@@ -128,6 +131,8 @@ impl Session {
128131
protocol,
129132
draining: false,
130133
datagram_capsule_blocked: false,
134+
anticipated_incoming_uni: 0,
135+
anticipated_incoming_bidi: 0,
131136
}
132137
}
133138

@@ -159,6 +164,8 @@ impl Session {
159164
protocol,
160165
draining: false,
161166
datagram_capsule_blocked: false,
167+
anticipated_incoming_uni: 0,
168+
anticipated_incoming_bidi: 0,
162169
})
163170
}
164171

@@ -172,6 +179,21 @@ impl Session {
172179
self.id
173180
}
174181

182+
pub(crate) const fn set_anticipated_incoming(&mut self, stream_type: StreamType, value: u16) {
183+
match stream_type {
184+
StreamType::UniDi => self.anticipated_incoming_uni = value,
185+
StreamType::BiDi => self.anticipated_incoming_bidi = value,
186+
}
187+
}
188+
189+
#[must_use]
190+
pub(crate) const fn anticipated_incoming(&self, stream_type: StreamType) -> u16 {
191+
match stream_type {
192+
StreamType::UniDi => self.anticipated_incoming_uni,
193+
StreamType::BiDi => self.anticipated_incoming_bidi,
194+
}
195+
}
196+
175197
/// Mark session as draining. Returns `true` if this was the first call
176198
/// (i.e. the session was not already draining).
177199
pub(crate) const fn set_draining(&mut self) -> bool {
@@ -361,6 +383,11 @@ impl Session {
361383
matches!(self.state, State::Active)
362384
}
363385

386+
#[must_use]
387+
pub(crate) const fn is_closing(&self) -> bool {
388+
self.state.closing_state()
389+
}
390+
364391
pub(crate) fn take_sub_streams(&mut self) -> (HashSet<StreamId>, HashSet<StreamId>) {
365392
self.protocol.take_sub_streams()
366393
}

0 commit comments

Comments
 (0)