diff --git a/rattan-core/src/cells/bandwidth/queue/mod.rs b/rattan-core/src/cells/bandwidth/queue/mod.rs index bca02677..2e2f6e8e 100644 --- a/rattan-core/src/cells/bandwidth/queue/mod.rs +++ b/rattan-core/src/cells/bandwidth/queue/mod.rs @@ -15,11 +15,15 @@ mod codel; mod drophead; mod droptail; mod infinite; +mod pie; +mod red; pub use codel::*; pub use drophead::*; pub use droptail::*; pub use infinite::*; +pub use pie::*; +pub use red::*; #[cfg(feature = "serde")] fn serde_default(t: &T) -> bool { diff --git a/rattan-core/src/cells/bandwidth/queue/pie.rs b/rattan-core/src/cells/bandwidth/queue/pie.rs new file mode 100644 index 00000000..a0eb571d --- /dev/null +++ b/rattan-core/src/cells/bandwidth/queue/pie.rs @@ -0,0 +1,957 @@ +//! PIE (Proportional Integral controller Enhanced) queue. +//! +//! # References +//! +//! - PIE paper: +//! +//! - RFC 8033 (PIE AQM): +//! +//! - Linux kernel PIE: +//! [`include/net/pie.h`](https://github.com/torvalds/linux/blob/master/include/net/pie.h) +//! and [`net/sched/sch_pie.c`](https://github.com/torvalds/linux/blob/master/net/sched/sch_pie.c) +//! (referenced as "kernel" throughout this documentation) +//! +//! # RFC 8033 version: Appendix A vs. Appendix B +//! +//! RFC 8033 contains two descriptions of the PIE algorithm: +//! +//! - **Appendix A** is the "original paper" algorithm: configurable α/β with +//! *dynamic* per-probability-scaling, accumulated-probability-based drop +//! decisions, and an optional timestamp-based delay estimation path. +//! +//! - **Appendix B** is a simplified pseudo-code reference implementation: +//! fixed `α = 0.125`, `β = 1.25`, *static* `p_increment` damping tables, +//! per-packet random drop (no accumulation), and always drain-rate-based +//! delay estimation. +//! +//! The kernel follows **Appendix A** (with its own extensions). This +//! implementation follows **Appendix B**. This is the most fundamental +//! design difference and explains nearly every other divergence listed below. +//! +//! # Differences from the Linux kernel implementation +//! +//! The 16 differences below fall into three categories: +//! +//! | Category | Sections | Explanation | +//! |----------|----------|-------------| +//! | **RFC 8033 Appendix A vs. B** | §3, §4, §5, §12 | The kernel follows Appendix A; this implementation follows Appendix B. These are deliberate design choices, not omissions. | +//! | **Kernel-specific extensions** | §6, §7, §8, §11, §13 | Features the Linux kernel added beyond what either RFC appendix describes. | +//! | **Implementation / architectural choices** | §1, §2, §9, §10, §14, §15, §16 | Differences arising from the simulation context (floating-point, event-driven, seedable RNG) or architectural constraints. | +//! +//! ## 1. Fixed-point integers (kernel) vs. floating-point (here) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | `prob` | `u64` scaled by `MAX_PROB` (`U64_MAX >> 8`) | `f64` in `[0.0, 1.0]` | +//! | `avg_dq_rate` | `u32` scaled by `PIE_SCALE` (shift 8) | `f64` in B/s | +//! | delay | `psched_time_t` (kernel time ticks) | `f64` in seconds | +//! | `burst_time` | `psched_time_t` ticks | `f64` in milliseconds | +//! | EWMA for drain rate | `avg = (avg - (avg >> 3)) + (count >> 3)` | `avg = 0.875 * avg + 0.125 * rate` | +//! | `p_increment` damping | dynamic α/β bit-shifts (see §4) | `f64` division by `2048`/`512`/`…` | +//! +//! The EWMA update is mathematically identical (`1/8 = 0.125` weight). The +//! damping thresholds are also identical (see §4 below). +//! +//! ## 2. Drop-probability update trigger +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | trigger | dedicated kernel timer, fires every `tupdate` jiffies | event-driven, on `enqueue()` | +//! | fires when idle? | yes — timer always runs | no — requires a packet arrival | +//! | grid | timer re-arms for `jiffies + tupdate` | forward-shifted fixed `t_update` grid | +//! | first fire | 500 ms after init | on first enqueue | +//! +//! **Kernel**: A dedicated timer (`adapt_timer`) fires every `tupdate` +//! regardless of whether packets are arriving. During prolonged idle periods +//! the kernel therefore continues to decay `prob` and update state. +//! +//! **Here**: Probability update happens inside `enqueue()`. When a packet +//! arrives and at least `t_update` has elapsed since the last update, one +//! call to `update_drop_probability()` is made and the timer is advanced by +//! exactly `t_update` (not reset to current time). This "fixed grid" +//! approach allows multiple intervals to be caught up over successive packet +//! arrivals after a long idle. However, if no packets arrive, no updates +//! occur. +//! +//! **Why event-driven instead of a dedicated timer?** This simulator operates +//! on *logical* (simulation) time, not wall-clock time. A wall-clock timer +//! cannot accurately target a logical-time instant, especially under +//! variable-speed simulation, pause/resume, or replay scenarios. Spawning an +//! OS timer per queue instance would be prohibitively expensive when +//! simulating hundreds or thousands of flows. +//! +//! ## 3. Alpha and Beta: configurable (kernel) vs. fixed (here) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | definition | `struct pie_params.alpha`, `.beta` (user-tunable 0–32) | `tilde_alpha = 0.125`, `tilde_beta = 1.25` (hard-coded) | +//! | internal scaling | `(param * MAX_PROB / PSCHED_TICKS_PER_SEC) >> 4` | direct `f64` | +//! | defaults | α = 2, β = 20 → 2/16 = 0.125, 20/16 = 1.25 | 0.125, 1.25 | +//! | tuning via | `tc qdisc ... pie alpha N beta M` | not configurable | +//! +//! The **default** values are equivalent: `2/16 = 0.125`, `20/16 = 1.25`. +//! However, the kernel exposes α and β as user-tunable knobs and dynamically +//! scales them based on current probability (described in §4). This +//! implementation hard-codes the RFC 8033 Appendix B base values and applies +//! static damping to `p_increment` instead. +//! +//! ## 4. Probability-increment damping +//! +//! Both implementations damp the probability adjustment when `prob` is small, +//! but through different mechanisms (Appendix A dynamically scales α/β; +//! Appendix B statically divides `p_increment` — thresholds and effective +//! damping factors are mathematically identical): +//! +//! **Kernel (Appendix A — dynamic α/β scaling):** +//! +//! When `prob < MAX_PROB / 10`: +//! 1. α and β are halved (`>>= 1`). +//! 2. Then repeatedly quartered (`>>= 2`) for each power-of-10 threshold: +//! `prob < MAX_PROB / 100` → quarter again, … up to `MAX_PROB / 10⁶`. +//! +//! **Here (Appendix B — static `p_increment` division):** +//! +//! `p_increment` is divided by a precomputed factor depending on `p`: +//! +//! | `p` range | Division factor | +//! |-----------|----------------:| +//! | `p ≥ 0.1` | 1 (no damping) | +//! | `0.01 ≤ p < 0.1` | 2 | +//! | `0.001 ≤ p < 0.01` | 8 | +//! | `0.0001 ≤ p < 0.001` | 32 | +//! | `0.00001 ≤ p < 0.0001` | 128 | +//! | `0.000001 ≤ p < 0.00001` | 512 | +//! | `p < 0.000001` | 2048 | +//! +//! The thresholds *and* the effective damping factors are **mathematically +//! identical** between the two approaches. The kernel modifies α/β before +//! computing the delta; this implementation computes the full delta first, +//! then divides. Because `delta = α·Δcur + β·Δold`, scaling α and β by +//! factor *k* is equivalent to scaling the delta by *k*. +//! +//! ## 5. Drop decision: probability accumulation (kernel) vs. per-packet random (here) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | mechanism | `accu_prob` accumulates probability across packets | `rand < p` each packet | +//! | drop trigger | accumulated value crosses threshold | immediate comparison | +//! | burst tolerance | natural — multiple low-prob packets needed to trigger | via explicit `burst_allowance` check | +//! | RFC version | Appendix A | Appendix B | +//! +//! **Kernel**: Maintains an `accu_prob` counter in `pie_drop_early()`. Each +//! arriving packet adds `local_prob` (possibly scaled by `bytemode`) to the +//! accumulator. A packet is dropped only when `accu_prob` crosses a +//! threshold (`(MAX_PROB / 2) * 17`). This naturally spaces out drops — +//! several low-probability packets must arrive before one is dropped. +//! +//! **Here**: Each `should_drop()` call draws a fresh uniform random value and +//! compares directly against `p`. This is the simpler Appendix B approach +//! and produces the same statistical drop rate over many packets; the +//! difference is that drops are less evenly spaced (potentially clumpier) +//! without accumulation. +//! +//! ## 6. ECN (Explicit Congestion Notification) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | ECN marking | supported (`TCA_PIE_ECN` flag) | not supported | +//! | mark condition | `prob ≤ MAX_PROB / 10` + packet is ECN-capable | N/A | +//! | mark counter | `ecn_mark` stat | N/A | +//! +//! The kernel can mark ECN-capable packets (setting the CE codepoint) instead +//! of dropping them when the drop probability is moderate. This +//! implementation always drops. Adding ECN support would require extending +//! the `Packet` trait with an ECN field. +//! +//! ## 7. Bytemode (packet-size-scaled probability) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | feature | optional `bytemode` flag (`TCA_PIE_BYTEMODE`) | not supported | +//! | scaling | `local_prob = prob * pkt_size / mtu` (for packets ≤ MTU) | N/A | +//! +//! When enabled, the kernel scales the drop probability proportionally to the +//! packet size, making larger packets more likely to be dropped. This +//! implementation always treats all packets equally regardless of size. +//! +//! ## 8. Non-linear boost for high delay (>250 ms) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | trigger | `qdelay > 250 ms` | not implemented | +//! | effect | `delta += 2%` of max probability | N/A | +//! +//! When the estimated queue delay exceeds 250 ms, the kernel adds an extra +//! 2% to the probability delta to more aggressively counteract severe +//! congestion. This boost is applied *after* the α/β damping, so it is not +//! subject to the same scaling. This implementation (following Appendix B) +//! has no such non-linear term; the sole source of `p_increment` is the +//! proportional-integral calculation, damped as described in §4. +//! +//! This is one of the more behaviourally significant differences: under +//! severe congestion (bufferbloat), the kernel will increase its drop +//! probability faster than this implementation. +//! +//! ## 9. Rapid decay condition and factor +//! +//! Both implementations reduce the drop probability when the queue is +//! consistently uncongested, but the *trigger condition* and *decay factor* +//! differ: +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | condition | `qdelay == 0` **and** `qdelay_old == 0` (exactly zero) | `cur_del < ref_del / 2` **and** `old_del < ref_del / 2` | +//! | guard | `update_prob == true` (no overflow/underflow this round) | none | +//! | retention factor | `63/64` ≈ 0.9844 (`prob -= prob / 64`) | `0.98` (`p *= 0.98`) | +//! +//! The kernel only decays when delay is *exactly* zero for two consecutive +//! update periods — a stricter condition. This implementation decays +//! whenever the delay is below half the target (7.5 ms at defaults), which +//! is a looser condition. Combined with the faster decay factor (0.98 vs. +//! 0.9844), this implementation may reduce `p` more aggressively during +//! lightly-loaded periods. +//! +//! ## 10. Burst allowance reduction mechanism +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | reduced in | `pie_process_dequeue()` — every dequeue | `update_drop_probability()` — only when update fires | +//! | reduced by | `dtime` (inter-dequeue interval, in psched ticks) | `elapsed_ms` (`t_update`, in milliseconds) | +//! | floor | 0 (via `max_t(psched_time_t, ...)`) | 0 (via `.max(0.0)`) | +//! | recharge condition | `prob == 0 && delay < target/2 && delay_old < target/2` | `p < EPSILON && cur_del < ref_del/2 && old_del < ref_del/2` | +//! | recharge value | 150 ms (`PSCHED_TICKS_PER_SEC * 150 / 1000`) | `max_burst` (default 150.0 ms) | +//! +//! The kernel reduces `burst_time` on *every* dequeue by the actual +//! inter-departure time, so burst allowance reflects real-time packet +//! departures. This implementation reduces `burst_allowance` only during +//! probability updates, by the `t_update` interval. During periods where +//! updates fire regularly (steady packet arrivals), both behave similarly. +//! During sparse arrivals where updates are caught up in bursts, the +//! reduction granularity may differ. +//! +//! ## 11. Variable reset on sustained good behaviour +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | full variable reset | `pie_vars_init()` when `qdelay < target/2`, `qdelay_old < target/2`, `prob == 0`, and rate estimator has a valid reading | not performed | +//! | effect of reset | clears `prob`, `avg_dq_rate`, `dq_count`, `dq_tstamp`, `accu_prob`; resets `burst_time` to 150 ms | only `burst_allowance` is recharged to `max_burst` | +//! +//! The kernel fully resets all PIE state variables when the queue has been +//! well-behaved for long enough. This implementation only recharges the +//! burst allowance (see §10) and leaves `avg_drate`, `p`, and measurement +//! state intact. After a prolonged idle or light-load period followed by +//! sudden congestion, this implementation may react with a stale (possibly +//! too-low) `avg_drate`, causing a transient over-estimation of delay. +//! +//! ## 12. Delay estimation: always drain-rate-based vs. dual-mode +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | timestamp mode | supported (`dq_rate_estimator = false`) | not supported | +//! | drain-rate mode | supported (`dq_rate_estimator = true`) | always active | +//! | timestamp source | `pie_skb_cb.enqueue_time` from skb control block | N/A | +//! | drain-rate threshold | 16 KiB (`QUEUE_THRESHOLD`) | 16 KiB (`dq_threshold`) | +//! | measurement entry | `backlog >= 16384` | `now_bytes > 16384` | +//! | EWMA weight | `1/8` (0.125) | `0.125` | +//! +//! The kernel can optionally measure delay directly from per-packet enqueue +//! timestamps when the drain-rate estimator is disabled. This +//! implementation always estimates delay as `now_bytes / avg_drate`, +//! equivalent to the kernel's `dq_rate_estimator = true` mode. +//! +//! The measurement cycle entry condition differs by one byte: the kernel +//! uses `>=` while this implementation uses `>`. With standard MTU packets +//! (~1514 B with L2 overhead) this off-by-one is not practically observable. +//! +//! ## 13. Update guard: zero delay with non-zero backlog +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | guard | when `qdelay == 0` but `backlog != 0`, skip this round's probability update | not implemented | +//! +//! The kernel refrains from updating `prob` when the drain-rate estimator +//! yields zero delay but the queue is not actually empty — this indicates +//! the estimator has not yet converged. This implementation always applies +//! the probability update; if `avg_drate ≈ 0` (estimator not yet +//! initialized), `cur_del` is forced to 0 via the `abs(avg_drate) < EPSILON` +//! check. +//! +//! ## 14. Random-number generation +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | source | `get_random_u64()` (kernel CSPRNG) | `rand::rngs::StdRng` (ChaCha12) | +//! | seeding | not seedable by userspace | user-configurable `seed` field | +//! +//! The kernel uses the kernel's CSPRNG for drop decisions. This +//! implementation uses a seedable ChaCha12 RNG, enabling deterministic, +//! reproducible simulation runs. +//! +//! ## 15. Queue architecture +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | structure | full `Qdisc` with netlink configuration | self-contained `VecDeque` | +//! | hard limit | single `limit` (packets) | dual `packet_limit` + `byte_limit` | +//! | statistics | `tc_pie_xstats` (prob, delay, packets_in, dropped, overlimit, maxq, ecn_mark) | none | +//! +//! The kernel's PIE is a full Linux qdisc with netlink-based configuration +//! (`tc qdisc ... pie`), statistics export, and lifecycle management. +//! This implementation is a queue *component* within a larger simulation +//! framework; it stores packets in an internal `VecDeque` and exposes the +//! `PacketQueue` trait. +//! +//! The kernel's single `limit` is a packet count. This implementation adds +//! an independent `byte_limit` for byte-oriented capacity management. +//! +//! ## 16. Configurable parameters +//! +//! The kernel exposes several parameters that have no counterpart here: +//! +//! | Kernel parameter | `TCA_PIE_*` attribute | Purpose | Status here | +//! |------------------|-----------------------|---------|-------------| +//! | `alpha` | `TCA_PIE_ALPHA` | PI controller α gain (0–32) | fixed at 0.125 | +//! | `beta` | `TCA_PIE_BETA` | PI controller β gain (0–32) | fixed at 1.25 | +//! | `ecn` | `TCA_PIE_ECN` | enable ECN marking | not supported | +//! | `bytemode` | `TCA_PIE_BYTEMODE` | scale drop prob by packet size | not supported | +//! | `dq_rate_estimator` | `TCA_PIE_DQ_RATE_ESTIMATOR` | enable drain-rate-based delay estimation | always on | + +use std::collections::VecDeque; + +use rand::{rngs::StdRng, RngExt, SeedableRng}; +#[cfg(feature = "serde")] +use serde::{Deserialize, Serialize}; +use tokio::time::{Duration, Instant}; +use tracing::{debug, warn}; + +#[cfg(feature = "serde")] +use super::serde_default; +use super::{BwType, PacketQueue}; +use crate::cells::Packet; + +/// Configuration for a PIE (Proportional Integral controller Enhanced) queue. +/// +/// # Quick-start configuration +/// +/// The defaults follow RFC 8033 and are suitable for most Internet traffic: +/// +/// - `ref_del = 15 ms` — typical target for wired Internet paths. +/// - `t_update = 15 ms` — RFC-specified update interval. +/// - `max_burst = 150 ms` — allows a burst of roughly 10 back-to-back +/// packets before the drop probability takes effect. +/// +/// For **data-centre** links (sub-millisecond RTT), consider reducing +/// `ref_del` (e.g. 1–5 ms) and `t_update` proportionally. For +/// **satellite** or high-latency links, increase `ref_del` accordingly. +/// +/// # Constructing a config +/// +/// ```no_run +/// # use rattan_core::cells::bandwidth::queue::PieQueueConfig; +/// # use rattan_core::cells::bandwidth::BwType; +/// # use std::time::Duration; +/// // Struct-literal with defaults: +/// let cfg = PieQueueConfig { ref_del: 0.005, ..Default::default() }; +/// +/// // Direct construction: +/// let cfg = PieQueueConfig::new(None, None, 0.005, 100.0, +/// Duration::from_millis(15), +/// BwType::default(), 42); +/// ``` +/// +/// # Field correspondence with the Linux kernel +/// +/// | Field | Kernel equivalent | Notes | +/// |-------|-------------------|-------| +/// | `ref_del` | `pie_params.target` | Kernel stores in µs, converted to psched ticks internally. Here stored as seconds (`f64`). Default: 15 ms (0.015 s). | +/// | `max_burst` | `pie_vars.burst_time` | Kernel initialises to `PSCHED_TICKS_PER_SEC * 150 / 1000` (150 ms in psched ticks). Here stored as milliseconds (`f64`). Default: 150.0 ms. | +/// | `t_update` | `pie_params.tupdate` | Kernel stores in jiffies (`usecs_to_jiffies`). Here stored as `Duration`. Default: 15 ms. | +/// | `packet_limit` | `pie_params.limit` / `sch->limit` | Kernel has a single packet-count limit. Here augmented with an independent `byte_limit`. | +/// | `byte_limit` | *(no direct equivalent)* | Byte-oriented hard limit; not present in the kernel. | +/// | `bw_type` | *(no direct equivalent)* | Configures L2 overhead for bandwidth calculation; simulation-specific. | +/// | `seed` | *(no direct equivalent)* | Deterministic RNG seed; the kernel uses unseedable CSPRNG (`get_random_u64()`). | +/// +/// Kernel parameters **not present** in this struct: `alpha`, `beta` (gains +/// are hard-coded at 0.125 / 1.25), `ecn`, `bytemode`, `dq_rate_estimator`. +/// See the module-level documentation §16 for details. +#[cfg_attr(feature = "serde", derive(Deserialize, Serialize), serde(default))] +#[derive(Debug, Clone)] +pub struct PieQueueConfig { + /// Hard limit on the number of packets in the queue. Packets arriving when + /// the queue already holds this many are dropped unconditionally. + /// + /// `None` means no limit. `Some(0)` makes the queue a zero-buffer (drop + /// everything). + /// + /// Default: `None`. + pub packet_limit: Option, + + /// Hard limit on the total bytes in the queue (L3 length + L2 overhead from + /// [`bw_type`](Self::bw_type)). Same semantics as + /// [`packet_limit`](Self::packet_limit). + /// + /// Default: `None`. + pub byte_limit: Option, + + /// Target queuing delay, in **seconds**. PIE adjusts the drop probability + /// to keep the estimated queue delay near this value. + /// + /// RFC 8033 recommends 15 ms (`0.015`) for wired Internet paths. Reduce + /// for data-centre links (e.g. 0.001–0.005); increase for high-latency + /// links. + /// + /// Valid range: `> 0`, must be finite. + /// + /// Default: `0.015` (15 ms). + pub ref_del: f64, + + /// Maximum burst allowance, in **milliseconds**. + /// + /// When PIE first starts dropping, a burst allowance equal to `max_burst` + /// lets short bursts through without drops. + /// + /// RFC 8033 recommends 150 ms. Reduce for low-RTT environments to make + /// PIE react faster; increase for bursty traffic. + /// + /// Valid range: `≥ 0`, must be finite. + /// + /// Default: `150.0` (150 ms). + pub max_burst: f64, + + /// Interval between drop-probability updates. + /// + /// PIE re-evaluates the drop probability on this cadence. RFC 8033 + /// specifies 15 ms. Shorter intervals make PIE more responsive but + /// increase the update frequency; longer intervals smooth the response. + /// + /// Must be `> Duration::ZERO`. + /// + /// Default: `Duration::from_millis(15)`. + #[cfg_attr(feature = "serde", serde(with = "crate::utils::serde::duration"))] + pub t_update: Duration, + + /// L2 overhead mode for byte accounting. The extra length from + /// [`BwType::extra_length()`] is added to each packet's L3 length when + /// checking [`byte_limit`](Self::byte_limit). + /// + /// Default: [`BwType::NetworkLayer`] (no overhead). + #[cfg_attr( + feature = "serde", + serde(default, skip_serializing_if = "serde_default") + )] + pub bw_type: BwType, + + /// Seed for the deterministic RNG used in drop decisions. Two queues with + /// identical configs, traffic, and seed make identical drop decisions, + /// enabling reproducible simulations. + /// + /// Default: `42`. + #[cfg_attr(feature = "serde", serde(default = "default_pie_seed"))] + pub seed: u64, +} + +impl Default for PieQueueConfig { + fn default() -> Self { + Self { + packet_limit: None, + byte_limit: None, + ref_del: 0.015, // RFC 8033 + max_burst: 150.0, + t_update: Duration::from_millis(15), + bw_type: BwType::default(), + seed: 42, + } + } +} + +#[cfg(feature = "serde")] +const fn default_pie_seed() -> u64 { + 42 +} + +impl PieQueueConfig { + fn validate(&self) -> Result<(), &'static str> { + if self.ref_del <= 0.0 || !self.ref_del.is_finite() { + return Err("PieQueueConfig: ref_del must be > 0 and finite"); + } + if self.max_burst < 0.0 || !self.max_burst.is_finite() { + return Err("PieQueueConfig: max_burst must be >= 0 and finite"); + } + if self.t_update <= Duration::ZERO { + return Err("PieQueueConfig: t_update must be > 0"); + } + Ok(()) + } + + pub fn new>, B: Into>>( + packet_limit: A, + byte_limit: B, + ref_del: f64, + max_burst: f64, + t_update: Duration, + bw_type: BwType, + seed: u64, + ) -> Self { + Self { + packet_limit: packet_limit.into(), + byte_limit: byte_limit.into(), + ref_del, + max_burst, + t_update, + bw_type, + seed, + } + } +} + +impl TryFrom for PieQueue

