From b4bad87c2b09d6846ea1daed835b0a979364a980 Mon Sep 17 00:00:00 2001 From: zhc Date: Sat, 29 Aug 2026 02:08:33 +0000 Subject: [PATCH 1/2] refactor(model): fold rwnd_remaining into set_rcv_buf A rwnd step had two mutually exclusive action fields, and which one it used decided how the receiver behaved -- but neither name said so. rwnd_remaining only reproduced the window it named if the application drained continuously, so the right edge slid with rcv_nxt; app_read_bytes meant the opposite, a fixed budget after which the window decayed from what was left unread. The reading behaviour was the whole difference, and it was carried by convention in the replayer rather than by the format. set_rcv_buf now carries that meaning: it sizes the buffer and states that the application keeps up with it, so the window is held there. That is what rwnd_remaining said, minus the second way of saying it. The two fields become orthogonal rather than alternatives, so every combination is meaningful -- a buffer, a read, both (resize then read), or neither (carry the previous step forward). With no invariant left to enforce, the hand-written Deserialize and its both-set error go away and Serialize/Deserialize derive. --- src/lib.rs | 72 ++++++----- src/model/rwnd.rs | 320 +++++++++++++++++----------------------------- 2 files changed, 154 insertions(+), 238 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 3775c28..7ef2924 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -238,51 +238,55 @@ pub trait DuplicateTrace: Send { fn next_duplicate(&mut self) -> Option<(DuplicatePattern, Duration)>; } -/// The action a rwnd trace instructs the receiver to take at a single step. -/// -/// At most one action is present per step; a step that only reconfigures the -/// receive buffer (`set_rcv_buf`) without any read or observed-remaining update -/// leaves [`RwndDecision::action`] as `None`. -/// -/// - `AppRead` drives the receiver model by simulating the application reading -/// `bytes` from the receive buffer; the resulting rwnd is computed from the -/// buffer state. -/// - `Remaining` skips the simulation and directly enforces an observed rwnd -/// of `rwnd` bytes — useful for replaying captured traces where only the -/// advertised window is known. -#[derive(Debug, Clone, PartialEq)] -pub enum RwndAction { - /// The simulated application reads this many bytes from the receive buffer at this step. - AppRead { bytes: u64 }, - /// The remaining rwnd value observed immediately after the app consumes data at this step. - Remaining { rwnd: u64 }, -} - /// A single receive-side decision emitted by a [`RwndTrace`]. /// -/// Each step of a rwnd trace produces one `RwndDecision` paired with a -/// [`Duration`] (see [`RwndTrace`]). Both fields are optional and independent: -/// a step may resize the socket buffer, advance the receive model, both, or -/// neither (though a step that sets neither is effectively a no-op). -#[derive(Debug, Clone, PartialEq)] +/// The two fields are independent, and between them they say what the receiver +/// does for this step's duration. +/// +/// `set_rcv_buf` sizes the receive buffer *and* states that the application is +/// keeping up with it: the buffer holds that many bytes, the advertised window +/// is held at that value, and the application drains continuously so the +/// window's right edge slides forward with the data received. That models a +/// receiver whose buffer stopped growing -- in-flight ends up limited by the +/// window, and the window itself stays put. +/// +/// `app_read_bytes` states the opposite situation: over this step the +/// application reads exactly that many bytes and then stops. The window is +/// whatever is left of the buffer once the unread backlog is subtracted, so it +/// shrinks as data arrives and reaches zero when the buffer fills. That models a +/// receiver whose application is the bottleneck. +/// +/// Because the fields are independent, all four combinations are meaningful and +/// none is a special case: +/// +/// | `set_rcv_buf` | `app_read_bytes` | meaning | +/// |---|---|---| +/// | `Some(n)` | `None` | buffer `n`, window pinned at `n`, application drains continuously | +/// | `None` | `Some(m)` | read `m` bytes against the standing buffer; window is what is left | +/// | `Some(n)` | `Some(m)` | resize the buffer to `n`, then read `m` bytes from it | +/// | `None` | `None` | carry the previous configuration forward for this step | +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct RwndDecision { - /// If `Some`, reconfigure the socket's receive buffer to this size at this step. + /// If `Some`, size the receive buffer to this many bytes and hold the + /// advertised window there, with the application draining continuously. pub set_rcv_buf: Option, - /// If `Some`, the app-read or observed-remaining action for this step. - pub action: Option, + /// If `Some`, the application reads exactly this many bytes over this step + /// and then stops; the window follows from what is left unread. + pub app_read_bytes: Option, } /// This is a trait that represents a trace of receive-window decisions over time. /// /// The trace is a sequence of `(rwnd_decision, duration)` pairs. The decision -/// describes how the socket's receive buffer, the application's read behavior, -/// and/or the observed remaining window change at this step; the duration is -/// how long this configuration lasts before the next step applies. +/// describes what the receiver does -- how large its buffer is and how its +/// application reads from it -- and the duration is how long that lasts before +/// the next step applies. /// /// For example, if the sequence is -/// `[(set_rcv_buf=64KB, app_read=1KB, 1s), (rwnd_remaining=32KB, 2s)]`, -/// then the receive buffer is resized to 64KB and the app reads 1KB for 1s, -/// then the observed rwnd becomes 32KB for 2s. +/// `[(set_rcv_buf=64KB, 1s), (app_read_bytes=1KB, 2s)]`, then for 1s the +/// receiver holds a 64KB window and drains it as fast as data arrives, and for +/// the next 2s its application reads only 1KB, so the window decays from 64KB +/// as the unread backlog grows. /// /// Each `next_rwnd` call returns **the next decision and its duration** in the /// sequence, or **None** when the trace is exhausted. Mirrors the shape of diff --git a/src/model/rwnd.rs b/src/model/rwnd.rs index 62506db..f6f7461 100644 --- a/src/model/rwnd.rs +++ b/src/model/rwnd.rs @@ -7,13 +7,22 @@ //! - [`StaticRwnd`]: A trace model with a single rwnd decision. //! - [`RepeatedRwndPattern`]: A trace model with a repeated rwnd pattern. //! +//! ## Step semantics +//! +//! A step carries two independent fields, `set_rcv_buf` and `app_read_bytes`, +//! and between them they say what the receiver does for the step's duration. +//! See [`RwndDecision`] for the full table; in short, `set_rcv_buf` states a +//! buffer the application keeps drained (so the window stays at that value), +//! and `app_read_bytes` states an application that reads only so much (so the +//! window decays as the backlog grows). +//! //! ## Examples //! //! An example to build model from configuration: //! //! ``` //! # use netem_trace::model::StaticRwndConfig; -//! # use netem_trace::{Duration, RwndTrace, RwndAction}; +//! # use netem_trace::{Duration, RwndTrace}; //! let mut static_rwnd = StaticRwndConfig::new() //! .set_rcv_buf(65536) //! .app_read(1024) @@ -21,7 +30,7 @@ //! .build(); //! let (decision, duration) = static_rwnd.next_rwnd().unwrap(); //! assert_eq!(decision.set_rcv_buf, Some(65536)); -//! assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +//! assert_eq!(decision.app_read_bytes, Some(1024)); //! assert_eq!(duration, Duration::from_secs(1)); //! assert_eq!(static_rwnd.next_rwnd(), None); //! ``` @@ -30,30 +39,21 @@ //! //! ``` //! # use netem_trace::model::{StaticRwndConfig, RwndTraceConfig}; -//! # use netem_trace::{Duration, RwndTrace, RwndAction}; +//! # use netem_trace::{Duration, RwndTrace}; //! # #[cfg(feature = "human")] -//! # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":\"1s\",\"rwnd_remaining\":32768}}],\"count\":2}}"; -//! // The content would be "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}],\"count\":2}}" +//! # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536}},{\"StaticRwndConfig\":{\"duration\":\"1s\",\"app_read_bytes\":1024}}],\"count\":2}}"; +//! // The content would be "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"app_read_bytes\":1024}}],\"count\":2}}" //! // if the `human` feature is not enabled. //! # #[cfg(not(feature = "human"))] -//! let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}],\"count\":2}}"; +//! let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"app_read_bytes\":1024}}],\"count\":2}}"; //! let des: Box = serde_json::from_str(config_file_content).unwrap(); //! let mut model = des.into_model(); //! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); -//! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); -//! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +//! assert_eq!(decision.set_rcv_buf, Some(65536)); //! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); -//! assert_eq!(model.next_rwnd(), None); +//! assert_eq!(decision.app_read_bytes, Some(1024)); //! ``` -//! -//! At most one of `app_read_bytes` or `rwnd_remaining` may be set per step — -//! never both. A step with neither produces [`RwndDecision::action`] as `None`, -//! which is valid for steps that only reconfigure the receive buffer. -use crate::{Duration, RwndAction, RwndDecision, RwndTrace}; +use crate::{Duration, RwndDecision, RwndTrace}; use dyn_clone::DynClone; /// This trait is used to convert a rwnd trace configuration into a rwnd trace model. @@ -70,7 +70,7 @@ pub trait RwndTraceConfig: DynClone + Send { dyn_clone::clone_trait_object!(RwndTraceConfig); #[cfg(feature = "serde")] -use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde::{Deserialize, Serialize}; /// The model of a static rwnd trace: a single decision valid for one duration. /// @@ -78,15 +78,14 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer}; /// /// ``` /// # use netem_trace::model::StaticRwndConfig; -/// # use netem_trace::{Duration, RwndTrace, RwndAction}; +/// # use netem_trace::{Duration, RwndTrace}; /// let mut static_rwnd = StaticRwndConfig::new() /// .set_rcv_buf(65536) -/// .app_read(1024) /// .duration(Duration::from_secs(1)) /// .build(); /// let (decision, duration) = static_rwnd.next_rwnd().unwrap(); /// assert_eq!(decision.set_rcv_buf, Some(65536)); -/// assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +/// assert_eq!(decision.app_read_bytes, None); /// assert_eq!(duration, Duration::from_secs(1)); /// assert_eq!(static_rwnd.next_rwnd(), None); /// ``` @@ -99,85 +98,39 @@ pub struct StaticRwnd { /// The configuration struct for [`StaticRwnd`]. /// /// The serialized JSON form is **flat**: a step looks like -/// `{"duration":"1s","set_rcv_buf":65536,"app_read_bytes":1024}` (or -/// `{"duration":"1s","rwnd_remaining":32768}`), never with an `action` wrapper. +/// `{"duration":"1s","set_rcv_buf":65536}` or +/// `{"duration":"1s","app_read_bytes":1024}`, and may carry both keys. /// -/// At most one of `app_read_bytes` / `rwnd_remaining` may be set; the deserializer -/// rejects inputs where both are present. A step with neither is valid and produces -/// [`RwndDecision::action`] as `None` (useful for steps that only reconfigure the -/// receive buffer). +/// The two fields are independent -- there is no invariant to enforce, so +/// `Serialize`/`Deserialize` are derived. A step with neither field is valid and +/// carries the previous configuration forward. +/// +/// Unknown keys are rejected rather than ignored. The schema dropped +/// `rwnd_remaining`, and serde's default of skipping what it does not recognise +/// would turn a trace written against the old schema into a run of steps that +/// state nothing -- a replay that looks healthy while reproducing no receiver at +/// all. Failing to deserialize names the offending field instead. +#[cfg_attr( + feature = "serde", + derive(Serialize, Deserialize), + serde(default, deny_unknown_fields) +)] #[derive(Debug, Clone, Default)] pub struct StaticRwndConfig { + #[cfg_attr( + feature = "human", + serde(with = "humantime_serde"), + serde(skip_serializing_if = "Option::is_none") + )] + #[cfg_attr( + all(feature = "serde", not(feature = "human")), + serde(skip_serializing_if = "Option::is_none") + )] pub duration: Option, + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] pub set_rcv_buf: Option, - pub action: Option, -} - -#[cfg(feature = "serde")] -impl<'de> Deserialize<'de> for StaticRwndConfig { - fn deserialize>(deserializer: D) -> Result { - #[derive(Deserialize, Default)] - #[serde(default)] - struct Helper { - #[cfg_attr(feature = "human", serde(with = "humantime_serde"))] - #[serde(default)] - duration: Option, - #[serde(default)] - set_rcv_buf: Option, - #[serde(default)] - app_read_bytes: Option, - #[serde(default)] - rwnd_remaining: Option, - } - - let h = Helper::deserialize(deserializer)?; - let action = match (h.app_read_bytes, h.rwnd_remaining) { - (Some(bytes), None) => Some(RwndAction::AppRead { bytes }), - (None, Some(rwnd)) => Some(RwndAction::Remaining { rwnd }), - (Some(_), Some(_)) => { - return Err(serde::de::Error::custom( - "rwnd step cannot set both `app_read_bytes` and `rwnd_remaining`", - )); - } - (None, None) => None, - }; - Ok(Self { - duration: h.duration, - set_rcv_buf: h.set_rcv_buf, - action, - }) - } -} - -#[cfg(feature = "serde")] -impl Serialize for StaticRwndConfig { - fn serialize(&self, serializer: S) -> Result { - #[derive(Serialize)] - struct Out { - #[serde(skip_serializing_if = "Option::is_none")] - #[cfg_attr(feature = "human", serde(with = "humantime_serde"))] - duration: Option, - #[serde(skip_serializing_if = "Option::is_none")] - set_rcv_buf: Option, - #[serde(skip_serializing_if = "Option::is_none")] - app_read_bytes: Option, - #[serde(skip_serializing_if = "Option::is_none")] - rwnd_remaining: Option, - } - - let (app_read_bytes, rwnd_remaining) = match &self.action { - Some(RwndAction::AppRead { bytes }) => (Some(*bytes), None), - Some(RwndAction::Remaining { rwnd }) => (None, Some(*rwnd)), - None => (None, None), - }; - Out { - duration: self.duration, - set_rcv_buf: self.set_rcv_buf, - app_read_bytes, - rwnd_remaining, - } - .serialize(serializer) - } + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] + pub app_read_bytes: Option, } /// The model contains an array of rwnd trace models. @@ -194,15 +147,16 @@ impl Serialize for StaticRwndConfig { /// /// ``` /// # use netem_trace::model::{StaticRwndConfig, RwndTraceConfig}; -/// # use netem_trace::{Duration, RwndTrace, RwndAction}; +/// # use netem_trace::{Duration, RwndTrace}; /// # #[cfg(feature = "human")] -/// # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":\"1s\",\"rwnd_remaining\":32768}}],\"count\":2}}"; +/// # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}],\"count\":2}}"; /// # #[cfg(not(feature = "human"))] -/// let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}],\"count\":2}}"; +/// let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}],\"count\":2}}"; /// let des: Box = serde_json::from_str(config_file_content).unwrap(); /// let mut model = des.into_model(); /// let (decision, _) = model.next_rwnd().unwrap(); -/// assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +/// assert_eq!(decision.set_rcv_buf, Some(65536)); +/// assert_eq!(decision.app_read_bytes, Some(1024)); /// ``` pub struct RepeatedRwndPattern { pub pattern: Vec>, @@ -228,7 +182,7 @@ impl RwndTrace for StaticRwnd { if duration.is_zero() { None } else { - Some((self.decision.clone(), duration)) + Some((self.decision, duration)) } } else { None @@ -279,7 +233,7 @@ impl StaticRwndConfig { Self { duration: None, set_rcv_buf: None, - action: None, + app_read_bytes: None, } } @@ -288,18 +242,16 @@ impl StaticRwndConfig { self } + /// Size the receive buffer, and hold the advertised window there with the + /// application draining continuously. pub fn set_rcv_buf(mut self, set_rcv_buf: u64) -> Self { self.set_rcv_buf = Some(set_rcv_buf); self } + /// The application reads exactly this many bytes over the step, then stops. pub fn app_read(mut self, bytes: u64) -> Self { - self.action = Some(RwndAction::AppRead { bytes }); - self - } - - pub fn remaining(mut self, rwnd: u64) -> Self { - self.action = Some(RwndAction::Remaining { rwnd }); + self.app_read_bytes = Some(bytes); self } @@ -307,7 +259,7 @@ impl StaticRwndConfig { StaticRwnd { decision: RwndDecision { set_rcv_buf: self.set_rcv_buf, - action: self.action, + app_read_bytes: self.app_read_bytes, }, duration: Some(self.duration.unwrap_or_else(|| Duration::from_secs(1))), } @@ -360,7 +312,6 @@ impl_rwnd_trace_config!(RepeatedRwndPatternConfig); #[cfg(test)] mod test { use super::*; - use crate::model::StaticRwndConfig; use crate::RwndTrace; #[test] @@ -372,35 +323,50 @@ mod test { .build(); let (decision, duration) = static_rwnd.next_rwnd().unwrap(); assert_eq!(decision.set_rcv_buf, Some(65536)); - assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); + assert_eq!(decision.app_read_bytes, Some(1024)); assert_eq!(duration, Duration::from_secs(1)); assert_eq!(static_rwnd.next_rwnd(), None); } + /// A buffer on its own is a complete statement: the window sits there and + /// the application keeps up. This is what used to need `rwnd_remaining`. #[test] - fn test_static_rwnd_model_remaining() { + fn test_static_rwnd_model_buffer_only() { let mut static_rwnd = StaticRwndConfig::new() - .remaining(32768) + .set_rcv_buf(32768) .duration(Duration::from_secs(2)) .build(); let (decision, duration) = static_rwnd.next_rwnd().unwrap(); - assert_eq!(decision.set_rcv_buf, None); - assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); + assert_eq!(decision.set_rcv_buf, Some(32768)); + assert_eq!(decision.app_read_bytes, None); assert_eq!(duration, Duration::from_secs(2)); assert_eq!(static_rwnd.next_rwnd(), None); } + /// A step may carry neither field, and then it simply holds whatever the + /// previous step configured for its duration. + #[test] + fn test_static_rwnd_model_carries_forward() { + let mut model = StaticRwndConfig::new() + .duration(Duration::from_secs(1)) + .build(); + let (decision, duration) = model.next_rwnd().unwrap(); + assert_eq!(decision.set_rcv_buf, None); + assert_eq!(decision.app_read_bytes, None); + assert_eq!(duration, Duration::from_secs(1)); + } + #[test] fn test_repeated_rwnd_pattern() { let pat = vec![ Box::new( StaticRwndConfig::new() - .app_read(1024) + .set_rcv_buf(65536) .duration(Duration::from_secs(1)), ) as Box, Box::new( StaticRwndConfig::new() - .remaining(32768) + .app_read(1024) .duration(Duration::from_secs(1)), ) as Box, ]; @@ -409,134 +375,80 @@ mod test { .count(2) .build(); let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::AppRead { bytes: 1024 })); + assert_eq!(next.0.set_rcv_buf, Some(65536)); assert_eq!(next.1, Duration::from_secs(1)); - let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::Remaining { rwnd: 32768 })); - let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::AppRead { bytes: 1024 })); - let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::Remaining { rwnd: 32768 })); + assert_eq!(model.next_rwnd().unwrap().0.app_read_bytes, Some(1024)); + assert_eq!(model.next_rwnd().unwrap().0.set_rcv_buf, Some(65536)); + assert_eq!(model.next_rwnd().unwrap().0.app_read_bytes, Some(1024)); assert_eq!(model.next_rwnd(), None); } #[test] #[cfg(feature = "serde")] - fn test_serde_roundtrip_app_read() { + fn test_serde_roundtrip_buffer_only() { let cfg = Box::new( StaticRwndConfig::new() .set_rcv_buf(65536) - .app_read(1024) .duration(Duration::from_secs(1)), ) as Box; let ser_str = serde_json::to_string(&cfg).unwrap(); #[cfg(feature = "human")] - let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}"; + let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536}}"; #[cfg(not(feature = "human"))] - let expected = "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}"; + let expected = + "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536}}"; assert_eq!(ser_str, expected); let des: Box = serde_json::from_str(&ser_str).unwrap(); let mut model = des.into_model(); let (decision, duration) = model.next_rwnd().unwrap(); assert_eq!(decision.set_rcv_buf, Some(65536)); - assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); + assert_eq!(decision.app_read_bytes, None); assert_eq!(duration, Duration::from_secs(1)); } + /// Both keys on one step is legal now, and round-trips. #[test] #[cfg(feature = "serde")] - fn test_serde_roundtrip_remaining() { - let cfg = Box::new( - StaticRwndConfig::new() - .remaining(32768) - .duration(Duration::from_secs(1)), - ) as Box; - let ser_str = serde_json::to_string(&cfg).unwrap(); - #[cfg(feature = "human")] - let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"rwnd_remaining\":32768}}"; - #[cfg(not(feature = "human"))] - let expected = "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}"; - assert_eq!(ser_str, expected); - - let des: Box = serde_json::from_str(&ser_str).unwrap(); + fn test_serde_roundtrip_both_fields() { + let json = "{\"StaticRwndConfig\":{\"set_rcv_buf\":131072,\"app_read_bytes\":4096}}"; + let des: Box = serde_json::from_str(json).unwrap(); let mut model = des.into_model(); let (decision, _) = model.next_rwnd().unwrap(); - assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); - } - - #[test] - #[cfg(feature = "serde")] - fn test_serde_rejects_both() { - // Omit duration to avoid the human/non-human format ambiguity; we're testing - // the action constraint, not duration parsing. - let json = "{\"StaticRwndConfig\":{\"app_read_bytes\":1024,\"rwnd_remaining\":32768}}"; - let result: Result, _> = serde_json::from_str(json); - let err = result - .err() - .expect("deserialization should have failed") - .to_string(); - assert!( - err.contains("cannot set both"), - "expected 'cannot set both' in error, got: {err}" - ); - } - - #[test] - fn test_static_rwnd_set_rcv_buf_only() { - let mut model = StaticRwndConfig::new() - .set_rcv_buf(131072) - .duration(Duration::from_secs(1)) - .build(); - let (decision, duration) = model.next_rwnd().unwrap(); assert_eq!(decision.set_rcv_buf, Some(131072)); - assert_eq!(decision.action, None); - assert_eq!(duration, Duration::from_secs(1)); - assert_eq!(model.next_rwnd(), None); + assert_eq!(decision.app_read_bytes, Some(4096)); } #[test] #[cfg(feature = "serde")] - fn test_serde_roundtrip_set_rcv_buf_only() { - let cfg = Box::new( - StaticRwndConfig::new() - .set_rcv_buf(131072) - .duration(Duration::from_secs(1)), - ) as Box; + fn test_serde_omits_absent_fields() { + let cfg = Box::new(StaticRwndConfig::new().app_read(0)) as Box; let ser_str = serde_json::to_string(&cfg).unwrap(); - #[cfg(feature = "human")] - let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":131072}}"; - #[cfg(not(feature = "human"))] - let expected = - "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":131072}}"; - assert_eq!(ser_str, expected); - - let des: Box = serde_json::from_str(&ser_str).unwrap(); - let mut model = des.into_model(); - let (decision, duration) = model.next_rwnd().unwrap(); - assert_eq!(decision.set_rcv_buf, Some(131072)); - assert_eq!(decision.action, None); - assert_eq!(duration, Duration::from_secs(1)); - assert_eq!(model.next_rwnd(), None); + assert!(!ser_str.contains("set_rcv_buf"), "got: {ser_str}"); + assert!(ser_str.contains("app_read_bytes"), "got: {ser_str}"); } + /// A trace written against the old schema names a field that no longer + /// exists. Rejecting it is the point: silently ignoring `rwnd_remaining` + /// would turn every window step into a no-op and replay a trace that says + /// nothing, which looks like a healthy run producing wrong numbers. #[test] #[cfg(feature = "serde")] - fn test_serde_action_none_when_neither_set() { - // A step with only set_rcv_buf and no action fields should deserialize to action: None. - let json = "{\"StaticRwndConfig\":{\"set_rcv_buf\":65536}}"; - let des: Box = serde_json::from_str(json).unwrap(); - let mut model = des.into_model(); - let (decision, _) = model.next_rwnd().unwrap(); - assert_eq!(decision.set_rcv_buf, Some(65536)); - assert_eq!(decision.action, None); + fn test_serde_rejects_the_old_rwnd_remaining_field() { + let json = "{\"StaticRwndConfig\":{\"rwnd_remaining\":32768}}"; + let result: Result, _> = serde_json::from_str(json); + let err = result + .err() + .expect("a trace using the removed field should not deserialize") + .to_string(); + assert!(err.contains("rwnd_remaining"), "got: {err}"); } #[test] fn test_repeated_rwnd_pattern_all_zero_duration_terminates() { // All inner models have duration == 0 and return None immediately. - // With count == 0 (infinite repeat) the old recursive implementation - // would spin forever; the loop-based one must return None promptly. + // With count == 0 (infinite repeat) a recursive implementation would + // spin forever; the loop-based one must return None promptly. let pat = vec![ Box::new( StaticRwndConfig::new() @@ -545,7 +457,7 @@ mod test { ) as Box, Box::new( StaticRwndConfig::new() - .remaining(32768) + .set_rcv_buf(32768) .duration(Duration::ZERO), ) as Box, ]; From c1d214fc123aec22afdf489aef30cdbe63ab945a Mon Sep 17 00:00:00 2001 From: zhc Date: Sat, 29 Aug 2026 05:37:10 +0000 Subject: [PATCH 2/2] docs(model): spell out what a step with both fields does --- src/lib.rs | 33 ++++++++++++++++++----- src/model/rwnd.rs | 67 ++++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 91 insertions(+), 9 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 7ef2924..c87d273 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -243,12 +243,12 @@ pub trait DuplicateTrace: Send { /// The two fields are independent, and between them they say what the receiver /// does for this step's duration. /// -/// `set_rcv_buf` sizes the receive buffer *and* states that the application is -/// keeping up with it: the buffer holds that many bytes, the advertised window -/// is held at that value, and the application drains continuously so the -/// window's right edge slides forward with the data received. That models a -/// receiver whose buffer stopped growing -- in-flight ends up limited by the -/// window, and the window itself stays put. +/// `set_rcv_buf` sizes the receive buffer. On its own it also states that the +/// application is keeping up with it: the buffer holds that many bytes, the +/// advertised window is held at that value, and the application drains +/// continuously so the window's right edge slides forward with the data +/// received. That models a receiver whose buffer stopped growing -- in-flight +/// ends up limited by the window, and the window itself stays put. /// /// `app_read_bytes` states the opposite situation: over this step the /// application reads exactly that many bytes and then stops. The window is @@ -265,6 +265,27 @@ pub trait DuplicateTrace: Send { /// | `None` | `Some(m)` | read `m` bytes against the standing buffer; window is what is left | /// | `Some(n)` | `Some(m)` | resize the buffer to `n`, then read `m` bytes from it | /// | `None` | `None` | carry the previous configuration forward for this step | +/// +/// # Both fields on one step +/// +/// The two apply in order: the buffer is resized first, and the read is taken +/// against the new size. Three consequences are worth stating outright, because +/// they are what distinguishes this from a buffer-only step: +/// +/// - **The window is not pinned.** The "application keeps up" half of +/// `set_rcv_buf` belongs to a step that states no read. Once `app_read_bytes` +/// is present it is the application's behaviour, so the window follows from +/// `n - unread` and decays as data arrives, exactly as for a read-only step. +/// A step of `set_rcv_buf: n` and `app_read_bytes: m` is therefore *not* +/// equivalent to a buffer-only step of `n` followed by a read of `m`. +/// - **The backlog survives the resize.** Only the capacity changes; bytes +/// already received and not yet read stay unread. Resizing does not discard, +/// deliver, or otherwise account for them. +/// - **The window saturates at zero.** With `unread` bytes outstanding the +/// window is `n.saturating_sub(unread)`, so resizing to a value at or below +/// the current backlog advertises a zero window until the application reads +/// its way back under the new size. This is the intended way to state a +/// receiver that shrank its buffer while behind. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct RwndDecision { /// If `Some`, size the receive buffer to this many bytes and hold the diff --git a/src/model/rwnd.rs b/src/model/rwnd.rs index f6f7461..abf5ec7 100644 --- a/src/model/rwnd.rs +++ b/src/model/rwnd.rs @@ -16,6 +16,14 @@ //! and `app_read_bytes` states an application that reads only so much (so the //! window decays as the backlog grows). //! +//! A step carrying **both** resizes the buffer first and takes the read against +//! the new size. The window is then `set_rcv_buf - unread`, not pinned at +//! `set_rcv_buf`: stating a read means the application is the bottleneck, so the +//! "application keeps up" half of a buffer-only step no longer applies. The +//! unread backlog carries across the resize untouched, so a resize to at or +//! below the current backlog advertises a zero window until the application +//! catches up. +//! //! ## Examples //! //! An example to build model from configuration: @@ -35,6 +43,46 @@ //! assert_eq!(static_rwnd.next_rwnd(), None); //! ``` //! +//! The step above carries both fields, so it resizes the buffer to 64 KiB and +//! then has the application read 1 KiB from it. The window that follows is +//! `65536 - unread` rather than a window pinned at 65536 -- compare the +//! buffer-only step below, which does pin it: +//! +//! ``` +//! # use netem_trace::model::StaticRwndConfig; +//! # use netem_trace::{Duration, RwndTrace}; +//! // Buffer only: the window is held at 65536 and the application keeps up. +//! let mut pinned = StaticRwndConfig::new() +//! .set_rcv_buf(65536) +//! .duration(Duration::from_secs(1)) +//! .build(); +//! let (decision, _) = pinned.next_rwnd().unwrap(); +//! assert_eq!(decision.set_rcv_buf, Some(65536)); +//! assert_eq!(decision.app_read_bytes, None); +//! +//! // Both: same buffer, but the application now reads only 1 KiB per step, so +//! // the window follows the backlog instead of staying at 65536. +//! let mut app_limited = StaticRwndConfig::new() +//! .set_rcv_buf(65536) +//! .app_read(1024) +//! .duration(Duration::from_secs(1)) +//! .build(); +//! let (decision, _) = app_limited.next_rwnd().unwrap(); +//! assert_eq!(decision.set_rcv_buf, Some(65536)); +//! assert_eq!(decision.app_read_bytes, Some(1024)); +//! +//! // Shrinking the buffer under a standing backlog is stated the same way; the +//! // window saturates at zero until the application reads its way back under it. +//! let mut shrink = StaticRwndConfig::new() +//! .set_rcv_buf(8192) +//! .app_read(0) +//! .duration(Duration::from_secs(1)) +//! .build(); +//! let (decision, _) = shrink.next_rwnd().unwrap(); +//! assert_eq!(decision.set_rcv_buf, Some(8192)); +//! assert_eq!(decision.app_read_bytes, Some(0)); +//! ``` +//! //! A more common use case is to build model from a configuration file (e.g. json file): //! //! ``` @@ -242,14 +290,24 @@ impl StaticRwndConfig { self } - /// Size the receive buffer, and hold the advertised window there with the - /// application draining continuously. + /// Size the receive buffer. + /// + /// On its own this also holds the advertised window at that value, with the + /// application draining continuously. Combined with [`Self::app_read`] it + /// only sets the size: the resize applies first, the read is taken against + /// the new size, and the window then follows `set_rcv_buf - unread` rather + /// than being pinned. pub fn set_rcv_buf(mut self, set_rcv_buf: u64) -> Self { self.set_rcv_buf = Some(set_rcv_buf); self } /// The application reads exactly this many bytes over the step, then stops. + /// + /// The window follows from what is left unread of the standing buffer, or + /// of the buffer this step sets when combined with [`Self::set_rcv_buf`]. + /// The backlog carries across such a resize untouched, so the window + /// saturates at zero if the new size is at or below it. pub fn app_read(mut self, bytes: u64) -> Self { self.app_read_bytes = Some(bytes); self @@ -314,8 +372,11 @@ mod test { use super::*; use crate::RwndTrace; + /// Both fields on one step: the buffer is resized and the read is taken + /// against the new size, so the decision carries both rather than one + /// overriding the other. #[test] - fn test_static_rwnd_model_app_read() { + fn test_static_rwnd_model_buffer_and_app_read() { let mut static_rwnd = StaticRwndConfig::new() .set_rcv_buf(65536) .app_read(1024)