Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion neqo-http3/src/connect_udp.rs
Original file line number Diff line number Diff line change
Expand Up @@ -250,7 +250,8 @@ impl Handler for Http3Connection {
error,
message,
now,
)
)?;
Ok(())
}

fn connect_udp_send_datagram<I: Into<DatagramTracking>>(
Expand Down
31 changes: 12 additions & 19 deletions neqo-http3/src/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1447,13 +1447,9 @@ impl Http3Connection {
error: u32,
message: &str,
now: Instant,
) -> Res<()> {
if self
.get_extended_connect_session(session_id)?
.borrow()
.connect_type()
!= connect_type
{
) -> Res<Rc<RefCell<extended_connect::session::Session>>> {
let session = self.get_extended_connect_session(session_id)?;
if session.borrow().connect_type() != connect_type {
return Err(Error::InvalidStreamId);
}
let send_stream = self
Expand All @@ -1468,7 +1464,7 @@ impl Http3Connection {
} else if send_stream.has_data_to_send() {
self.streams_with_pending_data.insert(session_id);
}
Ok(())
Ok(session)
}

fn get_extended_connect_session(
Expand Down Expand Up @@ -1683,21 +1679,18 @@ impl Http3Connection {
.datagram_queue_capacity(conn))
}

/// Expire stale outgoing datagrams on every active extended-CONNECT
/// session's queue and count them per session. Called once per
/// `process_http3` tick. The transport counts expiries on the queue
/// itself, whether its own timer sweep or this one sheds them, so this
/// sweep picks up the right number regardless of whether it runs before
/// or after `Connection::process_output` within a tick.
///
/// Returns the total number of datagrams expired, for the caller to fold
/// into a stats counter.
/// Record expired, sent and too-big datagram counts for each active
/// extended-CONNECT session. Returns the number expired.
pub(crate) fn expire_datagram_queues(&self, conn: &mut Connection, now: Instant) -> u64 {
self.recv_streams
.values()
.filter_map(|s| s.extended_connect_session())
.filter(|s| s.borrow().is_active())
.map(|s| s.borrow_mut().expire_datagrams(conn, now))
.map(|s| {
let mut s = s.borrow_mut();
s.report_sent_datagrams(conn);
Comment thread
jesup marked this conversation as resolved.
s.expire_datagrams(conn, now)
})
.sum()
}

Expand Down Expand Up @@ -1884,7 +1877,7 @@ impl Http3Connection {
wt: &Rc<RefCell<extended_connect::session::Session>>,
conn: &mut Connection,
) {
wt.borrow().drop_queued_datagrams(conn);
wt.borrow_mut().drop_queued_datagrams(conn);
Comment thread
jesup marked this conversation as resolved.

let (recv, send) = wt.borrow_mut().take_sub_streams();

Expand Down
46 changes: 39 additions & 7 deletions neqo-http3/src/features/extended_connect/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ use std::{
time::{Duration, Instant},
};

use neqo_common::{Bytes, Encoder, Header, MessageType, Role, qdebug, qtrace};
use neqo_common::{Bytes, Encoder, Header, MessageType, Role, qdebug, qtrace, to_u64};
#[cfg(test)]
use neqo_transport::DatagramQueueCapacity;
use neqo_transport::{
Expand Down Expand Up @@ -477,8 +477,8 @@ impl Session {
self.datagram_capsule_blocked = true;
}
res?;
// This path never touches the queue, so it carries no
// backpressure signal.
// Bypasses the queue, so count it sent now.
self.protocol.record_sent_outgoing_datagrams(1);
return Ok(DatagramQueueOutcome::Ok);
}

Expand Down Expand Up @@ -515,9 +515,31 @@ impl Session {
send_order,
)?;
qtrace!("[{self}] enqueued datagram: {outcome:?}");
match &outcome {
DatagramQueueOutcome::Ok | DatagramQueueOutcome::MaxBufferedReached => {}
DatagramQueueOutcome::Rejected => self.report_dropped(1),
DatagramQueueOutcome::Overflowed { dropped } => self.report_dropped(*dropped),
}
Ok(outcome)
}

fn report_dropped(&mut self, count: usize) {
self.protocol
.record_dropped_outgoing_datagrams(to_u64(count));
}

/// Drop this session's queue, recording its remaining datagrams and counts.
pub(crate) fn drop_queued_datagrams(&mut self, conn: &mut Connection) {
let dropped = conn.drop_session_datagrams(self.id);
self.protocol.record_sent_outgoing_datagrams(dropped.sent);
self.protocol
.record_expired_outgoing_datagrams(dropped.expired);
self.protocol
.record_dropped_outgoing_datagrams(dropped.too_big);
self.protocol
.record_dropped_outgoing_datagrams(dropped.queued);
}

/// Set the outgoing-datagram queue's `outgoingMaxBufferedDatagrams`, or
/// clear it back to no limit with `None`. Ignored once closing: the
/// queue is already dropped, and setting it would recreate it.
Expand Down Expand Up @@ -572,10 +594,12 @@ impl Session {
expired
}

/// Drop this session's queue, which lives on `conn` and would otherwise
/// outlive the session.
pub(crate) fn drop_queued_datagrams(&self, conn: &mut Connection) {
_ = conn.drop_session_datagrams(self.id);
/// Record the sent and too-big counts since the last call.
pub(crate) fn report_sent_datagrams(&mut self, conn: &mut Connection) {
let sent = conn.take_session_sent_datagrams(self.id);
self.protocol.record_sent_outgoing_datagrams(sent);
let too_big = conn.take_session_too_big_datagrams(self.id);
self.protocol.record_dropped_outgoing_datagrams(too_big);
}

pub(crate) fn datagram(&self, datagram: Bytes) {
Expand Down Expand Up @@ -782,10 +806,18 @@ pub(crate) trait Protocol: Debug + Display {
None
}

/// Record that `count` outgoing datagrams were sent.
/// A no-op default for protocols that don't track [`SessionStats`].
fn record_sent_outgoing_datagrams(&mut self, _count: u64) {}

/// Record that `count` outgoing datagrams expired before being sent.
/// A no-op default for protocols that don't track [`SessionStats`].
fn record_expired_outgoing_datagrams(&mut self, _count: u64) {}

/// Record that `count` outgoing datagrams were dropped unsent.
/// A no-op default for protocols that don't track [`SessionStats`].
fn record_dropped_outgoing_datagrams(&mut self, _count: u64) {}

fn protocol(&self) -> Option<&str> {
None
}
Expand Down
26 changes: 21 additions & 5 deletions neqo-http3/src/features/extended_connect/stats.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,9 @@

//! Per-session statistics exposed to the WebTransport API consumer.
//!
//! Limited to the members of [`WebTransportDatagramStats`], which describe
//! per-session datagram queue behaviour and so cannot be derived from the
//! connection-level counters in [`neqo_transport::Stats`]. Everything else
//! Per-session datagram queue counters, which cannot be derived from
//! [`neqo_transport::Stats`]. Only [`OutgoingDatagramStats::expired`] is a
//! [`WebTransportDatagramStats`] member (`expiredOutgoing`). Everything else
//! `getStats()` reports is scoped to the underlying connection, and is only
//! exposed at all when that connection is dedicated to a single session.
//!
Expand All @@ -24,6 +24,22 @@
)]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SessionStats {
/// Outgoing datagrams that expired (per `outgoingMaxAge`) before being sent.
pub datagrams_expired_outgoing: u64,
pub outgoing_datagrams: OutgoingDatagramStats,
}

/// What became of the datagrams handed to a session's outgoing queue.
#[expect(
clippy::module_name_repetitions,
reason = "stats::OutgoingDatagramStats is clearer than stats::OutgoingDatagram"
)]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct OutgoingDatagramStats {
/// Written into a packet, or to the control stream for the capsule
/// fallback. Not a delivery guarantee.
pub sent: u64,
/// Expired (per `outgoingMaxAge`) before being sent.
pub expired: u64,
/// Discarded unsent and unexpired: evicted or refused by the queue, too
/// big for the path MTU, or still queued at close.
pub dropped: u64,
}
Loading
Loading