{ + type Error = &'static str; + + fn try_from(config: PieQueueConfig) -> Result { + PieQueue::new(config) + } +} + +#[derive(Debug)] +pub struct PieQueue

{ + queue: VecDeque

, + config: PieQueueConfig, + now_bytes: usize, + old_del: f64, // previous delay (sec) + p: f64, // current drop probability + dq_count: usize, // departure count (bytes) + start_update: Option, // start time of t_update, set by first enqueue + start_measurement: Option, // Some(Instant) when in a measurement cycle, None when quit + avg_drate: f64, + burst_allowance: f64, + rng: StdRng, +} + +impl

Default for PieQueue

+where + P: Packet, +{ + fn default() -> Self { + Self::new(PieQueueConfig::default()) + .expect("PieQueueConfig::default() should never fail validation") + } +} + +impl

PieQueue

+where + P: Packet, +{ + fn update_drop_probability(&mut self) { + let elapsed_ms = self.config.t_update.as_secs_f64() * 1000.0; + + let cur_del = if self.avg_drate.abs() < f64::EPSILON { + 0.0 + } else { + self.now_bytes as f64 / self.avg_drate + }; + + let tilde_alpha = 0.125; // base value of alpha (Hz, 1/sec) + let tilde_beta = 1.25; // base value of beta (Hz, 1/sec) + let mut p_increment = + tilde_alpha * (cur_del - self.config.ref_del) + tilde_beta * (cur_del - self.old_del); + if self.p < 0.000001 { + p_increment /= 2048.0; + } else if self.p < 0.00001 { + p_increment /= 512.0; + } else if self.p < 0.0001 { + p_increment /= 128.0; + } else if self.p < 0.001 { + p_increment /= 32.0; + } else if self.p < 0.01 { + p_increment /= 8.0; + } else if self.p < 0.1 { + p_increment /= 2.0; + } + + // RFC 8033 Section 5.5: Cap Drop Adjustment + if self.p >= 0.1 { + p_increment = p_increment.min(0.02); + } + + self.p += p_increment; + + // RFC 8033 Section 4.2: Exponential decay when system is not congested + if cur_del < self.config.ref_del / 2.0 && self.old_del < self.config.ref_del / 2.0 { + self.p *= 0.98; + } + self.p = self.p.clamp(0.0, 1.0); + + // RFC 8033 Section 4.4: Burst Tolerance + if self.p < f64::EPSILON + && cur_del < self.config.ref_del / 2.0 + && self.old_del < self.config.ref_del / 2.0 + { + self.burst_allowance = self.config.max_burst; + } else { + self.burst_allowance = (self.burst_allowance - elapsed_ms).max(0.0); + } + self.old_del = cur_del; + self.start_update = Some(self.start_update.unwrap() + self.config.t_update); + } + + fn should_drop(&mut self) -> bool { + // RFC 8033 Section 4.4: Enqueue packet bypassing random drop if burst_allow > 0 + if self.burst_allowance > f64::EPSILON { + return false; + } + + // RFC 8033 Section 4.1: Bypass random drop logic to be work conserving + // MEAN_PKTSIZE is generally considered to be 1500 bytes (standard MTU) in RFCs. + // We add extra_length to align with how now_bytes is calculated (L2 vs L3). + let mean_pktsize = 1500 + self.get_extra_length(); + let bypass_drop = (self.old_del < self.config.ref_del / 2.0 && self.p < 0.2) + || self.now_bytes <= 2 * mean_pktsize; + if bypass_drop { + return false; + } + + let rand_val = self.rng.random_range(0.0..1.0); + rand_val < self.p + } + + fn update_avg_drate(&mut self, pkt_size: usize, now: Instant) { + let dq_threshold = 16384; // 16 KiB + + // Enter a measurement cycle + if self.now_bytes > dq_threshold && self.start_measurement.is_none() { + self.start_measurement = Some(now); + self.dq_count = 0; + } + + // Update departure rate if we are in a measurement cycle + if let Some(start) = self.start_measurement { + self.dq_count += pkt_size; + if self.dq_count >= dq_threshold { + let dq_int = now.saturating_duration_since(start).as_secs_f64(); + if dq_int > f64::EPSILON { + let dq_rate = self.dq_count as f64 / dq_int; + if self.avg_drate.abs() < f64::EPSILON { + self.avg_drate = dq_rate; + } else { + let epsilon = 0.125; + self.avg_drate = (1.0 - epsilon) * self.avg_drate + epsilon * dq_rate; + } + self.start_measurement = None; + self.dq_count = 0; + } + } + + // Exit measurement cycle if queue length drops below threshold + if self.now_bytes < dq_threshold { + self.start_measurement = None; + self.dq_count = 0; + } + } + } +} + +impl

PacketQueue

for PieQueue

+where + P: Packet, +{ + type Config = PieQueueConfig; + + fn new(config: PieQueueConfig) -> Result { + config.validate()?; + debug!(?config, "New PieQueue"); + let max_burst = config.max_burst; + let seed = config.seed; + Ok(Self { + queue: VecDeque::new(), + config, + now_bytes: 0, + old_del: 0.0, + p: 0.0, + dq_count: 0, + start_update: None, + start_measurement: None, + avg_drate: 0.0, + burst_allowance: max_burst, + rng: StdRng::seed_from_u64(seed), + }) + } + + fn configure(&mut self, config: Self::Config) { + if let Err(e) = config.validate() { + warn!("PieQueue: discard invalid configure: {}", e); + return; + } + if config.seed != self.config.seed { + self.rng = StdRng::seed_from_u64(config.seed); + } + self.config = config; + self.burst_allowance = self.burst_allowance.min(self.config.max_burst); + } + + fn is_zero_buffer(&self) -> bool { + self.config.packet_limit.is_some_and(|limit| limit == 0) + || self.config.byte_limit.is_some_and(|limit| limit == 0) + } + + fn enqueue(&mut self, packet: P) { + // Simulate time-driven with event-driven approach by using circular update logic + if self.start_update.is_none() { + self.start_update = Some(packet.get_timestamp()); + } + let mut interval_update = packet + .get_timestamp() + .saturating_duration_since(self.start_update.unwrap()); + while interval_update >= self.config.t_update { + self.update_drop_probability(); + interval_update = packet + .get_timestamp() + .saturating_duration_since(self.start_update.unwrap()); + } + + let packet_size = packet.l3_length() + self.get_extra_length(); + let below_hard_limit = self + .config + .packet_limit + .is_none_or(|limit| self.queue.len() < limit) + && self + .config + .byte_limit + .is_none_or(|limit| self.now_bytes + packet_size <= limit); + + if !below_hard_limit { + #[cfg(test)] + tracing::trace!( + queue_len = self.queue.len(), + now_bytes = self.now_bytes, + header = ?format!("{:X?}", &packet.as_slice()[0..std::cmp::min(56, packet.length())]), + "Drop packet(l3_len: {}, extra_len: {}) due to hard limit", packet.l3_length(), self.get_extra_length() + ); + return; + } + + if self.should_drop() { + #[cfg(test)] + tracing::trace!( + p = self.p, + old_delay = self.old_del, + header = ?format!("{:X?}", &packet.as_slice()[0..std::cmp::min(56, packet.length())]), + "Drop packet(l3_len: {}, extra_len: {}) due to PIE algorithm", packet.l3_length(), self.get_extra_length() + ); + return; + } + self.now_bytes += packet_size; + self.queue.push_back(packet); + } + + fn dequeue_at(&mut self, timestamp: Instant) -> Option

{ + if let Some(packet) = self.queue.pop_front() { + let pkt_size = packet.l3_length() + self.get_extra_length(); + self.now_bytes -= pkt_size; + self.update_avg_drate(pkt_size, timestamp); + Some(packet) + } else { + None + } + } + + fn is_empty(&self) -> bool { + self.queue.is_empty() + } + + #[inline(always)] + fn get_extra_length(&self) -> usize { + self.config.bw_type.extra_length() + } + + fn get_front_size(&self) -> Option { + self.queue + .front() + .map(|packet| self.get_packet_size(packet)) + } + + fn length(&self) -> usize { + self.queue.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cells::StdPacket; + + fn create_packet(size: usize) -> StdPacket { + let buf = vec![0u8; size]; + StdPacket::with_timestamp(&buf, Instant::now()) + } + + #[test_log::test] + fn test_pie_queue_basic() { + let config = PieQueueConfig::default(); + let mut queue: PieQueue = PieQueue::new(config).unwrap(); + assert!(queue.is_empty()); + + let pkt1 = create_packet(500); + queue.enqueue(pkt1); + assert!(!queue.is_empty()); + assert_eq!(queue.length(), 1); + + let dequeued = queue.dequeue_at(Instant::now()); + assert!(dequeued.is_some()); + assert!(queue.is_empty()); + } + + #[test_log::test] + fn test_pie_queue_hard_limit_packet() { + let config = PieQueueConfig { + packet_limit: Some(2), + ..Default::default() + }; + let mut queue: PieQueue = PieQueue::new(config).unwrap(); + + queue.enqueue(create_packet(100)); + queue.enqueue(create_packet(100)); + assert_eq!(queue.length(), 2); + + // This one should be dropped due to packet limit + queue.enqueue(create_packet(100)); + assert_eq!(queue.length(), 2); + } + + #[test_log::test] + fn test_pie_queue_hard_limit_byte() { + let config = PieQueueConfig { + byte_limit: Some(150), + ..Default::default() + }; + let mut queue: PieQueue = PieQueue::new(config).unwrap(); + + queue.enqueue(create_packet(100)); // l3 length 86. + assert_eq!(queue.length(), 1); + + // This one should be dropped due to byte limit (86 + 86 > 150) + queue.enqueue(create_packet(100)); + assert_eq!(queue.length(), 1); + } + + #[test_log::test] + fn test_pie_queue_burst_allowance() { + let config = PieQueueConfig::default(); + let mut queue: PieQueue = PieQueue::new(config).unwrap(); + + // Force a high drop probability + queue.p = 1.0; + queue.burst_allowance = 100.0; + + // burst_allowance > 0 bypasses random drop + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 1); + + // Fill queue > 2 * 1500 bytes to bypass work conserving logic later + queue.enqueue(create_packet(1500)); + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 3); + + queue.burst_allowance = 0.0; + // With burst_allowance = 0.0, queue > 2 * 1500 bytes, and p = 1.0, it should drop + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 3); + } + + #[test_log::test] + fn test_pie_queue_work_conserving() { + let config = PieQueueConfig::default(); + let mut queue: PieQueue = PieQueue::new(config.clone()).unwrap(); + queue.p = 1.0; + queue.burst_allowance = 0.0; + + // bypass_drop handles queue.now_bytes <= 3000 + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 1); + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 2); + + // For the 3rd element, now_bytes is 3000, so it bypasses based on byte length (<= 3000) + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 3); + + // For the 4th element, now_bytes is 4500, so it does not bypass based on byte length + // Since p = 1.0, it drops + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 3); + + // Now test bypass_drop condition: old_del < ref_del/2 and p < 0.2 + queue.p = 0.15; + queue.old_del = config.ref_del / 3.0; + queue.enqueue(create_packet(1500)); + assert_eq!(queue.length(), 4); + } + + #[test_log::test] + fn test_pie_queue_avg_drate_update() { + let config = PieQueueConfig::default(); + let mut queue: PieQueue = PieQueue::new(config).unwrap(); + + queue.enqueue(create_packet(10014)); // l3 length 10000 + queue.enqueue(create_packet(10014)); // l3 length 10000 + queue.enqueue(create_packet(10014)); // l3 length 10000 + assert_eq!(queue.now_bytes, 30000); + + // First dequeue triggers start of measurement cycle + assert!(queue.start_measurement.is_none()); + let deq_time1 = Instant::now(); + queue.dequeue_at(deq_time1); // dequeues 10000 bytes + assert!(queue.start_measurement.is_some()); + assert_eq!(queue.now_bytes, 20000); + + // Simulate time advancing for the next measurement + let mut pkt2 = create_packet(10014); + pkt2.delay_until(deq_time1 + Duration::from_millis(10)); + + // Enqueue the packet with advanced timestamp to update queue state + queue.enqueue(pkt2); + + // Second dequeue triggers calculation of avg_drate + queue.dequeue_at(deq_time1 + Duration::from_millis(10)); // dequeues 10000 bytes + assert!(queue.avg_drate > 0.0, "avg_drate should be calculated"); + assert!( + queue.start_measurement.is_none(), + "Should exit measurement cycle since now_bytes drops below threshold" + ); + } + + #[test_log::test] + fn test_pie_queue_update_drop_probability() { + let config = PieQueueConfig::default(); + let mut queue: PieQueue = PieQueue::new(config.clone()).unwrap(); + + // Fake high delay + queue.avg_drate = 1000.0; + queue.now_bytes = 100000; // cur_del = 100000 / 1000.0 = 100.0s > ref_del (0.015s) + queue.old_del = 50.0; // old_del = 50.0s + queue.p = 0.0; // start with p = 0 + + // Expected p_increment calculation: + // tilde_alpha = 0.125, tilde_beta = 1.25 + // p_increment = 0.125 * (100.0 - 0.015) + 1.25 * (100.0 - 50.0) + // p_increment = 12.498125 + 62.5 = 74.998125 + // Since initial p = 0.0 < 0.000001, p_increment /= 2048.0 + // p_increment = 74.998125 / 2048.0 ≈ 0.036620178 + // Final p should be exactly this value + let expected_p = (0.125 * (100.0 - config.ref_del) + 1.25 * (100.0 - 50.0)) / 2048.0; + + // Force next enqueue to trigger update_drop_probability() deterministically + let mut pkt = create_packet(14); + queue.start_update = Some(pkt.get_timestamp()); + pkt.delay_until(queue.start_update.unwrap() + Duration::from_millis(16)); + + // This enqueue will trigger update_drop_probability() + queue.enqueue(pkt); + + assert!( + (queue.p - expected_p).abs() < f64::EPSILON, + "Probability should be exactly calculated based on PIE formula. Expected: {}, Got: {}", + expected_p, + queue.p + ); + } +} diff --git a/rattan-core/src/cells/bandwidth/queue/red.rs b/rattan-core/src/cells/bandwidth/queue/red.rs new file mode 100644 index 00000000..ba91de85 --- /dev/null +++ b/rattan-core/src/cells/bandwidth/queue/red.rs @@ -0,0 +1,960 @@ +//! RED (Random Early Detection) queue with optional Adaptive mode (ARED). +//! +//! # References +//! +//! - RED paper: +//! - ARED paper: +//! - Linux kernel RED/ARED: +//! [`include/net/red.h`](https://github.com/torvalds/linux/blob/master/include/net/red.h) +//! and [`net/sched/sch_red.c`](https://github.com/torvalds/linux/blob/master/net/sched/sch_red.c) +//! (referenced as "kernel" throughout this documentation) +//! +//! # Differences from the Linux kernel implementation +//! +//! ## 1. Fixed-point integers (kernel) vs. floating-point (here) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | `qavg` | Wlog-scaled integer (`u32`) | `f64` in bytes | +//! | `max_P` | Q0.32 fixed-point (`u32`) | `f64` in `[0.0, 1.0]` | +//! | weight | `Wlog` — weight `W = 1 / (1 << Wlog)` | `w_q` — direct `f64` weight | +//! | division | replaced by `reciprocal_divide()` | native `f64` division | +//! +//! The EWMA update is mathematically equivalent: +//! +//! - Kernel: `qavg += backlog - (qavg >> Wlog)` +//! - Here: `qavg = (1.0 - w_q) * qavg + w_q * backlog` +//! +//! when `w_q = 1.0 / (1 << Wlog)`. +//! +//! ## 2. Idle-period average-queue decay +//! +//! When the queue is empty, both implementations decay the average towards +//! zero by simulating virtual packet departures. The modelling differs: +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | time unit | cell time via `Scell_log` | `pkt_tx_time` (µs per packet) | +//! | decay computation | precomputed `Stab[256]` lookup table | direct `powf(1.0 - w_q, m)` | +//! | idle-time cap | `Scell_max` (max `255 << Scell_log`) | none (exponent may grow arbitrarily) | +//! +//! Both approaches implement the same formula from the original RED paper +//! (Floyd & Jacobson, 1993, §5 "Calculating the average queue length"): +//! +//! > When a packet arrives and the queue is empty, we compute *m*, the number +//! > of packets that could have been transmitted by the gateway during the +//! > time that the line was free. We then imagine that *m* packets have +//! > arrived to an empty queue, and calculate the average queue size: +//! > +//! > **avg ← (1 − w_q)^m × avg** +//! +//! This implementation follows the paper literally: `m` is computed as +//! `idle_us / pkt_tx_time` (where `pkt_tx_time` is the transmission time of +//! one average packet), and decay is `avg *= powf(1.0 - w_q, m)`. The kernel +//! precomputes a logarithmic lookup table (`Stab[]`) indexed by idle duration +//! scaled by `Scell_log` as a fixed-point approximation of `(1 − W)^m`, +//! avoiding both floating-point and the `pow()` call at enqueue time. The +//! two are mathematically equivalent; the kernel trades some precision for +//! integer-only computation. +//! +//! The `pkt_tx_time` parameter has no direct kernel equivalent. The kernel +//! separates idle modelling into `Wlog` (EWMA weight) and `Scell_log` (cell +//! granularity), which are independently configurable. Here, `w_q` controls +//! the decay *rate* and `pkt_tx_time` controls how many virtual departures +//! (`m`) an idle interval represents. +//! +//! ## 3. Random-number generation +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | source | `get_random_u32()` (kernel CSPRNG) | `rand::rngs::StdRng` (ChaCha12) | +//! | seeding | not seedable by userspace | user-configurable `seed` field | +//! | cached? | one `qR` per cycle, reused across packets | fresh `random_range(0.0..1.0)` each check | +//! +//! The kernel draws one random value per "cycle" (from the first packet in the +//! between-threshold region until a mark/drop occurs) and caches it in `qR`. +//! This implementation draws a fresh uniform random value on every +//! `should_drop()` call. Both produce the same statistical behaviour +//! (geometric inter-drop spacing); the difference is purely an implementation +//! choice. +//! +//! The seedable RNG enables deterministic, reproducible simulation runs. +//! +//! ## 4. Drop-probability computation +//! +//! Both implementations realise the same RED probability curve, but through +//! different arithmetic paths: +//! +//! - **Kernel**: `red_mark_probability()` checks +//! `((qavg - qth_min) >> Wlog) * qcount >= qR`, where `qR` is uniform in +//! `[0, qth_delta)`. This is the "uniform random numbers" (URN) method +//! with a cached threshold. +//! +//! - **Here**: Classical formula +//! `p_b = max_p * (avg - min_th) / (max_th - min_th)`, +//! then `p_a = p_b / (1.0 - count * p_b)`, +//! and compare `rand_val < p_a`. +//! +//! Both converge to the same geometric inter-drop distribution. The kernel's +//! approach saves a division per packet via `reciprocal_divide()`; this +//! implementation's approach maps more directly onto the textbook RED +//! description. +//! +//! ## 5. Adaptive RED (ARED) update cadence +//! +//! This is the most behaviourally significant difference. +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | trigger | kernel timer, fires every `HZ/2` (500 ms) | event-driven, on `enqueue()` | +//! | fires when idle? | yes — timer always runs | no — requires a packet arrival | +//! | grid | timer re-arms for `jiffies + HZ/2` | forward-shifted fixed 500 ms grid | +//! +//! **Kernel**: A dedicated timer (`adapt_timer`) fires every 500 ms regardless +//! of whether packets are arriving. Each firing acquires the qdisc lock, +//! calls `red_adaptative_algo()`, and re-arms. During prolonged idle periods +//! the kernel therefore decays `max_P` towards 0.01 every 500 ms. +//! +//! **Here**: ARED adjustment happens inside `enqueue()`. When a packet +//! arrives and at least 500 ms have elapsed since the last adjustment, one +//! call to `update_max_p()` is made and the timer is advanced by exactly 500 +//! ms (not reset to current time). This "fixed grid" approach +//! allows multiple intervals to be caught up over successive packet arrivals +//! after a long idle. However, if no packets arrive, no adjustments occur. +//! +//! The practical impact: after a long idle followed by a sparse trickle of +//! packets, this implementation may converge `max_P` to its resting value more +//! slowly than the kernel, because each enqueue accounts for at most one 500 +//! ms interval. +//! +//! **Why event-driven instead of a dedicated timer?** This simulator operates +//! on *logical* (simulation) time, not wall-clock time. A wall-clock timer +//! cannot accurately target a logical-time instant, especially under +//! variable-speed simulation, pause/resume, or replay scenarios. Worse, +//! spawning an OS timer per queue instance would be prohibitively expensive +//! when simulating hundreds or thousands of flows. +//! +//! ## 6. ARED `max_P` bound enforcement +//! +//! Both clamp `max_P` to `[0.01, 0.50]` (the ARED paper's range), but +//! *when* the bounds are applied differs: +//! +//! - **Kernel**: checks `<= MAX_P_MAX` *before* increasing and `>= MAX_P_MIN` +//! *before* decreasing. No explicit post-adjustment clamp — the value can +//! drift slightly past the nominal bounds in edge cases. +//! - **Here**: always calls `clamp(0.01, 0.5)` unconditionally after every +//! adjustment, enforcing strict hard bounds. +//! +//! The ARED formulae themselves are identical: +//! +//! - Increase: `max_p += min(0.01, max_p / 4.0)` (when `qavg > target_max`) +//! - Decrease: `max_p *= 0.9` (when `qavg < target_min`) +//! - Targets: `target_min = min_th + 0.4 * (max_th - min_th)`, +//! `target_max = min_th + 0.6 * (max_th - min_th)` +//! +//! Note: kernel integer arithmetic truncates in `(max_P / 10) * 9` (beta +//! decay), while here `max_p *= 0.9` is exact in `f64`. The difference is +//! negligible in practice. +//! +//! ## 7. ECN (Explicit Congestion Notification) +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | ECN marking | supported (`TC_RED_ECN` flag) | not supported | +//! | ECN nodrop | supported (`TC_RED_NODROP` flag) | not supported | +//! | mark vs. drop counters | separate `prob_mark`/`prob_drop` | only drops | +//! +//! The kernel can mark ECN-capable packets instead of dropping them. This +//! implementation always drops. Adding ECN support would require extending +//! the `Packet` trait with an ECN field. +//! +//! ## 8. Queue architecture +//! +//! | Aspect | Kernel | This implementation | +//! |--------|--------|---------------------| +//! | structure | classful qdisc → child bfifo | self-contained `VecDeque` | +//! | hard limit | `limit` on child qdisc (bytes) | `packet_limit` + `byte_limit` directly in config | +//! +//! The kernel's RED is a classful qdisc; it owns a child (typically `bfifo`) +//! that holds the actual packet queue. The `limit` parameter lives on the +//! child. This implementation is a flat `VecDeque`-based queue with both +//! packet-count and byte-count hard limits checked before the RED drop +//! decision. +//! +//! ## 9. Parameter validation +//! +//! The kernel validates `fls(qth) + Wlog < 32` (overflow prevention given +//! fixed-point arithmetic), `Scell_log < 32`, and all `Stab[]` entries < 32. +//! This implementation validates the semantically equivalent constraints in +//! floating-point terms: `min_th < max_th`, `w_q ∈ (0.0, 1.0]`, +//! `max_p ∈ [0.0, 1.0]`, and `pkt_tx_time > 0`. +//! +//! The kernel's overflow check (`fls(qth) + Wlog < 32`) is specific to +//! `u32` fixed-point arithmetic where `qavg` is stored Wlog-scaled. With +//! `f64`, overflow is not a concern — the exponent range comfortably covers +//! any practical queue size. The `Scell_log` and `Stab[]` checks are +//! likewise tied to the kernel's lookup-table idle model and have no +//! equivalent here. The current validation set is sufficient for a +//! floating-point RED implementation. +//! +//! ## 10. `qcount` reset value +//! +//! - **Kernel**: resets `qcount = 0` after a probabilistic mark. +//! - **Here**: resets `count_packet = -1` after a drop. +//! +//! Both produce **identical behaviour**. In the kernel, the post-mark +//! sequence is `qcount = 0 → ++qcount = 1 → probability check`, which yields +//! `p_b` for the first packet of the new cycle. Here, the sequence is +//! `count_packet = -1 → count_packet += 1 = 0 → p_a = p_b / (1.0 - 0·p_b) +//! = p_b`, giving the same first-packet probability. Subsequent packets +//! accumulate geometrically in both cases. +//! +//! The value `-1` was chosen deliberately over `0` for **internal +//! consistency**: every path that resets `count_packet` (below `min_th`, +//! above `max_th`, hard-limit drop, and probabilistic drop) uses the same +//! sentinel `-1`. Using a uniform reset value across all branches makes the +//! code easier to reason about and avoids an unnecessary special case. +//! + +use std::collections::VecDeque; + +use rand::{rngs::StdRng, RngExt, SeedableRng}; +#[cfg(feature = "serde")] +use serde::{Deserialize, Serialize}; +use tokio::time::{Duration, Instant}; +use tracing::{debug, warn}; + +#[cfg(feature = "serde")] +use super::serde_default; +use super::{BwType, PacketQueue}; +use crate::cells::Packet; + +/// Configuration for a RED (Random Early Detection) queue. +/// # Quick-start configuration +/// +/// The defaults are sensible for a 100 Mbps link carrying TCP traffic with +/// 1500 B MTU. To adapt them for a different link rate **R** (Mbps): +/// +/// 1. Compute `pkt_tx_time = avpkt × 8 / R` (µs). For 1500 B packets: +/// `pkt_tx_time = 1500 * 8 / R`. At 1 Gbps this is 12 µs; at 10 Mbps, 1200 µs. +/// 2. Set `min_th` and `max_th` in **bytes** to bound the desired queuing +/// delay. With `pkt_tx_time` µs per packet, a threshold of `B` bytes +/// represents roughly `B / avpkt × pkt_tx_time` µs of extra delay. +/// Example: `min_th = 5 × avpkt` and `max_th = 15 × avpkt` (the defaults +/// at avpkt = 1500 B) give ~600 µs of buffer at `min_th` on a 100 Mbps link. +/// 3. `w_q` should be small enough that a single burst doesn't push `avg` past +/// `max_th`, but large enough that `avg` tracks the true queue length +/// within a few RTTs. The default (0.002) means the half-life is +/// `≈ 346 × pkt_tx_time` — about 42 ms at 100 Mbps with 1500 B packets. +/// 4. For most uses, enable `adaptive = true`. +/// +/// # Constructing a config +/// +/// ```no_run +/// # use rattan_core::cells::bandwidth::queue::RedQueueConfig; +/// // Struct-literal with defaults: +/// let cfg = RedQueueConfig { min_th: 5000, ..Default::default() }; +/// +/// // Setter chain: +/// let cfg = RedQueueConfig::default() +/// .with_min_th(5000) +/// .with_max_th(20000) +/// .with_adaptive(true); +/// ``` +/// +/// # Field correspondence with the Linux kernel +/// +/// | Field | Kernel equivalent | Notes | +/// |-------|-------------------|-------| +/// | `min_th` | `red_parms.qth_min` | Kernel stores as `u32`; here as `usize`. Default: 7500 (5 × 1500 B). | +/// | `max_th` | `red_parms.qth_max` | Kernel stores as `u32`; here as `usize`. Default: 22500 (15 × 1500 B). | +/// | `max_p` | `red_parms.max_P` | Kernel stores as Q0.32 fixed-point `u32`; here as `f64` in `[0.0, 1.0]`. Default: 0.02. | +/// | `w_q` | `red_parms.Wlog` | Kernel stores log₂ weight as `u8` (weight = 1/(1<limit` (from `tc_red_qopt.limit`) applied to the child qdisc. | +/// | `byte_limit` | `q->limit` / `tc_red_qopt.limit` | Kernel's limit is byte-oriented and applied to the child qdisc (bfifo); here applied directly. | +/// | `bw_type` | *(no direct equivalent)* | Configures L2 overhead for bandwidth calculation; simulation-specific. | +/// | `seed` | *(no direct equivalent)* | Deterministic RNG seed; the kernel uses unseedable CSPRNG (`get_random_u32()`). | +/// +/// Kernel parameters **not present** in this struct: `Plog` (probability scaling), +/// `Scell_log` (cell-size logarithm), `Stab[]` (precomputed idle-decay table), +/// ECN flags (`TC_RED_ECN`, `TC_RED_NODROP`). See the module-level documentation +/// for details. +#[cfg_attr(feature = "serde", derive(Deserialize, Serialize), serde(default))] +#[derive(Debug, Clone)] +pub struct RedQueueConfig { + /// Hard limit on the number of packets in the queue. Packets arriving when + /// the queue already holds this many are dropped unconditionally. + /// + /// `None` means no limit. `Some(0)` makes the queue a zero-buffer (drop + /// everything). + /// + /// Default: `None`. + pub packet_limit: Option, + + /// Hard limit on the total bytes in the queue (L3 length + L2 overhead from + /// [`bw_type`](Self::bw_type)). Same semantics as + /// [`packet_limit`](Self::packet_limit). + /// + /// Default: `None`. + pub byte_limit: Option, + + /// EWMA weight for the average queue length: + /// `avg = (1 − w_q) · avg + w_q · backlog`. + /// + /// Larger values make `avg` track the instantaneous queue length more + /// closely; smaller values smooth out bursts. The original RED paper + /// recommends `0.002`. + /// + /// Valid range: `(0.0, 1.0]`. `w_q = 1.0` makes `avg` equal the + /// instantaneous backlog (useful for testing). + /// + /// Default: `0.002`. + pub w_q: f64, + + /// Lower threshold for the average queue length, in **bytes**. + /// + /// When `avg < min_th`, no packets are dropped. Should be large enough to + /// absorb transient bursts without drops, but small enough to keep queuing + /// delay acceptable. A common rule of thumb is `5 × avpkt`. + /// + /// Must satisfy `min_th < max_th`. + /// + /// Default: `7500` (5 × 1500). + pub min_th: usize, + + /// Upper threshold for the average queue length, in **bytes**. + /// + /// When `avg ≥ max_th`, every packet is dropped. The original RED paper + /// suggests `max_th ≥ 3 × min_th`. + /// + /// Must satisfy `min_th < max_th`. + /// + /// Default: `22500` (3 × 7500). + pub max_th: usize, + + /// Maximum drop probability in the probabilistic region between `min_th` + /// and `max_th`. Small values mean gentle drop-back; large values mean + /// aggressive dropping. + /// + /// When `adaptive = true` (ARED mode), `max_p` is automatically tuned + /// within `[0.01, 0.5]` — the configured value is just the starting point. + /// + /// Valid range: `[0.0, 1.0]`. + /// + /// Default: `0.02` (2%). + pub max_p: f64, + + /// Transmission time of one average-sized packet, in **microseconds**. + /// + /// Used for idle-period average queue decay. Compute as + /// `avpkt (bytes) × 8 / link_rate (Mbps)`. Example: 1500 B at 100 Mbps → + /// `120 µs`. + /// + /// Valid range: `> 0`. + /// + /// Default: `120.0` (1500 B at 100 Mbps). + pub pkt_tx_time: f64, + + /// Enable Adaptive RED (ARED): `max_p` is dynamically adjusted every 500 ms + /// to keep `avg` within the target band. + /// + /// Enable when you don't know the right `max_p` for your traffic mix; + /// disable for reproducible experiments. + /// + /// Default: `false`. + pub adaptive: bool, + + /// L2 overhead mode for byte accounting. The extra length from + /// [`BwType::extra_length()`] is added to each packet's L3 length when + /// checking [`byte_limit`](Self::byte_limit). + /// + /// Default: [`BwType::NetworkLayer`] (no overhead). + #[cfg_attr( + feature = "serde", + serde(default, skip_serializing_if = "serde_default") + )] + pub bw_type: BwType, + + /// Seed for the deterministic RNG used in drop decisions. Two queues with + /// identical configs, traffic, and seed make identical drop decisions, + /// enabling reproducible simulations. + /// + /// Default: `42`. + #[cfg_attr(feature = "serde", serde(default = "default_red_seed"))] + pub seed: u64, +} + +impl Default for RedQueueConfig { + fn default() -> Self { + Self { + packet_limit: None, + byte_limit: None, + w_q: 0.002, + min_th: 7500, // 5 * 1500 bytes + max_th: 22500, // 15 * 1500 bytes + max_p: 0.02, + pkt_tx_time: 120.0, // 1500 bytes * 8 / 100Mbps = 120 us + adaptive: false, + bw_type: BwType::default(), + seed: 42, + } + } +} + +#[cfg(feature = "serde")] +const fn default_red_seed() -> u64 { + 42 +} + +impl RedQueueConfig { + fn validate(&self) -> Result<(), &'static str> { + if self.min_th >= self.max_th { + return Err("RedQueueConfig: min_th >= max_th"); + } + if self.w_q <= 0.0 || self.w_q > 1.0 || !self.w_q.is_finite() { + return Err("RedQueueConfig: w_q must be in (0.0, 1.0] and finite"); + } + if self.max_p < 0.0 || self.max_p > 1.0 || !self.max_p.is_finite() { + return Err("RedQueueConfig: max_p must be in [0.0, 1.0] and finite"); + } + if self.pkt_tx_time <= 0.0 || !self.pkt_tx_time.is_finite() { + return Err("RedQueueConfig: pkt_tx_time must be > 0 and finite"); + } + Ok(()) + } + + // A `RedQueueConfig` can be constructed in any of these ways: + // 1. Struct literal with defaults: + // RedQueueConfig { min_th: 100, ..Default::default() } + // 2. Setter chain (avoids the old giant `new()` with 9+ args): + // RedQueueConfig::default().with_min_th(100).with_adaptive(true) + // 3. Plain default (all fields take their `Default` values): + // RedQueueConfig::default() + pub fn with_packet_limit(mut self, limit: usize) -> Self { + self.packet_limit = Some(limit); + self + } + + pub fn with_byte_limit(mut self, limit: usize) -> Self { + self.byte_limit = Some(limit); + self + } + + pub fn with_w_q(mut self, w_q: f64) -> Self { + self.w_q = w_q; + self + } + + pub fn with_min_th(mut self, min_th: usize) -> Self { + self.min_th = min_th; + self + } + + pub fn with_max_th(mut self, max_th: usize) -> Self { + self.max_th = max_th; + self + } + + pub fn with_max_p(mut self, max_p: f64) -> Self { + self.max_p = max_p; + self + } + + pub fn with_pkt_tx_time(mut self, pkt_tx_time: f64) -> Self { + self.pkt_tx_time = pkt_tx_time; + self + } + + pub fn with_adaptive(mut self, adaptive: bool) -> Self { + self.adaptive = adaptive; + self + } + + pub fn with_bw_type(mut self, bw_type: BwType) -> Self { + self.bw_type = bw_type; + self + } + + pub fn with_seed(mut self, seed: u64) -> Self { + self.seed = seed; + self + } +} + +impl TryFrom for RedQueue

{ + type Error = &'static str; + + fn try_from(config: RedQueueConfig) -> Result { + RedQueue::new(config) + } +} + +#[derive(Debug)] +pub struct RedQueue

{ + queue: VecDeque

, + config: RedQueueConfig, + now_bytes: usize, + average_queue_length: f64, + count_packet: i32, // number of packets since last dropping + idle_start: Option, // start time of current idle period + latest_max_p_update: Option, // latest time when max_p was updated (used in adaptive mode), set by first enqueue + rng: StdRng, +} + +impl

Default for RedQueue

+where + P: Packet, +{ + fn default() -> Self { + Self::new(RedQueueConfig::default()) + .expect("RedQueueConfig::default() should never fail validation") + } +} + +impl

RedQueue

+where + P: Packet, +{ + fn update_avg(&mut self, packet: &P) { + if !self.is_empty() { + self.average_queue_length = (1.0 - self.config.w_q) * self.average_queue_length + + self.config.w_q * (self.now_bytes as f64); + return; + } + + if let Some(idle_start) = self.idle_start { + let now = packet.get_timestamp(); + let idle_duration = now.saturating_duration_since(idle_start); + let m = idle_duration.as_micros() as f64 / self.config.pkt_tx_time; + self.average_queue_length *= f64::powf(1.0 - self.config.w_q, m); + self.idle_start = Some(now); + } + } + + fn should_drop(&mut self) -> bool { + let avg = self.average_queue_length; + let min_th = self.config.min_th as f64; + let max_th = self.config.max_th as f64; + if avg >= min_th && avg < max_th { + self.count_packet += 1; + let p_b = self.config.max_p * (avg - min_th) / (max_th - min_th); + let p_a = if self.count_packet as f64 * p_b >= 1.0 { + 1.0 + } else { + p_b / (1.0 - self.count_packet as f64 * p_b) + }; + + let rand_val = self.rng.random_range(0.0..1.0); + if rand_val < p_a { + self.count_packet = -1; // first add, then calculate p_a + true + } else { + false + } + } else if avg >= max_th { + self.count_packet = -1; + true + } else { + self.count_packet = -1; + false + } + } + + fn update_max_p(&mut self) { + let target_min = + self.config.min_th as f64 + 0.4 * (self.config.max_th - self.config.min_th) as f64; + let target_max = + self.config.min_th as f64 + 0.6 * (self.config.max_th - self.config.min_th) as f64; + if self.average_queue_length > target_max { + self.config.max_p += (self.config.max_p / 4.0).min(0.01); + } else if self.average_queue_length < target_min { + self.config.max_p *= 0.9; + } + self.config.max_p = self.config.max_p.clamp(0.01, 0.5); + } +} + +impl

PacketQueue

for RedQueue

+where + P: Packet, +{ + type Config = RedQueueConfig; + + fn new(config: RedQueueConfig) -> Result { + config.validate()?; + debug!(?config, "New RedQueue"); + let seed = config.seed; + Ok(Self { + queue: VecDeque::new(), + config, + now_bytes: 0, + average_queue_length: 0.0, + count_packet: -1, + idle_start: None, + latest_max_p_update: None, + rng: StdRng::seed_from_u64(seed), + }) + } + + fn configure(&mut self, config: Self::Config) { + if let Err(e) = config.validate() { + warn!("RedQueue: discard invalid configure: {}", e); + return; + } + if config.seed != self.config.seed { + self.rng = StdRng::seed_from_u64(config.seed); + } + self.config = config; + } + + fn is_zero_buffer(&self) -> bool { + self.config.packet_limit.is_some_and(|limit| limit == 0) + || self.config.byte_limit.is_some_and(|limit| limit == 0) + } + + fn enqueue(&mut self, packet: P) { + self.update_avg(&packet); + + if self.config.adaptive { + let now = packet.get_timestamp(); + if self.latest_max_p_update.is_none() { + self.latest_max_p_update = Some(now); + } + if now.saturating_duration_since(self.latest_max_p_update.unwrap()) + >= Duration::from_millis(500) + { + self.update_max_p(); + // Advance on a fixed 500 ms grid rather than anchoring to the current packet arrival time. + // Anchoring to now would reset the timer on every trigger and silently skip adjustment intervals + // when arrivals are sparse (e.g. one packet after a 2 s idle would fire only once instead of catching up over multiple enqueues). + self.latest_max_p_update = + Some(self.latest_max_p_update.unwrap() + Duration::from_millis(500)); + } + } + + let packet_size = packet.l3_length() + self.get_extra_length(); + let below_hard_limit = self + .config + .packet_limit + .is_none_or(|limit| self.queue.len() < limit) + && self + .config + .byte_limit + .is_none_or(|limit| self.now_bytes + packet_size <= limit); + + if !below_hard_limit { + self.count_packet = -1; + #[cfg(test)] + tracing::trace!( + queue_len = self.queue.len(), + now_bytes = self.now_bytes, + header = ?format!("{:X?}", &packet.as_slice()[0..std::cmp::min(56, packet.length())]), + "Drop packet(l3_len: {}, extra_len: {}) due to hard limit", packet.l3_length(), self.get_extra_length() + ); + return; + } + + if self.should_drop() { + #[cfg(test)] + tracing::trace!( + avg = self.average_queue_length, + count = self.count_packet, + header = ?format!("{:X?}", &packet.as_slice()[0..std::cmp::min(56, packet.length())]), + "Drop packet(l3_len: {}, extra_len: {}) due to RED algorithm", packet.l3_length(), self.get_extra_length() + ); + return; + } + self.now_bytes += packet_size; + self.queue.push_back(packet); + self.idle_start = None; + } + + fn dequeue_at(&mut self, timestamp: Instant) -> Option

{ + if let Some(packet) = self.queue.pop_front() { + self.now_bytes -= packet.l3_length() + self.get_extra_length(); + if self.is_empty() { + self.idle_start = Some(timestamp); + } + Some(packet) + } else { + None + } + } + + fn is_empty(&self) -> bool { + self.queue.is_empty() + } + + #[inline(always)] + fn get_extra_length(&self) -> usize { + self.config.bw_type.extra_length() + } + + fn get_front_size(&self) -> Option { + self.queue + .front() + .map(|packet| self.get_packet_size(packet)) + } + + fn length(&self) -> usize { + self.queue.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cells::StdPacket; + + fn create_packet(size: usize) -> StdPacket { + let buf = vec![0u8; size]; + StdPacket::with_timestamp(&buf, Instant::now()) + } + + #[test_log::test] + fn test_red_queue_basic() { + let config = RedQueueConfig { + min_th: 1000, + max_th: 2000, + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + assert!(queue.is_empty()); + + let pkt1 = create_packet(500); + queue.enqueue(pkt1); + assert!(!queue.is_empty()); + assert_eq!(queue.length(), 1); + + let dequeued = queue.dequeue_at(Instant::now()); + assert!(dequeued.is_some()); + assert!(queue.is_empty()); + } + + #[test_log::test] + fn test_red_queue_hard_limit_packet() { + let config = RedQueueConfig { + packet_limit: Some(2), + min_th: 100000, // avoid red drop + max_th: 200000, + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + queue.enqueue(create_packet(100)); + queue.enqueue(create_packet(100)); + assert_eq!(queue.length(), 2); + + // This one should be dropped due to packet limit + queue.enqueue(create_packet(100)); + assert_eq!(queue.length(), 2); + } + + #[test_log::test] + fn test_red_queue_hard_limit_byte() { + let config = RedQueueConfig { + byte_limit: Some(150), + min_th: 100000, // avoid red drop + max_th: 200000, + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + queue.enqueue(create_packet(100)); // l3 length 86. + assert_eq!(queue.length(), 1); + + // This one should be dropped due to byte limit (86 + 86 > 150) + queue.enqueue(create_packet(100)); + assert_eq!(queue.length(), 1); + } + + #[test_log::test] + fn test_red_queue_max_th_drop() { + let config = RedQueueConfig { + min_th: 100, + max_th: 200, + w_q: 1.0, // max weight, avg matches instantly + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + // First packet + queue.enqueue(create_packet(100)); + + // Second packet + queue.enqueue(create_packet(300)); + + // At this point, queue length is 2, now_bytes is high enough. + // The next enqueue should see average_queue_length > max_th and drop the packet. + let before_len = queue.length(); + queue.enqueue(create_packet(100)); + assert_eq!( + queue.length(), + before_len, + "Packet should be dropped by RED max_th" + ); + } + + #[test_log::test] + fn test_red_queue_min_th_no_drop() { + let config = RedQueueConfig { + min_th: 1000, + max_th: 2000, + w_q: 1.0, // Instantly reach exact byte size + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + // First packet: queue empty, avg remains 0. + queue.enqueue(create_packet(514)); // L3 size = 514 - 14 (Ethernet header) = 500 + assert_eq!(queue.length(), 1); + + // Second packet: queue has 500 bytes. w_q=1.0 makes avg = 500. + // 500 < min_th(1000), so it should not drop. + queue.enqueue(create_packet(414)); // L3 size = 400 + assert_eq!(queue.length(), 2); + + // Check internal state: count_packet is -1 when avg < min_th + assert_eq!(queue.count_packet, -1); + } + + #[test_log::test] + fn test_red_queue_probabilistic_drop() { + let config = RedQueueConfig { + min_th: 100, + max_th: 300, + max_p: 0.5, + w_q: 1.0, + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + // First packet: queue empty, avg = 0. L3 size = 200. + queue.enqueue(create_packet(214)); + assert_eq!(queue.length(), 1); + + let mut drop_count = 0; + let total_packets = 1000; + + for _ in 0..total_packets { + // enqueue packets with L3 size 0 (total size 14). + // now_bytes stays at 200. w_q=1.0 makes avg exactly 200. + // 100 (min_th) <= avg(200) < 300 (max_th), entering probabilistic drop zone. + let before = queue.length(); + queue.enqueue(create_packet(14)); + if queue.length() == before { + drop_count += 1; + } + } + + // Calculate expected drop count based on RED algorithm + // When avg = 200, min_th = 100, max_th = 300, max_p = 0.5 + // p_b = max_p * (avg - min_th) / (max_th - min_th) = 0.5 * 100 / 200 = 0.25 + // Due to the uniform distribution of inter-drop times in RED, + // the expected drop rate is 2 * p_b / (1 + p_b) + // For p_b = 0.25, expected drop rate = 0.5 / 1.25 = 0.4 (40%) + let p_b = 0.25; + let expected_drop_rate = (2.0 * p_b) / (1.0 + p_b); + let expected_drop_count = (total_packets as f64 * expected_drop_rate) as usize; + + // Allow reasonable statistical variation (±3 standard deviations, 99.7% confidence) + // Standard deviation for binomial distribution: sqrt(n * p * (1-p)) + let std_dev = + (total_packets as f64 * expected_drop_rate * (1.0 - expected_drop_rate)).sqrt(); + let margin = (3.0 * std_dev) as usize; + + let lower_bound = expected_drop_count.saturating_sub(margin); + let upper_bound = expected_drop_count + margin; + + // Verify drop count is within expected statistical range + assert!( + drop_count >= lower_bound && drop_count <= upper_bound, + "Drop count {} should be in range [{}, {}] (expected {} ± {} based on RED algorithm with p_b=0.25)", + drop_count, lower_bound, upper_bound, expected_drop_count, margin + ); + + // Keep original assertions as sanity checks + assert!( + drop_count > 0, + "Should have dropped some packets probabilistically" + ); + assert!(drop_count < total_packets, "Should not drop all packets"); + } + + #[test_log::test] + fn test_red_queue_adaptive_max_p_increase() { + let config = RedQueueConfig { + min_th: 100, + max_th: 200, + max_p: 0.02, + w_q: 1.0, // Instantly update avg + adaptive: true, + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + // enqueue to make average_queue_length > target_max + // target_max = min_th + 0.6 * (max_th - min_th) = 100 + 60 = 160 + // We make avg = 180 (L3 size 180, total 194) + queue.enqueue(create_packet(194)); + assert_eq!(queue.average_queue_length, 0.0); // First enqueue updates avg based on empty queue rule (avg=0). + + // Second enqueue updates avg to 180 + queue.enqueue(create_packet(14)); + assert_eq!(queue.average_queue_length, 180.0); + + // Set latest_max_p_update artificially back, then enqueue a packet with current timestamp + let mut pkt3 = create_packet(14); + pkt3.delay_until(queue.latest_max_p_update.unwrap() + Duration::from_millis(600)); + + let before_max_p = queue.config.max_p; + + // Third enqueue triggers update_max_p + queue.enqueue(pkt3); + + let after_max_p = queue.config.max_p; + assert!( + after_max_p > before_max_p, + "max_p should increase when avg > target_max" + ); + } + + #[test_log::test] + fn test_red_queue_adaptive_max_p_decrease() { + let config = RedQueueConfig { + min_th: 100, + max_th: 200, + max_p: 0.05, // Starting with a high max_p + w_q: 1.0, + adaptive: true, + ..Default::default() + }; + let mut queue: RedQueue = RedQueue::new(config).unwrap(); + + // enqueue to make average_queue_length < target_min + // target_min = min_th + 0.4 * (max_th - min_th) = 100 + 40 = 140 + // We make avg = 120 (L3 size 120, total 134) + queue.enqueue(create_packet(134)); + assert_eq!(queue.average_queue_length, 0.0); + + queue.enqueue(create_packet(14)); + assert_eq!(queue.average_queue_length, 120.0); + + // Enqueue a packet with timestamp 600ms later to trigger update_max_p + let mut pkt3 = create_packet(14); + pkt3.delay_until(queue.latest_max_p_update.unwrap() + Duration::from_millis(600)); + + let before_max_p = queue.config.max_p; + + queue.enqueue(pkt3); + + let after_max_p = queue.config.max_p; + assert!( + after_max_p < before_max_p, + "max_p should decrease when avg < target_min" + ); + } +} diff --git a/rattan-core/src/config/bandwidth.rs b/rattan-core/src/config/bandwidth.rs index 80e8642f..5cf2a11c 100644 --- a/rattan-core/src/config/bandwidth.rs +++ b/rattan-core/src/config/bandwidth.rs @@ -35,6 +35,8 @@ pub enum BwCellBuildConfig { DropTail(bandwidth::BwCellConfig>), DropHead(bandwidth::BwCellConfig>), CoDel(bandwidth::BwCellConfig>), + Red(bandwidth::BwCellConfig>), + Pie(bandwidth::BwCellConfig>), } macro_rules! impl_bw_cell_into_factory { @@ -56,7 +58,14 @@ macro_rules! impl_bw_cell_into_factory { }; } -impl_bw_cell_into_factory!(InfiniteQueue, DropTailQueue, DropHeadQueue, CoDelQueue); +impl_bw_cell_into_factory!( + InfiniteQueue, + DropTailQueue, + DropHeadQueue, + CoDelQueue, + RedQueue, + PieQueue +); #[cfg_attr( feature = "serde", @@ -69,6 +78,8 @@ pub enum BwReplayCellBuildConfig { DropTail(BwReplayQueueConfig>), DropHead(BwReplayQueueConfig>), CoDel(BwReplayQueueConfig>), + Red(BwReplayQueueConfig>), + Pie(BwReplayQueueConfig>), } #[cfg_attr( @@ -218,4 +229,11 @@ macro_rules! impl_bw_replay_cell_into_factory { }; } -impl_bw_replay_cell_into_factory!(InfiniteQueue, DropTailQueue, DropHeadQueue, CoDelQueue); +impl_bw_replay_cell_into_factory!( + InfiniteQueue, + DropTailQueue, + DropHeadQueue, + CoDelQueue, + RedQueue, + PieQueue +); diff --git a/rattan-core/src/radix/mod.rs b/rattan-core/src/radix/mod.rs index d01b6f53..34eb1344 100644 --- a/rattan-core/src/radix/mod.rs +++ b/rattan-core/src/radix/mod.rs @@ -348,6 +348,12 @@ where crate::config::BwCellBuildConfig::CoDel(config) => { self.build_cell(id, config.into_factory())?; } + crate::config::BwCellBuildConfig::Red(config) => { + self.build_cell(id, config.into_factory())?; + } + crate::config::BwCellBuildConfig::Pie(config) => { + self.build_cell(id, config.into_factory())?; + } }, CellBuildConfig::BwReplay(bw_replay_config) => match bw_replay_config { crate::config::BwReplayCellBuildConfig::Infinite(config) => { @@ -362,6 +368,12 @@ where crate::config::BwReplayCellBuildConfig::CoDel(config) => { self.build_cell(id, config.into_factory())?; } + crate::config::BwReplayCellBuildConfig::Red(config) => { + self.build_cell(id, config.into_factory())?; + } + crate::config::BwReplayCellBuildConfig::Pie(config) => { + self.build_cell(id, config.into_factory())?; + } }, CellBuildConfig::Delay(config) => { self.build_cell(id, config.into_factory())?; diff --git a/src/channel.rs b/src/channel.rs index c91b89bf..ec091e10 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -5,7 +5,10 @@ use netem_trace::{Bandwidth, Delay}; use paste::paste; use rattan_core::{ cells::bandwidth::{ - queue::{CoDelQueueConfig, DropHeadQueueConfig, DropTailQueueConfig, InfiniteQueueConfig}, + queue::{ + CoDelQueueConfig, DropHeadQueueConfig, DropTailQueueConfig, InfiniteQueueConfig, + PieQueueConfig, RedQueueConfig, + }, BwCellConfig, }, config::{ @@ -92,6 +95,8 @@ enum QueueType { DropTail, DropHead, CoDel, + Red, + Pie, } // Deserialize queue args and create BwCellBuildConfig @@ -170,6 +175,12 @@ impl ChannelArgs { Some(QueueType::CoDel) => { bw_q_args_into_config!(CoDel, self.uplink_queue_args.clone(), bandwidth) } + Some(QueueType::Red) => { + bw_q_args_into_config!(Red, self.uplink_queue_args.clone(), bandwidth) + } + Some(QueueType::Pie) => { + bw_q_args_into_config!(Pie, self.uplink_queue_args.clone(), bandwidth) + } }; uplink_count += 1; cells_config.insert(format!("up_{uplink_count}"), cell_config); @@ -197,6 +208,12 @@ impl ChannelArgs { Some(QueueType::CoDel) => { bwreplay_q_args_into_config!(CoDel, self.uplink_queue_args.clone(), trace_file) } + Some(QueueType::Red) => { + bwreplay_q_args_into_config!(Red, self.uplink_queue_args.clone(), trace_file) + } + Some(QueueType::Pie) => { + bwreplay_q_args_into_config!(Pie, self.uplink_queue_args.clone(), trace_file) + } }; uplink_count += 1; cells_config.insert(format!("up_{uplink_count}"), cell_config); @@ -220,6 +237,12 @@ impl ChannelArgs { Some(QueueType::CoDel) => { bw_q_args_into_config!(CoDel, self.downlink_queue_args.clone(), bandwidth) } + Some(QueueType::Red) => { + bw_q_args_into_config!(Red, self.downlink_queue_args.clone(), bandwidth) + } + Some(QueueType::Pie) => { + bw_q_args_into_config!(Pie, self.downlink_queue_args.clone(), bandwidth) + } }; downlink_count += 1; cells_config.insert(format!("down_{downlink_count}"), cell_config); @@ -251,6 +274,12 @@ impl ChannelArgs { trace_file ) } + Some(QueueType::Red) => { + bwreplay_q_args_into_config!(Red, self.downlink_queue_args.clone(), trace_file) + } + Some(QueueType::Pie) => { + bwreplay_q_args_into_config!(Pie, self.downlink_queue_args.clone(), trace_file) + } }; downlink_count += 1; cells_config.insert(format!("down_{downlink_count}"), cell_config); diff --git a/src/visualize_trace.rs b/src/visualize_trace.rs index ebd3a198..a41c771f 100644 --- a/src/visualize_trace.rs +++ b/src/visualize_trace.rs @@ -261,6 +261,8 @@ pub fn write_visualize_trace( BwCellBuildConfig::Infinite(config) => config.bandwidth, BwCellBuildConfig::DropHead(config) => config.bandwidth, BwCellBuildConfig::DropTail(config) => config.bandwidth, + BwCellBuildConfig::Red(config) => config.bandwidth, + BwCellBuildConfig::Pie(config) => config.bandwidth, } .unwrap_or(Bandwidth::from_bps(u64::MAX)); trace_record.add_cell( @@ -292,6 +294,8 @@ pub fn write_visualize_trace( BwReplayCellBuildConfig::DropHead(config) => config.get_trace(), BwReplayCellBuildConfig::DropTail(config) => config.get_trace(), BwReplayCellBuildConfig::Infinite(config) => config.get_trace(), + BwReplayCellBuildConfig::Red(config) => config.get_trace(), + BwReplayCellBuildConfig::Pie(config) => config.get_trace(), }?; let trace_points = expand_bw_trace(trace.as_mut(), start, end_time)