diff --git a/src/client.rs b/src/client.rs index cb7e677a..b75c1334 100644 --- a/src/client.rs +++ b/src/client.rs @@ -348,6 +348,7 @@ pub struct Builder { /// /// When this gets exhausted, we issue a GOAWAY with `ENHANCE_YOUR_CALM`. data_frame_budget: proto::DataFrameBudget, + data_frame_overhead_threshold: usize, } #[derive(Debug)] @@ -669,6 +670,7 @@ impl Builder { stream_id: 1.into(), local_max_error_reset_streams: Some(proto::DEFAULT_LOCAL_RESET_COUNT_MAX), data_frame_budget: proto::DataFrameBudget::Auto, + data_frame_overhead_threshold: proto::DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD, } } @@ -1174,6 +1176,46 @@ impl Builder { self } + /// Sets the payload length below which a received DATA frame is charged + /// framing overhead against the connection's DATA frame budget. + /// + /// A received DATA frame whose payload is shorter than this threshold + /// consumes `threshold - payload_len` of the budget until the application + /// reads it; a frame at or above the threshold consumes nothing, and one + /// that exceeds it restores budget. Lowering the threshold therefore + /// narrows the range of payload sizes that can be charged at all, without + /// changing how the budget itself is accounted. + /// + /// This is useful for a peer that legitimately receives many small DATA + /// frames — for example a server-sent event stream whose events are tens + /// to low hundreds of bytes each. With the default threshold, such a + /// stream is charged on every frame, and whether it survives depends on + /// both the frame size and the connection window, since the number of + /// frames that can be buffered unread is bounded by the window. Setting + /// the threshold at or below the smallest legitimate frame size removes + /// that dependency entirely: those frames stop being charged, so the + /// stream is unaffected by later changes to the window. + /// + /// Empty DATA frames are limited separately and are not affected by this + /// setting, so lowering the threshold does not change how many empty + /// frames a peer may send. + /// + /// The default is 256 bytes, which is the behaviour when this is not + /// called. + /// + /// # Panics + /// + /// This function panics if `threshold` is 0, which would disable the + /// charge for every payload size. + pub fn data_frame_overhead_threshold(&mut self, threshold: usize) -> &mut Self { + assert!( + threshold > 0, + "data_frame_overhead_threshold must be greater than 0" + ); + self.data_frame_overhead_threshold = threshold; + self + } + /// Sets the first stream ID to something other than 1. #[cfg(feature = "unstable")] pub fn initial_stream_id(&mut self, stream_id: u32) -> &mut Self { @@ -1367,6 +1409,7 @@ where data_frame_budget: builder .data_frame_budget .resolve(builder.initial_target_connection_window_size), + data_frame_overhead_threshold: builder.data_frame_overhead_threshold, }, ); let send_request = SendRequest { diff --git a/src/proto/connection.rs b/src/proto/connection.rs index 757e2b27..5e571608 100644 --- a/src/proto/connection.rs +++ b/src/proto/connection.rs @@ -84,6 +84,7 @@ pub(crate) struct Config { pub local_error_reset_streams_max: Option, pub settings: frame::Settings, pub data_frame_budget: usize, + pub data_frame_overhead_threshold: usize, } #[derive(Clone, Copy, Debug)] @@ -145,6 +146,7 @@ where .map(|max| max as usize), local_max_error_reset_streams: config.local_error_reset_streams_max, data_frame_budget: config.data_frame_budget, + data_frame_overhead_threshold: config.data_frame_overhead_threshold, } } let streams = Streams::new(streams_config(&config)); diff --git a/src/proto/streams/counts.rs b/src/proto/streams/counts.rs index 1cad3dfd..16b279da 100644 --- a/src/proto/streams/counts.rs +++ b/src/proto/streams/counts.rs @@ -70,6 +70,10 @@ pub(super) struct Counts { /// connection-level budget for DATA framing overhead. data_frame_budget: Budget, + /// payload length below which a received DATA frame is charged framing + /// overhead against `data_frame_budget`. + data_frame_overhead_threshold: usize, + /// Number of empty, non-final DATA frames received over the lifetime of /// the connection. num_recv_empty_data_frames: usize, @@ -91,6 +95,7 @@ impl Counts { max_local_error_reset_streams: config.local_max_error_reset_streams, num_local_error_reset_streams: 0, data_frame_budget: Budget::new(config.data_frame_budget), + data_frame_overhead_threshold: config.data_frame_overhead_threshold, num_recv_empty_data_frames: 0, } } @@ -106,12 +111,12 @@ impl Counts { return Err(BudgetExhausted); } Ok(()) - } else if payload_len < DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD { + } else if payload_len < self.data_frame_overhead_threshold { self.data_frame_budget - .consume(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD - payload_len) + .consume(self.data_frame_overhead_threshold - payload_len) } else { self.data_frame_budget - .replenish(payload_len - DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD); + .replenish(payload_len - self.data_frame_overhead_threshold); Ok(()) } } @@ -119,9 +124,9 @@ impl Counts { /// Releases the framing overhead of a DATA frame that is no longer /// buffered internally. pub fn release_data_frame(&mut self, payload_len: usize) { - if payload_len != 0 && payload_len < DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD { + if payload_len != 0 && payload_len < self.data_frame_overhead_threshold { self.data_frame_budget - .replenish(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD - payload_len); + .replenish(self.data_frame_overhead_threshold - payload_len); } } @@ -356,6 +361,10 @@ mod tests { use crate::frame::DEFAULT_INITIAL_WINDOW_SIZE; fn counts() -> Counts { + counts_with_threshold(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD) + } + + fn counts_with_threshold(threshold: usize) -> Counts { Counts::new( peer::Dyn::Server, &Config { @@ -370,7 +379,8 @@ mod tests { remote_init_window_sz: DEFAULT_INITIAL_WINDOW_SIZE, remote_max_initiated: None, local_max_error_reset_streams: None, - data_frame_budget: DEFAULT_DATA_FRAME_BUDGET, + data_frame_budget: threshold * 100, + data_frame_overhead_threshold: threshold, }, ) } @@ -441,4 +451,69 @@ mod tests { } assert!(counts.record_data_frame(0).is_err()); } + + #[test] + fn a_lowered_threshold_stops_charging_frames_at_or_above_it() { + let threshold = 16; + let mut counts = counts_with_threshold(threshold); + + // At or above the threshold nothing is charged, so an unlimited number + // of such frames may stay buffered unread. This is the point of making + // the threshold configurable: a peer that knows the smallest payload it + // legitimately receives can put the threshold at or below it. + for _ in 0..1_000_000 { + counts.record_data_frame(threshold).unwrap(); + } + assert_eq!(counts.data_frame_budget.available, threshold * 100); + } + + #[test] + fn a_lowered_threshold_still_charges_frames_below_it() { + let threshold = 16; + let mut counts = counts_with_threshold(threshold); + + // The other direction: shortening the charged range must not turn the + // charge off for the payload sizes that remain inside it. + let mut sent = 0; + while counts.record_data_frame(1).is_ok() { + sent += 1; + assert!( + sent < 10_000, + "budget never ran out for sub-threshold frames" + ); + } + assert_eq!(sent, (threshold * 100) / (threshold - 1)); + } + + #[test] + fn the_threshold_does_not_change_the_empty_data_frame_limit() { + // Empty frames are limited by count, not by budget, so the number a + // peer may send must be identical for every threshold. Without this, + // lowering the threshold to admit small legitimate frames could be + // read as weakening the empty-frame limit. + for threshold in [1, 16, DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD, 1024] { + let mut counts = counts_with_threshold(threshold); + for _ in 0..MAX_RECV_EMPTY_DATA_FRAMES { + counts.record_data_frame(0).unwrap(); + } + assert!( + counts.record_data_frame(0).is_err(), + "threshold {threshold} changed the empty DATA frame limit" + ); + } + } + + #[test] + fn released_frames_replenish_at_the_configured_threshold() { + let threshold = 16; + let mut counts = counts_with_threshold(threshold); + + // Record and release must use the same threshold, or the budget drifts + // in one direction over the life of the connection. + for _ in 0..1_000_000 { + counts.record_data_frame(1).unwrap(); + counts.release_data_frame(1); + } + assert_eq!(counts.data_frame_budget.available, threshold * 100); + } } diff --git a/src/proto/streams/mod.rs b/src/proto/streams/mod.rs index c0845507..77e88100 100644 --- a/src/proto/streams/mod.rs +++ b/src/proto/streams/mod.rs @@ -83,6 +83,12 @@ pub struct Config { /// /// Default 25600 bytes pub data_frame_budget: usize, + + /// payload length (in bytes) below which a received DATA frame is charged + /// framing overhead against `data_frame_budget`. + /// + /// Default 256 bytes + pub data_frame_overhead_threshold: usize, } trait DebugStructExt<'a, 'b> { diff --git a/src/proto/streams/recv.rs b/src/proto/streams/recv.rs index c56b368f..afe8790d 100644 --- a/src/proto/streams/recv.rs +++ b/src/proto/streams/recv.rs @@ -1319,6 +1319,7 @@ mod tests { remote_max_initiated: None, local_max_error_reset_streams: None, data_frame_budget: DEFAULT_DATA_FRAME_BUDGET, + data_frame_overhead_threshold: DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD, }; let mut recv = Recv::new(peer::Dyn::Server, &config); let mut store = Store::new(); diff --git a/src/server.rs b/src/server.rs index 316d6a18..448e24ce 100644 --- a/src/server.rs +++ b/src/server.rs @@ -263,6 +263,7 @@ pub struct Builder { /// /// When this gets exhausted, we issue a GOAWAY with `ENHANCE_YOUR_CALM`. data_frame_budget: proto::DataFrameBudget, + data_frame_overhead_threshold: usize, } /// Send a response back to the client @@ -662,6 +663,7 @@ impl Builder { max_send_buffer_size: proto::DEFAULT_MAX_SEND_BUFFER_SIZE, local_max_error_reset_streams: Some(proto::DEFAULT_LOCAL_RESET_COUNT_MAX), data_frame_budget: proto::DataFrameBudget::Auto, + data_frame_overhead_threshold: proto::DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD, } } @@ -1068,6 +1070,46 @@ impl Builder { self } + /// Sets the payload length below which a received DATA frame is charged + /// framing overhead against the connection's DATA frame budget. + /// + /// A received DATA frame whose payload is shorter than this threshold + /// consumes `threshold - payload_len` of the budget until the application + /// reads it; a frame at or above the threshold consumes nothing, and one + /// that exceeds it restores budget. Lowering the threshold therefore + /// narrows the range of payload sizes that can be charged at all, without + /// changing how the budget itself is accounted. + /// + /// This is useful for a peer that legitimately receives many small DATA + /// frames — for example a server-sent event stream whose events are tens + /// to low hundreds of bytes each. With the default threshold, such a + /// stream is charged on every frame, and whether it survives depends on + /// both the frame size and the connection window, since the number of + /// frames that can be buffered unread is bounded by the window. Setting + /// the threshold at or below the smallest legitimate frame size removes + /// that dependency entirely: those frames stop being charged, so the + /// stream is unaffected by later changes to the window. + /// + /// Empty DATA frames are limited separately and are not affected by this + /// setting, so lowering the threshold does not change how many empty + /// frames a peer may send. + /// + /// The default is 256 bytes, which is the behaviour when this is not + /// called. + /// + /// # Panics + /// + /// This function panics if `threshold` is 0, which would disable the + /// charge for every payload size. + pub fn data_frame_overhead_threshold(&mut self, threshold: usize) -> &mut Self { + assert!( + threshold > 0, + "data_frame_overhead_threshold must be greater than 0" + ); + self.data_frame_overhead_threshold = threshold; + self + } + /// Creates a new configured HTTP/2 server backed by `io`. /// /// It is expected that `io` already be in an appropriate state to commence @@ -1537,6 +1579,9 @@ where .builder .data_frame_budget .resolve(self.builder.initial_target_connection_window_size), + data_frame_overhead_threshold: self + .builder + .data_frame_overhead_threshold, }, );