mirror of
https://github.com/jmcorgan/fips.git
synced 2026-07-30 19:46:15 +00:00
Add ECN congestion signaling and transport congestion detection
Implement hop-by-hop ECN congestion signaling through the FMP layer, transport-level congestion detection via kernel drop counters, and chaos harness integration for end-to-end validation. FMP/session ECN plumbing: - Thread ce_flag parsed at link layer through dispatch_link_message, handle_session_datagram, handle_session_payload, and handle_encrypted_session_msg to session delivery - Replace hardcoded false in session-layer record_recv() with actual ce_flag, activating ecn_ce_count tracking in session MMP ECN congestion detection and CE relay: - Add EcnConfig (node.ecn.*) with configurable loss_threshold (5%) and etx_threshold (3.0) for transit congestion detection - Add send_encrypted_link_message_with_ce() that ORs FLAG_CE into FMP header flags; original method delegates with ce_flag=false - Compute outgoing_ce = incoming_ce || local congestion on next-hop link, enabling hop-by-hop CE relay through transit nodes IPv6 ECN-CE marking: - Mark ECN-CE (0b11) in IPv6 Traffic Class on received DataPackets before TUN delivery when FMP CE flag is set - Only marks ECN-capable packets (ECT(0)/ECT(1)); Not-ECT packets unchanged per RFC 3168 Transport congestion abstraction and UDP kernel drop detection: - Add TransportCongestion struct to transport layer for transport- agnostic local congestion indicators - Replace tokio::UdpSocket with AsyncFd<socket2::Socket> using libc::recvmsg() with ancillary data parsing - Enable SO_RXQ_OVFL for kernel receive buffer drop counter on every packet, wiring up previously-stubbed UdpStats.kernel_drops - Add TransportDropState for per-transport delta tracking with 1s tick sampling via sample_transport_congestion() - Extend detect_congestion() with transport kernel drop check alongside MMP loss metrics Congestion monitoring and control: - Add CongestionStats (ce_forwarded, ce_received, congestion_detected, kernel_drop_events) to NodeStats with snapshot serialization - Wire counters into forwarding path, session handler, and transport drop sampling with rate-limited warn logging (5s interval) - Expose congestion data in show_routing control query and ecn_ce_count in show_mmp peer entries - Add congestion counters to fipstop routing tab in two-column layout Chaos harness integration: - Add query_routing(), query_transports(), snapshot_all_congestion() to chaos control module - Add congestion/kernel-drop log analysis in logs module - Add congestion-stress scenario: 10-node tree, 1 Mbps bandwidth, 5-10% netem loss, heavy iperf3 traffic - Add IngressConfig for tc ingress policing with per-peer policer filters simulating upstream bandwidth bottlenecks - Add iperf3 JSON result capture to traffic manager for throughput measurement across scenarios - Add ECN A/B test scenarios (ecn-ab-on/off.yaml) with ingress policing and comparison script - Enable TCP ECN negotiation (tcp_ecn=1 sysctl) in container entrypoint for end-to-end CE propagation Tests: - 10 ECN unit/integration tests: mark_ipv6_ecn_ce variants, CE relay chain (3-node propagation), EcnConfig serde roundtrip - 3 transport drop congestion detection unit tests Documentation: - Update fips-mesh-layer.md: replace outdated CE Echo stub with full ECN Congestion Signaling section covering detection logic, CE relay, IPv6 marking, session tracking, and monitoring counters - Update fips-configuration.md: add node.ecn.* parameter table and ecn block in complete reference YAML - Update fips-transport-layer.md: add Congestion Reporting section with TransportCongestion struct, congestion() trait method, and per-transport status; document AsyncFd/recvmsg/SO_RXQ_OVFL in UDP - Update chaos README: add congestion/ECN scenario docs, ingress traffic control, and iperf3 JSON capture sections - Update README.md: add ECN to features list and "What works today"; update transport and tooling entries
This commit is contained in:
Johnathan Corgan
@@ -40,6 +40,8 @@ sessions across the mesh.
|
||||
unmodified IP applications
|
||||
- **Metrics Measurement Protocol** — per-link RTT, loss, jitter, and goodput
|
||||
measurement
|
||||
- **ECN congestion signaling** — hop-by-hop CE flag relay with RFC 3168 IPv6
|
||||
marking, transport kernel drop detection
|
||||
- **Operator visibility** — `fipsctl` control socket interface for runtime
|
||||
inspection of peers, links, sessions, tree state, and metrics
|
||||
- **Zero configuration** — sensible defaults; a node can start with no config
|
||||
@@ -187,13 +189,15 @@ UDP/IP overlays but has not been tested beyond small meshes.
|
||||
- Noise IK (link layer) and Noise XK (session layer) encryption
|
||||
- IPv6 TUN adapter with DNS resolution of `.fips` names
|
||||
- Per-link metrics (RTT, loss, jitter, goodput)
|
||||
- Runtime inspection via `fipsctl`
|
||||
- ECN congestion signaling (hop-by-hop CE relay, IPv6 CE marking, kernel drop detection)
|
||||
- UDP, TCP, and Ethernet transports
|
||||
- Runtime inspection via `fipsctl` and `fipstop`
|
||||
- Docker-based integration and chaos testing
|
||||
|
||||
### Near-term priorities
|
||||
|
||||
- Peer discovery via Nostr relays (bootstrap without static peer lists)
|
||||
- Additional transports (Ethernet, Tor)
|
||||
- Additional transports (Bluetooth, Tor)
|
||||
- Improved routing resilience under churn
|
||||
- Security audit of cryptographic protocols
|
||||
- CI pipeline and published crate
|
||||
|
||||
@@ -163,6 +163,25 @@ Controls tree construction and parent selection.
|
||||
Bloom filter size (1 KB), hash count (5), and size classes are protocol
|
||||
constants and not configurable.
|
||||
|
||||
### ECN Signaling (`node.ecn.*`)
|
||||
|
||||
Controls hop-by-hop ECN (Explicit Congestion Notification) signaling. When
|
||||
enabled, transit nodes detect congestion on outgoing links (via MMP loss/ETX
|
||||
metrics or kernel buffer drops) and set the CE flag on forwarded FMP frames.
|
||||
Destination nodes mark ECN-capable IPv6 packets with CE before TUN delivery
|
||||
per RFC 3168, enabling end-host TCP congestion control to react.
|
||||
|
||||
| Parameter | Type | Default | Description |
|
||||
|-----------|------|---------|-------------|
|
||||
| `node.ecn.enabled` | bool | `true` | Enable ECN congestion signaling (CE flag relay and local congestion detection) |
|
||||
| `node.ecn.loss_threshold` | f64 | `0.05` | MMP loss rate threshold for CE marking (0.0–1.0). When the outgoing link's loss rate meets or exceeds this value, forwarded packets are CE-marked. |
|
||||
| `node.ecn.etx_threshold` | f64 | `3.0` | MMP ETX threshold for CE marking (≥1.0). When the outgoing link's ETX meets or exceeds this value, forwarded packets are CE-marked. |
|
||||
|
||||
Congestion detection triggers on any of: outgoing link loss ≥ `loss_threshold`,
|
||||
outgoing link ETX ≥ `etx_threshold`, or kernel receive buffer drops detected on
|
||||
any local transport. CE is relayed hop-by-hop: once set on any hop, the flag
|
||||
stays set for all subsequent hops to the destination.
|
||||
|
||||
### Session / Data Plane (`node.session.*`)
|
||||
|
||||
Controls end-to-end session behavior and packet queuing.
|
||||
@@ -478,6 +497,10 @@ node:
|
||||
mode: full # full | lightweight | minimal
|
||||
log_interval_secs: 30
|
||||
owd_window_size: 32
|
||||
ecn:
|
||||
enabled: true # ECN congestion signaling (CE flag relay)
|
||||
loss_threshold: 0.05 # MMP loss rate threshold for CE marking (5%)
|
||||
etx_threshold: 3.0 # MMP ETX threshold for CE marking
|
||||
control:
|
||||
enabled: true
|
||||
socket_path: null # null = auto ($XDG_RUNTIME_DIR → /run/fips → /tmp fallback)
|
||||
|
||||
@@ -465,12 +465,44 @@ MMP reports on different timers), inter-frame processing delays inflate spin
|
||||
bit RTT measurements unpredictably. Timestamp-echo from ReceiverReports
|
||||
(with dwell-time compensation) is the sole SRTT source.
|
||||
|
||||
### CE Echo
|
||||
### ECN Congestion Signaling
|
||||
|
||||
The CE (Congestion Experienced) echo flag (bit 1 in the FMP flags byte) is
|
||||
reserved for ECN signaling. The transport trait does not currently expose
|
||||
ECN marking, so the CE echo flag is never set. One-way delay trend serves
|
||||
as the sole pre-loss congestion indicator.
|
||||
The CE (Congestion Experienced) flag (bit 1 in the FMP flags byte) provides
|
||||
hop-by-hop congestion signaling through the mesh. Transit nodes detect
|
||||
congestion on outgoing links and set CE on forwarded packets; once set, the
|
||||
flag stays set for all subsequent hops to the destination.
|
||||
|
||||
**Congestion detection** (`detect_congestion()`) triggers on any of:
|
||||
|
||||
- Outgoing link MMP loss rate ≥ `node.ecn.loss_threshold` (default 5%)
|
||||
- Outgoing link MMP ETX ≥ `node.ecn.etx_threshold` (default 3.0)
|
||||
- Kernel receive buffer drops detected on any local transport (via
|
||||
`SO_RXQ_OVFL` on UDP)
|
||||
|
||||
**CE relay**: The forwarding path computes `outgoing_ce = incoming_ce ||
|
||||
local_congestion`. The `send_encrypted_link_message_with_ce()` method ORs
|
||||
`FLAG_CE` into the FMP header flags when ce is true. The original
|
||||
`send_encrypted_link_message()` delegates with `ce_flag=false`, leaving the
|
||||
20+ existing call sites unchanged.
|
||||
|
||||
**IPv6 ECN-CE marking**: When a CE-flagged DataPacket arrives at its final
|
||||
destination, the IPv6 Traffic Class ECN bits are marked CE (0b11) before
|
||||
TUN delivery — but only for ECN-capable packets (ECT(0) or ECT(1)). Not-ECT
|
||||
packets are never marked per RFC 3168. The host TCP stack then echoes ECE in
|
||||
ACKs, triggering sender cwnd reduction through standard congestion control.
|
||||
|
||||
**Session-layer tracking**: The `ecn_ce_count` field in MMP ReceiverReports
|
||||
tracks CE-flagged packets received per link, providing end-to-end visibility
|
||||
into congestion propagation.
|
||||
|
||||
**Monitoring**: `CongestionStats` tracks four counters — `ce_forwarded`,
|
||||
`ce_received`, `congestion_detected`, and `kernel_drop_events` — exposed via
|
||||
`fipsctl show routing` (congestion block) and `fipstop` (routing tab).
|
||||
Rate-limited warn logging (5s interval) alerts on congestion detection events.
|
||||
|
||||
See `node.ecn.*` in
|
||||
[fips-configuration.md](fips-configuration.md#ecn-signaling-nodeecn) for
|
||||
tuning parameters.
|
||||
|
||||
### Operator Logging
|
||||
|
||||
@@ -540,6 +572,7 @@ an attacker sends invalid packets to elicit responses.
|
||||
| Inner header timestamps | **Implemented** |
|
||||
| Path MTU tracking (SessionDatagram) | **Implemented** |
|
||||
| Metrics Measurement Protocol (MMP) | **Implemented** |
|
||||
| ECN congestion signaling (CE relay, IPv6 marking) | **Implemented** |
|
||||
| Rekey with index rotation | Planned |
|
||||
| Allowlist/blocklist | Planned |
|
||||
|
||||
|
||||
@@ -215,8 +215,15 @@ buffer fills in ~2.5 ms; any stall in the async receive loop (decryption,
|
||||
routing, forwarding overhead) causes the kernel to silently drop incoming
|
||||
datagrams.
|
||||
|
||||
FIPS uses the `socket2` crate to configure socket buffers at bind time,
|
||||
before the receive loop starts:
|
||||
FIPS uses `socket2::Socket` wrapped in `tokio::io::unix::AsyncFd` for the
|
||||
UDP receive path. This replaces `tokio::UdpSocket` and enables direct
|
||||
`libc::recvmsg()` calls with ancillary data parsing — specifically the
|
||||
`SO_RXQ_OVFL` socket option, which delivers a cumulative kernel receive
|
||||
buffer drop counter on every received packet. The drop counter feeds into
|
||||
the ECN congestion detection system (see
|
||||
[fips-mesh-layer.md](fips-mesh-layer.md#ecn-congestion-signaling)).
|
||||
|
||||
Socket buffers are configured at bind time via `socket2`:
|
||||
|
||||
| Parameter | Default | Description |
|
||||
| ---------------- | ------- | ------------------------------------ |
|
||||
@@ -486,6 +493,7 @@ start() → lifecycle Bring transport up (bind socket, o
|
||||
stop() → lifecycle Bring transport down
|
||||
send(addr, data) → delivery Send datagram to transport address
|
||||
close_connection(addr)→ () Close a specific connection (no-op for connectionless)
|
||||
congestion() → TransportCongestion Local congestion indicators (optional)
|
||||
discover() → Vec<DiscoveredPeer> Report discovered FIPS endpoints (optional)
|
||||
auto_connect() → bool Auto-connect discovered peers (default: false)
|
||||
accept_connections() → bool Accept inbound handshakes (default: true)
|
||||
@@ -520,6 +528,30 @@ TransportType {
|
||||
|
||||
Predefined types exist for UDP, TCP, Ethernet, WiFi, Tor, and Serial.
|
||||
|
||||
### Congestion Reporting
|
||||
|
||||
Transports optionally report local congestion indicators via a
|
||||
`TransportCongestion` struct, providing a transport-agnostic interface for
|
||||
the node layer's ECN congestion detection:
|
||||
|
||||
```text
|
||||
TransportCongestion {
|
||||
recv_drops: Option<u64> Cumulative kernel-dropped packets (monotonic)
|
||||
}
|
||||
```
|
||||
|
||||
The node samples each transport's congestion state on a 1-second tick via
|
||||
`sample_transport_congestion()`. `TransportDropState` tracks per-transport
|
||||
drop deltas: when new drops appear (rising edge), the `dropping` flag is
|
||||
set, and `detect_congestion()` in the forwarding path triggers CE marking
|
||||
on all forwarded datagrams.
|
||||
|
||||
| Transport | Congestion Source | Mechanism |
|
||||
| --------- | ----------------- | --------- |
|
||||
| UDP | `SO_RXQ_OVFL` kernel drop counter | `recvmsg()` ancillary data on every packet |
|
||||
| TCP | Not yet implemented | Returns `None` (TCP handles congestion internally) |
|
||||
| Ethernet | Not yet implemented | Returns `None` |
|
||||
|
||||
### Transport Addresses
|
||||
|
||||
Transport addresses (`TransportAddr`) are opaque byte vectors. The transport
|
||||
@@ -542,7 +574,7 @@ transitions through `Starting` to `Up` (operational). `stop()` moves to
|
||||
|
||||
| Transport | Status | Notes |
|
||||
| --------- | ------ | ----- |
|
||||
| UDP/IP | **Implemented** | Primary transport, async send/receive, configurable MTU |
|
||||
| UDP/IP | **Implemented** | Primary transport, AsyncFd/recvmsg, SO_RXQ_OVFL kernel drop detection |
|
||||
| TCP/IP | **Implemented** | FMP header-based framing, connect-on-send, per-connection MSS MTU |
|
||||
| Ethernet | **Implemented** | AF_PACKET SOCK_DGRAM, EtherType 0x2121, beacon discovery, Linux only |
|
||||
| WiFi | Future direction | Infrastructure mode = Ethernet driver |
|
||||
|
||||
@@ -78,7 +78,14 @@ fn draw_routing_stats(frame: &mut Frame, data: &serde_json::Value, area: Rect) {
|
||||
let inner = block.inner(area);
|
||||
frame.render_widget(block, area);
|
||||
|
||||
let mut lines = vec![
|
||||
let cols = Layout::horizontal([
|
||||
Constraint::Percentage(50),
|
||||
Constraint::Percentage(50),
|
||||
])
|
||||
.split(inner);
|
||||
|
||||
// Left column: Forwarding + Discovery
|
||||
let mut left = vec![
|
||||
helpers::section_header("Forwarding"),
|
||||
fwd_line(data, "Received", "received_packets", "received_bytes"),
|
||||
fwd_line(data, "Delivered", "delivered_packets", "delivered_bytes"),
|
||||
@@ -109,17 +116,28 @@ fn draw_routing_stats(frame: &mut Frame, data: &serde_json::Value, area: Rect) {
|
||||
helpers::kv_line("Identity Miss", &helpers::nested_u64(data, "discovery", "resp_identity_miss")),
|
||||
helpers::kv_line("Proof Failed", &helpers::nested_u64(data, "discovery", "resp_proof_failed")),
|
||||
helpers::kv_line("Decode Error", &helpers::nested_u64(data, "discovery", "resp_decode_error")),
|
||||
Line::from(""),
|
||||
];
|
||||
|
||||
// Right column: Error Signals + Congestion
|
||||
let mut right = vec![
|
||||
helpers::section_header("Error Signals"),
|
||||
helpers::kv_line("Coords Required", &helpers::nested_u64(data, "error_signals", "coords_required")),
|
||||
helpers::kv_line("Path Broken", &helpers::nested_u64(data, "error_signals", "path_broken")),
|
||||
helpers::kv_line("MTU Exceeded", &helpers::nested_u64(data, "error_signals", "mtu_exceeded")),
|
||||
Line::from(""),
|
||||
helpers::section_header("Congestion"),
|
||||
helpers::kv_line("CE Forwarded", &helpers::nested_u64(data, "congestion", "ce_forwarded")),
|
||||
helpers::kv_line("CE Received", &helpers::nested_u64(data, "congestion", "ce_received")),
|
||||
helpers::kv_line("Congestion Detected", &helpers::nested_u64(data, "congestion", "congestion_detected")),
|
||||
helpers::kv_line("Kernel Drops", &helpers::nested_u64(data, "congestion", "kernel_drop_events")),
|
||||
];
|
||||
|
||||
let max_lines = inner.height as usize;
|
||||
lines.truncate(max_lines);
|
||||
let max_lines = cols[0].height as usize;
|
||||
left.truncate(max_lines);
|
||||
right.truncate(max_lines);
|
||||
|
||||
frame.render_widget(Paragraph::new(lines), inner);
|
||||
frame.render_widget(Paragraph::new(left), cols[0]);
|
||||
frame.render_widget(Paragraph::new(right), cols[1]);
|
||||
}
|
||||
|
||||
fn draw_coord_cache(frame: &mut Frame, app: &App, area: Rect) {
|
||||
|
||||
@@ -414,6 +414,50 @@ impl BuffersConfig {
|
||||
fn default_dns_channel() -> usize { 64 }
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// ECN Congestion Signaling
|
||||
// ============================================================================
|
||||
|
||||
/// ECN congestion signaling configuration (`node.ecn.*`).
|
||||
///
|
||||
/// Controls the FMP CE relay chain: transit nodes detect congestion on outgoing
|
||||
/// links and set the CE flag in forwarded datagrams. The destination marks
|
||||
/// IPv6 ECN-CE on ECN-capable packets before TUN delivery.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct EcnConfig {
|
||||
/// Enable ECN congestion signaling (`node.ecn.enabled`).
|
||||
#[serde(default = "EcnConfig::default_enabled")]
|
||||
pub enabled: bool,
|
||||
|
||||
/// Loss rate threshold for marking CE (`node.ecn.loss_threshold`).
|
||||
/// When the outgoing link's loss rate meets or exceeds this value,
|
||||
/// the transit node sets CE on forwarded datagrams.
|
||||
#[serde(default = "EcnConfig::default_loss_threshold")]
|
||||
pub loss_threshold: f64,
|
||||
|
||||
/// ETX threshold for marking CE (`node.ecn.etx_threshold`).
|
||||
/// When the outgoing link's ETX meets or exceeds this value,
|
||||
/// the transit node sets CE on forwarded datagrams.
|
||||
#[serde(default = "EcnConfig::default_etx_threshold")]
|
||||
pub etx_threshold: f64,
|
||||
}
|
||||
|
||||
impl Default for EcnConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
enabled: true,
|
||||
loss_threshold: 0.05,
|
||||
etx_threshold: 3.0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl EcnConfig {
|
||||
fn default_enabled() -> bool { true }
|
||||
fn default_loss_threshold() -> f64 { 0.05 }
|
||||
fn default_etx_threshold() -> f64 { 3.0 }
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Node Configuration (Root)
|
||||
// ============================================================================
|
||||
@@ -493,6 +537,10 @@ pub struct NodeConfig {
|
||||
/// Metrics Measurement Protocol — session layer (`node.session_mmp.*`).
|
||||
#[serde(default)]
|
||||
pub session_mmp: SessionMmpConfig,
|
||||
|
||||
/// ECN congestion signaling (`node.ecn.*`).
|
||||
#[serde(default)]
|
||||
pub ecn: EcnConfig,
|
||||
}
|
||||
|
||||
impl Default for NodeConfig {
|
||||
@@ -516,6 +564,7 @@ impl Default for NodeConfig {
|
||||
control: ControlConfig::default(),
|
||||
mmp: MmpConfig::default(),
|
||||
session_mmp: SessionMmpConfig::default(),
|
||||
ecn: EcnConfig::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -526,3 +575,35 @@ impl NodeConfig {
|
||||
fn default_heartbeat_interval_secs() -> u64 { 10 }
|
||||
fn default_link_dead_timeout_secs() -> u64 { 30 }
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_ecn_config_defaults() {
|
||||
let c = EcnConfig::default();
|
||||
assert!(c.enabled);
|
||||
assert!((c.loss_threshold - 0.05).abs() < 1e-9);
|
||||
assert!((c.etx_threshold - 3.0).abs() < 1e-9);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_ecn_config_yaml_roundtrip() {
|
||||
let yaml = "loss_threshold: 0.10\netx_threshold: 2.5\nenabled: false\n";
|
||||
let c: EcnConfig = serde_yaml::from_str(yaml).unwrap();
|
||||
assert!(!c.enabled);
|
||||
assert!((c.loss_threshold - 0.10).abs() < 1e-9);
|
||||
assert!((c.etx_threshold - 2.5).abs() < 1e-9);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_ecn_config_partial_yaml() {
|
||||
// Only specify loss_threshold — others should get defaults
|
||||
let yaml = "loss_threshold: 0.02\n";
|
||||
let c: EcnConfig = serde_yaml::from_str(yaml).unwrap();
|
||||
assert!(c.enabled); // default
|
||||
assert!((c.loss_threshold - 0.02).abs() < 1e-9);
|
||||
assert!((c.etx_threshold - 3.0).abs() < 1e-9); // default
|
||||
}
|
||||
}
|
||||
|
||||
@@ -119,11 +119,11 @@ pub fn show_peers(node: &Node) -> Value {
|
||||
if let Some(smoothed_etx) = mmp.metrics.smoothed_etx() {
|
||||
mmp_json["smoothed_etx"] = json!(smoothed_etx);
|
||||
}
|
||||
if let Some(srtt) = mmp.metrics.srtt_ms() {
|
||||
if let Some(setx) = mmp.metrics.smoothed_etx() {
|
||||
if let Some(srtt) = mmp.metrics.srtt_ms()
|
||||
&& let Some(setx) = mmp.metrics.smoothed_etx()
|
||||
{
|
||||
mmp_json["lqi"] = json!(setx * (1.0 + srtt / 100.0));
|
||||
}
|
||||
}
|
||||
peer_json["mmp"] = mmp_json;
|
||||
}
|
||||
|
||||
@@ -265,11 +265,11 @@ pub fn show_sessions(node: &Node) -> Value {
|
||||
if let Some(smoothed_etx) = mmp.metrics.smoothed_etx() {
|
||||
mmp_json["smoothed_etx"] = json!(smoothed_etx);
|
||||
}
|
||||
if let Some(srtt) = mmp.metrics.srtt_ms() {
|
||||
if let Some(setx) = mmp.metrics.smoothed_etx() {
|
||||
if let Some(srtt) = mmp.metrics.srtt_ms()
|
||||
&& let Some(setx) = mmp.metrics.smoothed_etx()
|
||||
{
|
||||
mmp_json["sqi"] = json!(setx * (1.0 + srtt / 100.0));
|
||||
}
|
||||
}
|
||||
session_json["mmp"] = mmp_json;
|
||||
}
|
||||
|
||||
@@ -362,6 +362,7 @@ pub fn show_mmp(node: &Node) -> Value {
|
||||
|
||||
link_layer["delivery_ratio_forward"] = json!(metrics.delivery_ratio_forward);
|
||||
link_layer["delivery_ratio_reverse"] = json!(metrics.delivery_ratio_reverse);
|
||||
link_layer["ecn_ce_count"] = json!(metrics.last_ecn_ce_count());
|
||||
|
||||
Some(json!({
|
||||
"peer": hex::encode(addr.as_bytes()),
|
||||
@@ -487,6 +488,7 @@ pub fn show_routing(node: &Node) -> Value {
|
||||
"forwarding": serde_json::to_value(&node_stats.forwarding).unwrap_or_default(),
|
||||
"discovery": serde_json::to_value(&node_stats.discovery).unwrap_or_default(),
|
||||
"error_signals": serde_json::to_value(&node_stats.errors).unwrap_or_default(),
|
||||
"congestion": serde_json::to_value(&node_stats.congestion).unwrap_or_default(),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -188,6 +188,11 @@ impl MmpMetrics {
|
||||
pub fn goodput_bps(&self) -> f64 {
|
||||
self.goodput_bps
|
||||
}
|
||||
|
||||
/// Cumulative ECN CE count from the most recent ReceiverReport.
|
||||
pub fn last_ecn_ce_count(&self) -> u32 {
|
||||
self.prev_rr_ecn_ce
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for MmpMetrics {
|
||||
|
||||
@@ -360,6 +360,10 @@ impl ReceiverState {
|
||||
pub fn last_recv_time(&self) -> Option<Instant> {
|
||||
self.last_recv_time
|
||||
}
|
||||
|
||||
pub fn ecn_ce_count(&self) -> u32 {
|
||||
self.ecn_ce_count
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ReceiverState {
|
||||
|
||||
@@ -8,7 +8,7 @@ impl Node {
|
||||
/// Dispatch a decrypted link message to the appropriate handler.
|
||||
///
|
||||
/// Link messages are protocol messages exchanged between authenticated peers.
|
||||
pub(in crate::node) async fn dispatch_link_message(&mut self, from: &NodeAddr, plaintext: &[u8]) {
|
||||
pub(in crate::node) async fn dispatch_link_message(&mut self, from: &NodeAddr, plaintext: &[u8], ce_flag: bool) {
|
||||
if plaintext.is_empty() {
|
||||
return;
|
||||
}
|
||||
@@ -19,7 +19,7 @@ impl Node {
|
||||
match msg_type {
|
||||
0x00 => {
|
||||
// SessionDatagram
|
||||
self.handle_session_datagram(from, payload).await;
|
||||
self.handle_session_datagram(from, payload, ce_flag).await;
|
||||
}
|
||||
0x01 => {
|
||||
// SenderReport
|
||||
|
||||
@@ -146,6 +146,6 @@ impl Node {
|
||||
peer.touch(packet.timestamp_ms);
|
||||
|
||||
// Dispatch to link message handler (msg_type + payload, inner header stripped)
|
||||
self.dispatch_link_message(&node_addr, link_message).await;
|
||||
self.dispatch_link_message(&node_addr, link_message, ce_flag).await;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,14 +14,15 @@ use crate::protocol::{
|
||||
CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionSetup,
|
||||
};
|
||||
use crate::NodeAddr;
|
||||
use tracing::debug;
|
||||
use std::time::{Duration, Instant};
|
||||
use tracing::{debug, warn};
|
||||
|
||||
impl Node {
|
||||
/// Handle an incoming SessionDatagram from a peer.
|
||||
///
|
||||
/// Called by `dispatch_link_message` for msg_type 0x00. The payload
|
||||
/// has already had its msg_type byte stripped by dispatch.
|
||||
pub(in crate::node) async fn handle_session_datagram(&mut self, _from: &NodeAddr, payload: &[u8]) {
|
||||
pub(in crate::node) async fn handle_session_datagram(&mut self, _from: &NodeAddr, payload: &[u8], incoming_ce: bool) {
|
||||
self.stats_mut().forwarding.record_received(payload.len());
|
||||
|
||||
let mut datagram = match SessionDatagram::decode(payload) {
|
||||
@@ -50,7 +51,7 @@ impl Node {
|
||||
// Local delivery: dispatch to session layer handlers
|
||||
if datagram.dest_addr == *self.node_addr() {
|
||||
self.stats_mut().forwarding.record_delivered(payload.len());
|
||||
self.handle_session_payload(&datagram.src_addr, &datagram.payload, datagram.path_mtu)
|
||||
self.handle_session_payload(&datagram.src_addr, &datagram.payload, datagram.path_mtu, incoming_ce)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
@@ -77,10 +78,25 @@ impl Node {
|
||||
}
|
||||
}
|
||||
|
||||
// ECN CE relay: propagate incoming CE and detect local congestion
|
||||
let local_congestion = self.detect_congestion(&next_hop_addr);
|
||||
let outgoing_ce = incoming_ce || local_congestion;
|
||||
if local_congestion {
|
||||
self.stats_mut().congestion.record_congestion_detected();
|
||||
let now = Instant::now();
|
||||
let should_log = self.last_congestion_log
|
||||
.map(|t| now.duration_since(t) >= Duration::from_secs(5))
|
||||
.unwrap_or(true);
|
||||
if should_log {
|
||||
self.last_congestion_log = Some(now);
|
||||
warn!(next_hop = %next_hop_addr, "Congestion detected, CE flag set on forwarded packet");
|
||||
}
|
||||
}
|
||||
|
||||
// Forward: re-encode (includes 0x00 type byte) and send
|
||||
let encoded = datagram.encode();
|
||||
if let Err(e) = self
|
||||
.send_encrypted_link_message(&next_hop_addr, &encoded)
|
||||
.send_encrypted_link_message_with_ce(&next_hop_addr, &encoded, outgoing_ce)
|
||||
.await
|
||||
{
|
||||
match e {
|
||||
@@ -100,6 +116,9 @@ impl Node {
|
||||
}
|
||||
} else {
|
||||
self.stats_mut().forwarding.record_forwarded(encoded.len());
|
||||
if outgoing_ce {
|
||||
self.stats_mut().congestion.record_ce_forwarded();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -336,4 +355,55 @@ impl Node {
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Detect congestion for CE marking on forwarded datagrams.
|
||||
///
|
||||
/// Checks two signal sources:
|
||||
/// 1. Outgoing link MMP metrics (loss rate, ETX) against configured thresholds
|
||||
/// 2. Local transport congestion (kernel drops on any transport)
|
||||
///
|
||||
/// Returns `true` if any signal indicates congestion.
|
||||
pub(in crate::node) fn detect_congestion(&self, next_hop: &NodeAddr) -> bool {
|
||||
if !self.config.node.ecn.enabled {
|
||||
return false;
|
||||
}
|
||||
// Outgoing link MMP metrics
|
||||
if let Some(peer) = self.peers.get(next_hop)
|
||||
&& let Some(mmp) = peer.mmp()
|
||||
{
|
||||
let metrics = &mmp.metrics;
|
||||
if metrics.loss_rate() >= self.config.node.ecn.loss_threshold
|
||||
|| metrics.etx >= self.config.node.ecn.etx_threshold
|
||||
{
|
||||
return true;
|
||||
}
|
||||
}
|
||||
// Local transport congestion (kernel drops)
|
||||
self.transport_drops.values().any(|s| s.dropping)
|
||||
}
|
||||
|
||||
/// Sample transport congestion indicators.
|
||||
///
|
||||
/// Called from the tick handler (1s interval). For each transport,
|
||||
/// queries the cumulative kernel drop counter and sets the `dropping`
|
||||
/// flag if new drops occurred since the previous sample.
|
||||
pub(in crate::node) fn sample_transport_congestion(&mut self) {
|
||||
let mut new_drop_events = Vec::new();
|
||||
for (&tid, transport) in &self.transports {
|
||||
let congestion = transport.congestion();
|
||||
let state = self.transport_drops.entry(tid).or_default();
|
||||
if let Some(current) = congestion.recv_drops {
|
||||
let new_drops = current > state.prev_drops;
|
||||
if new_drops && !state.dropping {
|
||||
new_drop_events.push(tid);
|
||||
}
|
||||
state.dropping = new_drops;
|
||||
state.prev_drops = current;
|
||||
}
|
||||
}
|
||||
for tid in new_drop_events {
|
||||
self.stats_mut().congestion.record_kernel_drop_event();
|
||||
warn!(transport_id = tid.as_u32(), "Kernel recv drops first observed on transport");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,5 +7,5 @@ mod forwarding;
|
||||
mod handshake;
|
||||
mod mmp;
|
||||
mod rx_loop;
|
||||
mod session;
|
||||
pub(in crate::node) mod session;
|
||||
mod timeout;
|
||||
|
||||
@@ -118,6 +118,7 @@ impl Node {
|
||||
self.check_link_heartbeats().await;
|
||||
self.purge_stale_lookups(now_ms);
|
||||
self.poll_transport_discovery().await;
|
||||
self.sample_transport_congestion();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,6 +43,7 @@ impl Node {
|
||||
src_addr: &NodeAddr,
|
||||
payload: &[u8],
|
||||
path_mtu: u16,
|
||||
ce_flag: bool,
|
||||
) {
|
||||
let prefix = match FspCommonPrefix::parse(payload) {
|
||||
Some(p) => p,
|
||||
@@ -88,7 +89,7 @@ impl Node {
|
||||
}
|
||||
}
|
||||
FSP_PHASE_ESTABLISHED => {
|
||||
self.handle_encrypted_session_msg(src_addr, payload, path_mtu).await;
|
||||
self.handle_encrypted_session_msg(src_addr, payload, path_mtu, ce_flag).await;
|
||||
}
|
||||
_ => {
|
||||
debug!(phase = prefix.phase, "Unknown FSP phase");
|
||||
@@ -105,7 +106,7 @@ impl Node {
|
||||
/// 4. AEAD decrypt with AAD = header_bytes
|
||||
/// 5. Strip FSP inner header → timestamp, msg_type, inner_flags
|
||||
/// 6. Dispatch by msg_type
|
||||
async fn handle_encrypted_session_msg(&mut self, src_addr: &NodeAddr, payload: &[u8], path_mtu: u16) {
|
||||
async fn handle_encrypted_session_msg(&mut self, src_addr: &NodeAddr, payload: &[u8], path_mtu: u16, ce_flag: bool) {
|
||||
// Parse the 12-byte encrypted header (includes the 4-byte prefix)
|
||||
let header = match FspEncryptedHeader::parse(payload) {
|
||||
Some(h) => h,
|
||||
@@ -209,7 +210,7 @@ impl Node {
|
||||
{
|
||||
let now = std::time::Instant::now();
|
||||
mmp.receiver.record_recv(
|
||||
header.counter, timestamp, plaintext.len(), false, now,
|
||||
header.counter, timestamp, plaintext.len(), ce_flag, now,
|
||||
);
|
||||
// Spin bit: advance state machine for correct TX reflection.
|
||||
// RTT samples not fed into SRTT — timestamp-echo provides
|
||||
@@ -233,8 +234,13 @@ impl Node {
|
||||
match SessionMessageType::from_byte(msg_type) {
|
||||
Some(SessionMessageType::DataPacket) => {
|
||||
// msg_type 0x10: deliver rest (IPv6 payload) to TUN
|
||||
let mut packet = rest.to_vec();
|
||||
if ce_flag {
|
||||
mark_ipv6_ecn_ce(&mut packet);
|
||||
self.stats_mut().congestion.record_ce_received();
|
||||
}
|
||||
if let Some(tun_tx) = &self.tun_tx {
|
||||
if let Err(e) = tun_tx.send(rest.to_vec()) {
|
||||
if let Err(e) = tun_tx.send(packet) {
|
||||
debug!(error = %e, "Failed to deliver decrypted packet to TUN");
|
||||
}
|
||||
} else {
|
||||
@@ -1472,3 +1478,31 @@ impl Node {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Mark ECN-CE in an IPv6 packet's Traffic Class field.
|
||||
///
|
||||
/// IPv6 Traffic Class occupies bits across bytes 0 and 1:
|
||||
/// byte[0] bits[3:0] = TC[7:4]
|
||||
/// byte[1] bits[7:4] = TC[3:0]
|
||||
/// ECN is TC[1:0]. Only marks CE (0b11) if the packet is ECN-capable
|
||||
/// (ECT(0) or ECT(1)). Packets with ECN=0b00 (Not-ECT) are never marked
|
||||
/// per RFC 3168.
|
||||
///
|
||||
/// No checksum update needed: IPv6 has no header checksum, and the Traffic
|
||||
/// Class field is not part of the TCP/UDP pseudo-header.
|
||||
pub(in crate::node) fn mark_ipv6_ecn_ce(packet: &mut [u8]) {
|
||||
if packet.len() < 2 {
|
||||
return;
|
||||
}
|
||||
// Extract 8-bit Traffic Class from IPv6 header bytes 0-1
|
||||
let tc = ((packet[0] & 0x0F) << 4) | (packet[1] >> 4);
|
||||
let ecn = tc & 0x03;
|
||||
// Only mark CE on ECN-capable packets (ECT(0)=0b10 or ECT(1)=0b01)
|
||||
if ecn == 0 {
|
||||
return;
|
||||
}
|
||||
// Set both ECN bits to 1 (CE = 0b11)
|
||||
let new_tc = tc | 0x03;
|
||||
packet[0] = (packet[0] & 0xF0) | (new_tc >> 4);
|
||||
packet[1] = (new_tc << 4) | (packet[1] & 0x0F);
|
||||
}
|
||||
|
||||
+39
-2
@@ -35,7 +35,7 @@ use crate::transport::ethernet::EthernetTransport;
|
||||
use crate::tree::TreeState;
|
||||
use crate::upper::icmp_rate_limit::IcmpRateLimiter;
|
||||
use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx};
|
||||
use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_SP};
|
||||
use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_CE, FLAG_SP};
|
||||
use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity};
|
||||
use rand::Rng;
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
@@ -194,6 +194,18 @@ impl RecentRequest {
|
||||
/// Key for addr_to_link reverse lookup.
|
||||
type AddrKey = (TransportId, TransportAddr);
|
||||
|
||||
/// Per-transport kernel drop tracking for congestion detection.
|
||||
///
|
||||
/// Sampled every tick (1s). The `dropping` flag indicates whether new
|
||||
/// kernel drops were observed since the previous sample.
|
||||
#[derive(Debug, Default)]
|
||||
struct TransportDropState {
|
||||
/// Previous `recv_drops` sample (cumulative counter).
|
||||
prev_drops: u64,
|
||||
/// True if drops increased since the last sample.
|
||||
dropping: bool,
|
||||
}
|
||||
|
||||
/// A running FIPS node instance.
|
||||
///
|
||||
/// This is the top-level container holding all node state.
|
||||
@@ -248,6 +260,8 @@ pub struct Node {
|
||||
// === Transports & Links ===
|
||||
/// Active transports (owned by Node).
|
||||
transports: HashMap<TransportId, TransportHandle>,
|
||||
/// Per-transport kernel drop tracking for congestion detection.
|
||||
transport_drops: HashMap<TransportId, TransportDropState>,
|
||||
/// Active links.
|
||||
links: HashMap<LinkId, Link>,
|
||||
/// Reverse lookup: (transport_id, remote_addr) -> link_id.
|
||||
@@ -358,6 +372,10 @@ pub struct Node {
|
||||
/// Timestamp of last periodic parent re-evaluation (for pacing).
|
||||
last_parent_reeval: Option<std::time::Instant>,
|
||||
|
||||
// === Congestion Logging ===
|
||||
/// Timestamp of last congestion detection log (rate-limited to 5s).
|
||||
last_congestion_log: Option<std::time::Instant>,
|
||||
|
||||
// === Display Names ===
|
||||
/// Human-readable names for configured peers (alias or short npub).
|
||||
/// Populated at startup from peer config.
|
||||
@@ -427,6 +445,7 @@ impl Node {
|
||||
coord_cache,
|
||||
recent_requests: HashMap::new(),
|
||||
transports: HashMap::new(),
|
||||
transport_drops: HashMap::new(),
|
||||
links: HashMap::new(),
|
||||
addr_to_link: HashMap::new(),
|
||||
packet_tx: None,
|
||||
@@ -462,6 +481,7 @@ impl Node {
|
||||
),
|
||||
retry_pending: HashMap::new(),
|
||||
last_parent_reeval: None,
|
||||
last_congestion_log: None,
|
||||
peer_aliases: HashMap::new(),
|
||||
})
|
||||
}
|
||||
@@ -522,6 +542,7 @@ impl Node {
|
||||
coord_cache,
|
||||
recent_requests: HashMap::new(),
|
||||
transports: HashMap::new(),
|
||||
transport_drops: HashMap::new(),
|
||||
links: HashMap::new(),
|
||||
addr_to_link: HashMap::new(),
|
||||
packet_tx: None,
|
||||
@@ -557,6 +578,7 @@ impl Node {
|
||||
),
|
||||
retry_pending: HashMap::new(),
|
||||
last_parent_reeval: None,
|
||||
last_congestion_log: None,
|
||||
peer_aliases: HashMap::new(),
|
||||
}
|
||||
}
|
||||
@@ -1288,6 +1310,18 @@ impl Node {
|
||||
&mut self,
|
||||
node_addr: &NodeAddr,
|
||||
plaintext: &[u8],
|
||||
) -> Result<(), NodeError> {
|
||||
self.send_encrypted_link_message_with_ce(node_addr, plaintext, false).await
|
||||
}
|
||||
|
||||
/// Like `send_encrypted_link_message` but allows setting the FMP CE flag.
|
||||
///
|
||||
/// Used by the forwarding path to relay congestion signals hop-by-hop.
|
||||
pub(super) async fn send_encrypted_link_message_with_ce(
|
||||
&mut self,
|
||||
node_addr: &NodeAddr,
|
||||
plaintext: &[u8],
|
||||
ce_flag: bool,
|
||||
) -> Result<(), NodeError> {
|
||||
let peer = self.peers.get_mut(node_addr)
|
||||
.ok_or(NodeError::PeerNotFound(*node_addr))?;
|
||||
@@ -1312,7 +1346,10 @@ impl Node {
|
||||
let sp_flag = peer.mmp()
|
||||
.map(|mmp| mmp.spin_bit.tx_bit())
|
||||
.unwrap_or(false);
|
||||
let flags = if sp_flag { FLAG_SP } else { 0 };
|
||||
let mut flags = if sp_flag { FLAG_SP } else { 0 };
|
||||
if ce_flag {
|
||||
flags |= FLAG_CE;
|
||||
}
|
||||
|
||||
let session = peer.noise_session_mut().ok_or_else(|| NodeError::SendFailed {
|
||||
node_addr: *node_addr,
|
||||
|
||||
@@ -245,6 +245,46 @@ impl ErrorSignalStats {
|
||||
}
|
||||
}
|
||||
|
||||
/// Congestion event statistics — ECN CE tracking and detection triggers.
|
||||
#[derive(Default)]
|
||||
pub struct CongestionStats {
|
||||
/// Packets forwarded with the CE flag set (incoming CE or locally detected).
|
||||
pub ce_forwarded: u64,
|
||||
/// CE-flagged packets received at this node as final destination.
|
||||
pub ce_received: u64,
|
||||
/// Number of times detect_congestion() returned true.
|
||||
pub congestion_detected: u64,
|
||||
/// Rising-edge transport kernel drop events (not-dropping → dropping).
|
||||
pub kernel_drop_events: u64,
|
||||
}
|
||||
|
||||
impl CongestionStats {
|
||||
pub fn record_ce_forwarded(&mut self) {
|
||||
self.ce_forwarded += 1;
|
||||
}
|
||||
|
||||
pub fn record_ce_received(&mut self) {
|
||||
self.ce_received += 1;
|
||||
}
|
||||
|
||||
pub fn record_congestion_detected(&mut self) {
|
||||
self.congestion_detected += 1;
|
||||
}
|
||||
|
||||
pub fn record_kernel_drop_event(&mut self) {
|
||||
self.kernel_drop_events += 1;
|
||||
}
|
||||
|
||||
pub fn snapshot(&self) -> CongestionStatsSnapshot {
|
||||
CongestionStatsSnapshot {
|
||||
ce_forwarded: self.ce_forwarded,
|
||||
ce_received: self.ce_received,
|
||||
congestion_detected: self.congestion_detected,
|
||||
kernel_drop_events: self.kernel_drop_events,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Aggregate node statistics.
|
||||
#[derive(Default)]
|
||||
pub struct NodeStats {
|
||||
@@ -253,6 +293,7 @@ pub struct NodeStats {
|
||||
pub tree: TreeStats,
|
||||
pub bloom: BloomStats,
|
||||
pub errors: ErrorSignalStats,
|
||||
pub congestion: CongestionStats,
|
||||
}
|
||||
|
||||
impl NodeStats {
|
||||
@@ -267,6 +308,7 @@ impl NodeStats {
|
||||
tree: self.tree.snapshot(),
|
||||
bloom: self.bloom.snapshot(),
|
||||
errors: self.errors.snapshot(),
|
||||
congestion: self.congestion.snapshot(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -356,6 +398,14 @@ pub struct ErrorSignalStatsSnapshot {
|
||||
pub mtu_exceeded: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize)]
|
||||
pub struct CongestionStatsSnapshot {
|
||||
pub ce_forwarded: u64,
|
||||
pub ce_received: u64,
|
||||
pub congestion_detected: u64,
|
||||
pub kernel_drop_events: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize)]
|
||||
pub struct NodeStatsSnapshot {
|
||||
pub forwarding: ForwardingStatsSnapshot,
|
||||
@@ -363,4 +413,5 @@ pub struct NodeStatsSnapshot {
|
||||
pub tree: TreeStatsSnapshot,
|
||||
pub bloom: BloomStatsSnapshot,
|
||||
pub errors: ErrorSignalStatsSnapshot,
|
||||
pub congestion: CongestionStatsSnapshot,
|
||||
}
|
||||
|
||||
@@ -24,7 +24,7 @@ async fn test_forwarding_decode_error() {
|
||||
let mut node = make_node();
|
||||
let from = make_node_addr(0xAA);
|
||||
// Too-short payload: should log error and return without panic
|
||||
node.handle_session_datagram(&from, &[0x00; 5]).await;
|
||||
node.handle_session_datagram(&from, &[0x00; 5], false).await;
|
||||
}
|
||||
|
||||
// --- TTL ---
|
||||
@@ -39,7 +39,7 @@ async fn test_forwarding_hop_limit_exhausted() {
|
||||
.with_ttl(0);
|
||||
let encoded = dg.encode();
|
||||
// Dispatch with payload after msg_type byte
|
||||
node.handle_session_datagram(&from, &encoded[1..]).await;
|
||||
node.handle_session_datagram(&from, &encoded[1..], false).await;
|
||||
// No panic, no send (node has no peers)
|
||||
}
|
||||
|
||||
@@ -56,7 +56,7 @@ async fn test_forwarding_hop_limit_one_drops_at_transit() {
|
||||
.with_ttl(1);
|
||||
let encoded = dg.encode();
|
||||
// Should succeed — ttl=1 decrements to 0 but packet is still processed
|
||||
node.handle_session_datagram(&from, &encoded[1..]).await;
|
||||
node.handle_session_datagram(&from, &encoded[1..], false).await;
|
||||
}
|
||||
|
||||
// --- Local delivery ---
|
||||
@@ -69,7 +69,7 @@ async fn test_forwarding_local_delivery() {
|
||||
let dg = SessionDatagram::new(from, my_addr, vec![0x10, 0x00, 0x00, 0x00]);
|
||||
let encoded = dg.encode();
|
||||
// Should detect local delivery and return without forwarding
|
||||
node.handle_session_datagram(&from, &encoded[1..]).await;
|
||||
node.handle_session_datagram(&from, &encoded[1..], false).await;
|
||||
}
|
||||
|
||||
// --- Direct peer forwarding ---
|
||||
@@ -92,7 +92,7 @@ async fn test_forwarding_direct_peer() {
|
||||
// Handle on node 0: should forward to node 1 (direct peer)
|
||||
nodes[0]
|
||||
.node
|
||||
.handle_session_datagram(&node0_addr, &encoded[1..])
|
||||
.handle_session_datagram(&node0_addr, &encoded[1..], false)
|
||||
.await;
|
||||
|
||||
// Process packets — node 1 should receive the forwarded datagram
|
||||
@@ -135,7 +135,7 @@ async fn test_coord_cache_warming_session_setup() {
|
||||
|
||||
// Handle the datagram (will be local delivery or no-route, but cache warming
|
||||
// happens before routing decision)
|
||||
node.handle_session_datagram(&from, &encoded[1..]).await;
|
||||
node.handle_session_datagram(&from, &encoded[1..], false).await;
|
||||
|
||||
// After: both src and dest coords should be cached
|
||||
let cached_src = node.coord_cache().get(&src_addr, now_ms);
|
||||
@@ -175,7 +175,7 @@ async fn test_coord_cache_warming_session_ack() {
|
||||
assert!(node.coord_cache().get(&src_addr, now_ms).is_none());
|
||||
assert!(node.coord_cache().get(&dest_addr, now_ms).is_none());
|
||||
|
||||
node.handle_session_datagram(&from, &encoded[1..]).await;
|
||||
node.handle_session_datagram(&from, &encoded[1..], false).await;
|
||||
|
||||
// SessionAck caches both src_coords and dest_coords
|
||||
let cached_src = node.coord_cache().get(&src_addr, now_ms);
|
||||
@@ -217,7 +217,7 @@ async fn test_coord_cache_warming_encrypted_msg_with_coords() {
|
||||
assert!(node.coord_cache().get(&src_addr, now_ms).is_none());
|
||||
assert!(node.coord_cache().get(&dest_addr, now_ms).is_none());
|
||||
|
||||
node.handle_session_datagram(&from, &encoded[1..]).await;
|
||||
node.handle_session_datagram(&from, &encoded[1..], false).await;
|
||||
|
||||
assert!(
|
||||
node.coord_cache().get(&src_addr, now_ms).is_some(),
|
||||
@@ -250,7 +250,7 @@ async fn test_coord_cache_warming_encrypted_msg_no_coords() {
|
||||
.unwrap()
|
||||
.as_millis() as u64;
|
||||
|
||||
node.handle_session_datagram(&from, &encoded[1..]).await;
|
||||
node.handle_session_datagram(&from, &encoded[1..], false).await;
|
||||
|
||||
assert!(
|
||||
node.coord_cache().get(&src_addr, now_ms).is_none(),
|
||||
@@ -567,3 +567,202 @@ async fn test_forwarding_with_cache_warming_enables_routing() {
|
||||
|
||||
cleanup_nodes(&mut nodes).await;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// ECN Tests
|
||||
// ============================================================================
|
||||
|
||||
use crate::node::handlers::session::mark_ipv6_ecn_ce;
|
||||
use crate::node::TransportDropState;
|
||||
use crate::transport::TransportId;
|
||||
|
||||
/// Build a minimal IPv6 header (40 bytes) with specified ECN bits.
|
||||
fn make_ipv6_packet_with_ecn(ecn: u8) -> Vec<u8> {
|
||||
let mut pkt = vec![0u8; 40];
|
||||
let tc = ecn; // DSCP=0, ECN=ecn
|
||||
pkt[0] = 0x60 | (tc >> 4);
|
||||
pkt[1] = tc << 4;
|
||||
pkt
|
||||
}
|
||||
|
||||
/// Extract ECN bits from an IPv6 packet.
|
||||
fn read_ecn(pkt: &[u8]) -> u8 {
|
||||
let tc = ((pkt[0] & 0x0F) << 4) | (pkt[1] >> 4);
|
||||
tc & 0x03
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_mark_ecn_ce_on_ect0() {
|
||||
let mut pkt = make_ipv6_packet_with_ecn(0b10);
|
||||
assert_eq!(read_ecn(&pkt), 0b10);
|
||||
mark_ipv6_ecn_ce(&mut pkt);
|
||||
assert_eq!(read_ecn(&pkt), 0b11);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_mark_ecn_ce_on_ect1() {
|
||||
let mut pkt = make_ipv6_packet_with_ecn(0b01);
|
||||
assert_eq!(read_ecn(&pkt), 0b01);
|
||||
mark_ipv6_ecn_ce(&mut pkt);
|
||||
assert_eq!(read_ecn(&pkt), 0b11);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_mark_ecn_ce_on_not_ect() {
|
||||
let mut pkt = make_ipv6_packet_with_ecn(0b00);
|
||||
mark_ipv6_ecn_ce(&mut pkt);
|
||||
assert_eq!(read_ecn(&pkt), 0b00);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_mark_ecn_ce_already_ce() {
|
||||
let mut pkt = make_ipv6_packet_with_ecn(0b11);
|
||||
mark_ipv6_ecn_ce(&mut pkt);
|
||||
assert_eq!(read_ecn(&pkt), 0b11);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_mark_ecn_ce_preserves_dscp_and_flow_label() {
|
||||
let mut pkt = vec![0u8; 40];
|
||||
// DSCP=0b101100 (46=EF), ECN=ECT(0)=0b10 → TC=0xB2
|
||||
let tc: u8 = 0xB2;
|
||||
pkt[0] = 0x60 | (tc >> 4); // 0x6B
|
||||
pkt[1] = (tc << 4) | 0x0A; // 0x2A (flow label high nibble = 0xA)
|
||||
pkt[2] = 0xBC;
|
||||
pkt[3] = 0xDE;
|
||||
|
||||
mark_ipv6_ecn_ce(&mut pkt);
|
||||
|
||||
let new_tc = ((pkt[0] & 0x0F) << 4) | (pkt[1] >> 4);
|
||||
assert_eq!(new_tc, 0xB3, "TC should be 0xB3 (DSCP preserved, ECN=CE)");
|
||||
assert_eq!(pkt[0] >> 4, 6, "Version nibble preserved");
|
||||
assert_eq!(pkt[1] & 0x0F, 0x0A, "Flow label high nibble preserved");
|
||||
assert_eq!(pkt[2], 0xBC, "Flow label byte 2 preserved");
|
||||
assert_eq!(pkt[3], 0xDE, "Flow label byte 3 preserved");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_mark_ecn_ce_short_packet() {
|
||||
let mut pkt = vec![0x60];
|
||||
mark_ipv6_ecn_ce(&mut pkt);
|
||||
assert_eq!(pkt, vec![0x60]);
|
||||
|
||||
let mut empty: Vec<u8> = vec![];
|
||||
mark_ipv6_ecn_ce(&mut empty);
|
||||
assert!(empty.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_ce_relay_through_forwarding() {
|
||||
// 3-node chain: 0 -- 1 -- 2
|
||||
// Send a datagram with CE set from node 0 to node 1.
|
||||
// Node 1 should relay CE to node 2.
|
||||
let edges = vec![(0, 1), (1, 2)];
|
||||
let mut nodes = run_tree_test(3, &edges, false).await;
|
||||
verify_tree_convergence(&nodes);
|
||||
populate_all_coord_caches(&mut nodes);
|
||||
|
||||
let node0_addr = *nodes[0].node.node_addr();
|
||||
let node1_addr = *nodes[1].node.node_addr();
|
||||
let node2_addr = *nodes[2].node.node_addr();
|
||||
|
||||
// Record ecn_ce_count at node 2 before
|
||||
let ce_before = nodes[2]
|
||||
.node
|
||||
.get_peer(&node1_addr)
|
||||
.and_then(|p| p.mmp())
|
||||
.map(|m| m.receiver.ecn_ce_count())
|
||||
.unwrap_or(0);
|
||||
|
||||
// Build a SessionDatagram from node 0 to node 2
|
||||
let dg = SessionDatagram::new(
|
||||
node0_addr,
|
||||
node2_addr,
|
||||
vec![0x10, 0x00, 0x04, 0x00, 1, 2, 3, 4],
|
||||
);
|
||||
let encoded = dg.encode();
|
||||
|
||||
// Send from node 0 to node 1 with CE flag set
|
||||
nodes[0]
|
||||
.node
|
||||
.send_encrypted_link_message_with_ce(&node1_addr, &encoded, true)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Process: node 1 receives (CE set), forwards to node 2 (CE relayed)
|
||||
for _ in 0..3 {
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
process_available_packets(&mut nodes).await;
|
||||
}
|
||||
|
||||
// Node 2's link-layer MMP should have received a CE-flagged frame from node 1
|
||||
let ce_after = nodes[2]
|
||||
.node
|
||||
.get_peer(&node1_addr)
|
||||
.and_then(|p| p.mmp())
|
||||
.map(|m| m.receiver.ecn_ce_count())
|
||||
.unwrap_or(0);
|
||||
|
||||
assert!(
|
||||
ce_after > ce_before,
|
||||
"Node 2 should see CE flag relayed from node 1 (before={ce_before}, after={ce_after})"
|
||||
);
|
||||
|
||||
cleanup_nodes(&mut nodes).await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_congestion_with_transport_drops() {
|
||||
let mut node = make_node();
|
||||
|
||||
// No drops — detect_congestion should return false for any address
|
||||
let fake_addr = NodeAddr::from_bytes([1; 16]);
|
||||
assert!(!node.detect_congestion(&fake_addr));
|
||||
|
||||
// Simulate transport kernel drops
|
||||
let tid = TransportId::new(1);
|
||||
node.transport_drops.insert(tid, TransportDropState {
|
||||
prev_drops: 100,
|
||||
dropping: true,
|
||||
});
|
||||
|
||||
// Now detect_congestion should return true (local transport congestion)
|
||||
assert!(node.detect_congestion(&fake_addr));
|
||||
|
||||
// Clear the dropping flag — should return false again
|
||||
node.transport_drops.get_mut(&tid).unwrap().dropping = false;
|
||||
assert!(!node.detect_congestion(&fake_addr));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_congestion_disabled_ecn() {
|
||||
let mut node = make_node();
|
||||
node.config.node.ecn.enabled = false;
|
||||
|
||||
// Even with transport drops, disabled ECN should return false
|
||||
let tid = TransportId::new(1);
|
||||
node.transport_drops.insert(tid, TransportDropState {
|
||||
prev_drops: 50,
|
||||
dropping: true,
|
||||
});
|
||||
|
||||
let fake_addr = NodeAddr::from_bytes([1; 16]);
|
||||
assert!(!node.detect_congestion(&fake_addr));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_sample_transport_congestion() {
|
||||
let mut node = make_node();
|
||||
|
||||
// Insert a transport drop state with a baseline
|
||||
let tid = TransportId::new(1);
|
||||
node.transport_drops.insert(tid, TransportDropState {
|
||||
prev_drops: 0,
|
||||
dropping: false,
|
||||
});
|
||||
|
||||
// No transports registered — sample_transport_congestion is a no-op
|
||||
// (transport_drops entry stays unchanged)
|
||||
node.sample_transport_congestion();
|
||||
assert!(!node.transport_drops[&tid].dropping);
|
||||
}
|
||||
|
||||
@@ -791,6 +791,21 @@ pub trait Transport {
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Transport Congestion
|
||||
// ============================================================================
|
||||
|
||||
/// Transport-local congestion indicators.
|
||||
///
|
||||
/// All fields are optional — transports report what they can.
|
||||
/// Consumers compute deltas from cumulative counters.
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct TransportCongestion {
|
||||
/// Cumulative packets dropped by kernel/OS before reaching the application.
|
||||
/// Monotonically increasing since transport start.
|
||||
pub recv_drops: Option<u64>,
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Transport Handle
|
||||
// ============================================================================
|
||||
@@ -971,6 +986,20 @@ impl TransportHandle {
|
||||
self.state().is_operational()
|
||||
}
|
||||
|
||||
/// Query transport-local congestion indicators.
|
||||
///
|
||||
/// Returns a snapshot of congestion signals that the transport can
|
||||
/// observe locally (e.g., kernel receive buffer drops). Fields are
|
||||
/// `None` when the transport doesn't support that signal.
|
||||
pub fn congestion(&self) -> TransportCongestion {
|
||||
match self {
|
||||
TransportHandle::Udp(t) => t.congestion(),
|
||||
#[cfg(target_os = "linux")]
|
||||
TransportHandle::Ethernet(_) => TransportCongestion::default(),
|
||||
TransportHandle::Tcp(_) => TransportCongestion::default(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Get transport-specific stats as a JSON value.
|
||||
///
|
||||
/// Returns a snapshot of counters for the specific transport type.
|
||||
|
||||
+38
-61
@@ -6,13 +6,13 @@ use super::{
|
||||
DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError,
|
||||
TransportId, TransportState, TransportType,
|
||||
};
|
||||
mod socket;
|
||||
mod stats;
|
||||
use socket::{AsyncUdpSocket, UdpRawSocket};
|
||||
use stats::UdpStats;
|
||||
use crate::config::UdpConfig;
|
||||
use socket2::{Domain, Protocol, Socket, Type};
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use tokio::net::UdpSocket;
|
||||
use tokio::task::JoinHandle;
|
||||
use tracing::{debug, info, trace, warn};
|
||||
|
||||
@@ -31,7 +31,7 @@ pub struct UdpTransport {
|
||||
/// Current state.
|
||||
state: TransportState,
|
||||
/// Bound socket (None until started).
|
||||
socket: Option<Arc<UdpSocket>>,
|
||||
socket: Option<AsyncUdpSocket>,
|
||||
/// Channel for delivering received packets to Node.
|
||||
packet_tx: PacketTx,
|
||||
/// Receive loop task handle.
|
||||
@@ -73,16 +73,18 @@ impl UdpTransport {
|
||||
self.local_addr
|
||||
}
|
||||
|
||||
/// Get a reference to the socket (only valid after start).
|
||||
pub fn socket(&self) -> Option<&Arc<UdpSocket>> {
|
||||
self.socket.as_ref()
|
||||
}
|
||||
|
||||
/// Get the transport statistics.
|
||||
pub fn stats(&self) -> &Arc<UdpStats> {
|
||||
&self.stats
|
||||
}
|
||||
|
||||
/// Query transport-local congestion indicators.
|
||||
pub fn congestion(&self) -> super::TransportCongestion {
|
||||
super::TransportCongestion {
|
||||
recv_drops: Some(self.stats.kernel_drops.load(std::sync::atomic::Ordering::Relaxed)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Start the transport asynchronously.
|
||||
///
|
||||
/// Binds the UDP socket and spawns the receive loop.
|
||||
@@ -100,56 +102,20 @@ impl UdpTransport {
|
||||
.parse()
|
||||
.map_err(|e| TransportError::StartFailed(format!("invalid bind address: {}", e)))?;
|
||||
|
||||
// Create socket via socket2 for buffer size control
|
||||
let domain = if bind_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
|
||||
let sock2 = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))
|
||||
.map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?;
|
||||
sock2.set_nonblocking(true)
|
||||
.map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?;
|
||||
sock2.bind(&bind_addr.into())
|
||||
.map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?;
|
||||
// Create, bind, and configure UDP socket
|
||||
let raw_socket = UdpRawSocket::open(
|
||||
bind_addr,
|
||||
self.config.recv_buf_size(),
|
||||
self.config.send_buf_size(),
|
||||
)?;
|
||||
|
||||
// Set socket buffer sizes
|
||||
let recv_buf = self.config.recv_buf_size();
|
||||
let send_buf = self.config.send_buf_size();
|
||||
sock2.set_recv_buffer_size(recv_buf)
|
||||
.map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?;
|
||||
sock2.set_send_buffer_size(send_buf)
|
||||
.map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?;
|
||||
let actual_recv = raw_socket.recv_buffer_size()?;
|
||||
let actual_send = raw_socket.send_buffer_size()?;
|
||||
self.local_addr = Some(raw_socket.local_addr());
|
||||
|
||||
let actual_recv = sock2.recv_buffer_size()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?;
|
||||
let actual_send = sock2.send_buffer_size()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?;
|
||||
|
||||
if actual_recv < recv_buf {
|
||||
warn!(
|
||||
requested = recv_buf,
|
||||
actual = actual_recv,
|
||||
"UDP recv buffer clamped by kernel (increase net.core.rmem_max)"
|
||||
);
|
||||
}
|
||||
if actual_send < send_buf {
|
||||
warn!(
|
||||
requested = send_buf,
|
||||
actual = actual_send,
|
||||
"UDP send buffer clamped by kernel (increase net.core.wmem_max)"
|
||||
);
|
||||
}
|
||||
|
||||
// Convert to tokio UdpSocket
|
||||
let std_socket: std::net::UdpSocket = sock2.into();
|
||||
let socket = UdpSocket::from_std(std_socket)
|
||||
.map_err(|e| TransportError::StartFailed(format!("tokio socket failed: {}", e)))?;
|
||||
|
||||
self.local_addr = Some(
|
||||
socket
|
||||
.local_addr()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))?,
|
||||
);
|
||||
|
||||
let socket = Arc::new(socket);
|
||||
self.socket = Some(socket.clone());
|
||||
// Wrap in AsyncFd for tokio integration
|
||||
let async_socket = raw_socket.into_async()?;
|
||||
self.socket = Some(async_socket.clone());
|
||||
|
||||
// Spawn receive loop
|
||||
let transport_id = self.transport_id;
|
||||
@@ -158,7 +124,7 @@ impl UdpTransport {
|
||||
let stats = self.stats.clone();
|
||||
|
||||
let recv_task = tokio::spawn(async move {
|
||||
udp_receive_loop(socket, transport_id, packet_tx, mtu, stats).await;
|
||||
udp_receive_loop(async_socket, transport_id, packet_tx, mtu, stats).await;
|
||||
});
|
||||
|
||||
self.recv_task = Some(recv_task);
|
||||
@@ -231,7 +197,7 @@ impl UdpTransport {
|
||||
let socket_addr = parse_socket_addr(addr)?;
|
||||
let socket = self.socket.as_ref().ok_or(TransportError::NotStarted)?;
|
||||
|
||||
match socket.send_to(data, socket_addr).await {
|
||||
match socket.send_to(data, &socket_addr).await {
|
||||
Ok(bytes_sent) => {
|
||||
self.stats.record_send(bytes_sent);
|
||||
trace!(
|
||||
@@ -244,7 +210,7 @@ impl UdpTransport {
|
||||
}
|
||||
Err(e) => {
|
||||
self.stats.record_send_error();
|
||||
Err(TransportError::SendFailed(format!("{}", e)))
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -305,7 +271,7 @@ fn parse_socket_addr(addr: &TransportAddr) -> Result<SocketAddr, TransportError>
|
||||
|
||||
/// UDP receive loop - runs as a spawned task.
|
||||
async fn udp_receive_loop(
|
||||
socket: Arc<UdpSocket>,
|
||||
socket: AsyncUdpSocket,
|
||||
transport_id: TransportId,
|
||||
packet_tx: PacketTx,
|
||||
mtu: u16,
|
||||
@@ -318,8 +284,9 @@ async fn udp_receive_loop(
|
||||
|
||||
loop {
|
||||
match socket.recv_from(&mut buf).await {
|
||||
Ok((len, remote_addr)) => {
|
||||
Ok((len, remote_addr, kernel_drops)) => {
|
||||
stats.record_recv(len);
|
||||
stats.set_kernel_drops(kernel_drops as u64);
|
||||
|
||||
let data = buf[..len].to_vec();
|
||||
let addr = TransportAddr::from_string(&remote_addr.to_string());
|
||||
@@ -573,4 +540,14 @@ mod tests {
|
||||
let binary = TransportAddr::new(vec![0xff, 0x80]);
|
||||
assert!(parse_socket_addr(&binary).is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_congestion_reports_kernel_drops() {
|
||||
let (tx, _rx) = packet_channel(100);
|
||||
let transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx);
|
||||
|
||||
// Before start, congestion should still report (from stats)
|
||||
let cong = transport.congestion();
|
||||
assert_eq!(cong.recv_drops, Some(0));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,288 @@
|
||||
//! UDP socket wrapper with `SO_RXQ_OVFL` kernel drop counter support.
|
||||
//!
|
||||
//! Provides sync send/recv operations over a `socket2::Socket` with
|
||||
//! `recvmsg()` ancillary data parsing for the kernel receive buffer
|
||||
//! drop counter. The async wrapper uses `tokio::io::unix::AsyncFd`
|
||||
//! for integration with the tokio runtime.
|
||||
//!
|
||||
//! Follows the pattern established by `transport/ethernet/socket.rs`.
|
||||
|
||||
use crate::transport::TransportError;
|
||||
use socket2::{Domain, Protocol, Socket, Type};
|
||||
use std::net::SocketAddr;
|
||||
use std::os::unix::io::{AsRawFd, RawFd};
|
||||
use std::sync::Arc;
|
||||
use tokio::io::unix::AsyncFd;
|
||||
use tracing::warn;
|
||||
|
||||
/// Wrapper around a `socket2::Socket` providing sync send/recv with
|
||||
/// `SO_RXQ_OVFL` ancillary data parsing.
|
||||
pub struct UdpRawSocket {
|
||||
inner: Socket,
|
||||
local_addr: SocketAddr,
|
||||
}
|
||||
|
||||
impl UdpRawSocket {
|
||||
/// Create, bind, and configure a UDP socket.
|
||||
///
|
||||
/// Enables `SO_RXQ_OVFL` for kernel drop counting (non-fatal if
|
||||
/// unsupported). Sets non-blocking mode for async integration.
|
||||
pub fn open(
|
||||
bind_addr: SocketAddr,
|
||||
recv_buf_size: usize,
|
||||
send_buf_size: usize,
|
||||
) -> Result<Self, TransportError> {
|
||||
let domain = if bind_addr.is_ipv4() {
|
||||
Domain::IPV4
|
||||
} else {
|
||||
Domain::IPV6
|
||||
};
|
||||
let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))
|
||||
.map_err(|e| TransportError::StartFailed(format!("socket create failed: {}", e)))?;
|
||||
|
||||
sock.set_nonblocking(true)
|
||||
.map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?;
|
||||
|
||||
sock.bind(&bind_addr.into())
|
||||
.map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?;
|
||||
|
||||
// Set socket buffer sizes
|
||||
sock.set_recv_buffer_size(recv_buf_size)
|
||||
.map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?;
|
||||
sock.set_send_buffer_size(send_buf_size)
|
||||
.map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?;
|
||||
|
||||
let actual_recv = sock
|
||||
.recv_buffer_size()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?;
|
||||
let actual_send = sock
|
||||
.send_buffer_size()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?;
|
||||
|
||||
if actual_recv < recv_buf_size {
|
||||
warn!(
|
||||
requested = recv_buf_size,
|
||||
actual = actual_recv,
|
||||
"UDP recv buffer clamped by kernel (increase net.core.rmem_max)"
|
||||
);
|
||||
}
|
||||
if actual_send < send_buf_size {
|
||||
warn!(
|
||||
requested = send_buf_size,
|
||||
actual = actual_send,
|
||||
"UDP send buffer clamped by kernel (increase net.core.wmem_max)"
|
||||
);
|
||||
}
|
||||
|
||||
// Enable SO_RXQ_OVFL for kernel drop counter in recvmsg ancillary data.
|
||||
// Non-fatal: older kernels or non-Linux platforms may not support it.
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
let enable: libc::c_int = 1;
|
||||
let ret = unsafe {
|
||||
libc::setsockopt(
|
||||
sock.as_raw_fd(),
|
||||
libc::SOL_SOCKET,
|
||||
libc::SO_RXQ_OVFL,
|
||||
&enable as *const _ as *const libc::c_void,
|
||||
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
|
||||
)
|
||||
};
|
||||
if ret < 0 {
|
||||
warn!(
|
||||
"setsockopt(SO_RXQ_OVFL) failed: {}",
|
||||
std::io::Error::last_os_error()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let local_addr = sock
|
||||
.local_addr()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))?
|
||||
.as_socket()
|
||||
.ok_or_else(|| {
|
||||
TransportError::StartFailed("local address is not an IP socket".into())
|
||||
})?;
|
||||
|
||||
Ok(Self {
|
||||
inner: sock,
|
||||
local_addr,
|
||||
})
|
||||
}
|
||||
|
||||
/// Get the local bound address.
|
||||
pub fn local_addr(&self) -> SocketAddr {
|
||||
self.local_addr
|
||||
}
|
||||
|
||||
/// Get the actual receive buffer size granted by the kernel.
|
||||
pub fn recv_buffer_size(&self) -> Result<usize, TransportError> {
|
||||
self.inner
|
||||
.recv_buffer_size()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))
|
||||
}
|
||||
|
||||
/// Get the actual send buffer size granted by the kernel.
|
||||
pub fn send_buffer_size(&self) -> Result<usize, TransportError> {
|
||||
self.inner
|
||||
.send_buffer_size()
|
||||
.map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))
|
||||
}
|
||||
|
||||
/// Synchronous send to a destination address.
|
||||
///
|
||||
/// Returns the number of bytes sent, or an `io::Error`.
|
||||
pub fn send_to(&self, data: &[u8], dest: &SocketAddr) -> std::io::Result<usize> {
|
||||
let dest: socket2::SockAddr = (*dest).into();
|
||||
self.inner.send_to(data, &dest)
|
||||
}
|
||||
|
||||
/// Synchronous receive with `SO_RXQ_OVFL` ancillary data parsing.
|
||||
///
|
||||
/// Returns `(bytes_read, source_addr, kernel_drops)`. The `kernel_drops`
|
||||
/// value is a cumulative counter since socket creation; it is 0 if
|
||||
/// `SO_RXQ_OVFL` is not supported.
|
||||
pub fn recv_from(&self, buf: &mut [u8]) -> std::io::Result<(usize, SocketAddr, u32)> {
|
||||
let fd = self.inner.as_raw_fd();
|
||||
|
||||
let mut iov = libc::iovec {
|
||||
iov_base: buf.as_mut_ptr() as *mut libc::c_void,
|
||||
iov_len: buf.len(),
|
||||
};
|
||||
|
||||
// Control message buffer sized for SO_RXQ_OVFL (u32).
|
||||
// CMSG_SPACE computes the aligned size including header.
|
||||
const CMSG_BUF_SIZE: usize = unsafe { libc::CMSG_SPACE(4) } as usize;
|
||||
let mut cmsg_buf = [0u8; CMSG_BUF_SIZE];
|
||||
|
||||
let mut src_addr: libc::sockaddr_storage = unsafe { std::mem::zeroed() };
|
||||
let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
|
||||
msg.msg_name = &mut src_addr as *mut _ as *mut libc::c_void;
|
||||
msg.msg_namelen = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
|
||||
msg.msg_iov = &mut iov;
|
||||
msg.msg_iovlen = 1;
|
||||
msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void;
|
||||
msg.msg_controllen = cmsg_buf.len() as _;
|
||||
|
||||
let n = unsafe { libc::recvmsg(fd, &mut msg, 0) };
|
||||
if n < 0 {
|
||||
return Err(std::io::Error::last_os_error());
|
||||
}
|
||||
|
||||
// Parse source address from sockaddr_storage
|
||||
let addr = sockaddr_to_socket_addr(&src_addr)?;
|
||||
|
||||
// Walk cmsg chain for SO_RXQ_OVFL drop counter
|
||||
let mut drops: u32 = 0;
|
||||
#[cfg(target_os = "linux")]
|
||||
unsafe {
|
||||
let mut cmsg = libc::CMSG_FIRSTHDR(&msg);
|
||||
while !cmsg.is_null() {
|
||||
if (*cmsg).cmsg_level == libc::SOL_SOCKET
|
||||
&& (*cmsg).cmsg_type == libc::SO_RXQ_OVFL
|
||||
{
|
||||
let data = libc::CMSG_DATA(cmsg);
|
||||
drops = std::ptr::read_unaligned(data as *const u32);
|
||||
}
|
||||
cmsg = libc::CMSG_NXTHDR(&msg, cmsg);
|
||||
}
|
||||
}
|
||||
|
||||
Ok((n as usize, addr, drops))
|
||||
}
|
||||
|
||||
/// Wrap this socket in a tokio `AsyncFd` for async I/O.
|
||||
pub fn into_async(self) -> Result<AsyncUdpSocket, TransportError> {
|
||||
let async_fd = AsyncFd::new(self)
|
||||
.map_err(|e| TransportError::StartFailed(format!("AsyncFd::new failed: {}", e)))?;
|
||||
Ok(AsyncUdpSocket {
|
||||
inner: Arc::new(async_fd),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl AsRawFd for UdpRawSocket {
|
||||
fn as_raw_fd(&self) -> RawFd {
|
||||
self.inner.as_raw_fd()
|
||||
}
|
||||
}
|
||||
|
||||
/// Async wrapper around `UdpRawSocket` using tokio's `AsyncFd`.
|
||||
///
|
||||
/// `Arc`-shareable between send and receive tasks. `AsyncFd<T>` is
|
||||
/// `Sync` when `T: Send`, which `socket2::Socket` satisfies.
|
||||
#[derive(Clone)]
|
||||
pub struct AsyncUdpSocket {
|
||||
inner: Arc<AsyncFd<UdpRawSocket>>,
|
||||
}
|
||||
|
||||
impl AsyncUdpSocket {
|
||||
/// Send a payload to a destination address.
|
||||
pub async fn send_to(
|
||||
&self,
|
||||
data: &[u8],
|
||||
dest: &SocketAddr,
|
||||
) -> Result<usize, TransportError> {
|
||||
loop {
|
||||
let mut guard = self
|
||||
.inner
|
||||
.writable()
|
||||
.await
|
||||
.map_err(|e| TransportError::SendFailed(format!("writable wait: {}", e)))?;
|
||||
|
||||
match guard.try_io(|inner| inner.get_ref().send_to(data, dest)) {
|
||||
Ok(Ok(n)) => return Ok(n),
|
||||
Ok(Err(e)) => return Err(TransportError::SendFailed(format!("{}", e))),
|
||||
Err(_would_block) => continue,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Receive a payload, source address, and kernel drop counter.
|
||||
///
|
||||
/// Returns `(bytes_read, source_addr, kernel_drops)`.
|
||||
pub async fn recv_from(
|
||||
&self,
|
||||
buf: &mut [u8],
|
||||
) -> Result<(usize, SocketAddr, u32), TransportError> {
|
||||
loop {
|
||||
let mut guard = self
|
||||
.inner
|
||||
.readable()
|
||||
.await
|
||||
.map_err(|e| TransportError::RecvFailed(format!("readable wait: {}", e)))?;
|
||||
|
||||
match guard.try_io(|inner| inner.get_ref().recv_from(buf)) {
|
||||
Ok(Ok(result)) => return Ok(result),
|
||||
Ok(Err(e)) => return Err(TransportError::RecvFailed(format!("{}", e))),
|
||||
Err(_would_block) => continue,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert a `libc::sockaddr_storage` to `std::net::SocketAddr`.
|
||||
fn sockaddr_to_socket_addr(
|
||||
storage: &libc::sockaddr_storage,
|
||||
) -> std::io::Result<SocketAddr> {
|
||||
match storage.ss_family as libc::c_int {
|
||||
libc::AF_INET => {
|
||||
let addr: &libc::sockaddr_in =
|
||||
unsafe { &*(storage as *const _ as *const libc::sockaddr_in) };
|
||||
let ip = std::net::Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr));
|
||||
let port = u16::from_be(addr.sin_port);
|
||||
Ok(SocketAddr::from((ip, port)))
|
||||
}
|
||||
libc::AF_INET6 => {
|
||||
let addr: &libc::sockaddr_in6 =
|
||||
unsafe { &*(storage as *const _ as *const libc::sockaddr_in6) };
|
||||
let ip = std::net::Ipv6Addr::from(addr.sin6_addr.s6_addr);
|
||||
let port = u16::from_be(addr.sin6_port);
|
||||
Ok(SocketAddr::from((ip, port)))
|
||||
}
|
||||
family => Err(std::io::Error::new(
|
||||
std::io::ErrorKind::InvalidData,
|
||||
format!("unsupported address family: {}", family),
|
||||
)),
|
||||
}
|
||||
}
|
||||
@@ -105,6 +105,51 @@ Explicit topologies exercising non-UDP transports.
|
||||
static peer config. Netem mutation (30% fraction, every 20-40s) and link
|
||||
flaps (1 link max, 10-20s down).
|
||||
|
||||
### Congestion and ECN
|
||||
|
||||
Scenarios testing ECN congestion signaling and transport-level congestion
|
||||
detection.
|
||||
|
||||
| Scenario | Nodes | Topology | Duration | What it tests |
|
||||
| ------------------ | ----- | -------- | -------- | ---------------------------------------------------------- |
|
||||
| congestion-stress | 10 | Tree | 120s | CE marking under kernel drops and MMP loss detection |
|
||||
| ecn-ab-on / ecn-ab-off | 6 | Tree | 120s | A/B throughput comparison: ECN enabled vs disabled |
|
||||
|
||||
- **congestion-stress**: 10-node tree with 1 Mbps egress bandwidth caps,
|
||||
5-10% netem loss, and heavy iperf3 traffic. Ingress policing (1000 kbps)
|
||||
and small `recv_buf_size` (4 KB) trigger both MMP loss detection and
|
||||
`SO_RXQ_OVFL` kernel socket drops. Validates end-to-end CE propagation:
|
||||
transit nodes detect congestion, set CE flag, destinations receive
|
||||
CE-marked packets, `ecn_ce_count` reported in MMP.
|
||||
- **ecn-ab-on / ecn-ab-off**: Paired scenarios with identical conditions
|
||||
(6-node tree, 10 Mbps egress, 1000 kbps ingress policing, 10ms link
|
||||
delay, 8 KB recv buffer) differing only in `ecn.enabled`.
|
||||
`ecn-ab-test.sh` runs both and compares throughput and congestion
|
||||
counters. Initial results: +10.2% recv throughput with ECN enabled.
|
||||
|
||||
### Ingress Traffic Control
|
||||
|
||||
Scenarios can include `ingress` configuration to simulate upstream bandwidth
|
||||
bottlenecks using tc ingress policing:
|
||||
|
||||
```yaml
|
||||
ingress:
|
||||
enabled: true
|
||||
tiers_kbps: [1000] # per-peer rate limit in kbps
|
||||
burst_bytes: 10000 # policer burst allowance
|
||||
```
|
||||
|
||||
Per-peer u32 filters on the ingress qdisc (`parent ffff:`) rate-limit
|
||||
inbound packets. Combined with small `recv_buf_size`, this reliably triggers
|
||||
`SO_RXQ_OVFL` kernel socket drops for congestion detection testing.
|
||||
|
||||
### iperf3 JSON Capture
|
||||
|
||||
Traffic sessions capture iperf3 results using `--json` output. Results are
|
||||
collected per-session from containers and saved as `iperf3-results.json` in
|
||||
the scenario output directory, enabling automated throughput analysis across
|
||||
scenario runs.
|
||||
|
||||
## CLI Options
|
||||
|
||||
| Option | Description |
|
||||
|
||||
Executable
+127
@@ -0,0 +1,127 @@
|
||||
#!/usr/bin/env bash
|
||||
# ECN A/B Throughput Test
|
||||
#
|
||||
# Runs two identical chaos scenarios — one with ECN enabled, one disabled —
|
||||
# and compares iperf3 throughput and congestion counter results.
|
||||
#
|
||||
# Usage: ./ecn-ab-test.sh [--seed N] [--duration N]
|
||||
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")"
|
||||
|
||||
EXTRA_ARGS=()
|
||||
while [[ $# -gt 0 ]]; do
|
||||
case "$1" in
|
||||
--seed|--duration)
|
||||
EXTRA_ARGS+=("$1" "$2"); shift 2 ;;
|
||||
*)
|
||||
echo "Unknown arg: $1"; exit 1 ;;
|
||||
esac
|
||||
done
|
||||
|
||||
echo "=== ECN A/B Throughput Test ==="
|
||||
echo ""
|
||||
|
||||
# --- Run A: ECN ON ---
|
||||
echo "--- Phase A: ECN ENABLED ---"
|
||||
sudo python3 -m sim scenarios/ecn-ab-on.yaml "${EXTRA_ARGS[@]}" || true
|
||||
echo ""
|
||||
|
||||
# --- Run B: ECN OFF ---
|
||||
echo "--- Phase B: ECN DISABLED ---"
|
||||
sudo python3 -m sim scenarios/ecn-ab-off.yaml "${EXTRA_ARGS[@]}" || true
|
||||
echo ""
|
||||
|
||||
# --- Compare results ---
|
||||
echo "=== Results ==="
|
||||
echo ""
|
||||
|
||||
python3 - <<'PYEOF'
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
def load_results(path):
|
||||
if not os.path.exists(path):
|
||||
return []
|
||||
with open(path) as f:
|
||||
return json.load(f)
|
||||
|
||||
def extract_throughput(results):
|
||||
"""Extract per-session throughput in Mbps from iperf3 JSON."""
|
||||
sessions = []
|
||||
for r in results:
|
||||
meta = r.get("_meta", {})
|
||||
end = r.get("end", {})
|
||||
# sum_sent / sum_received contain aggregate stats
|
||||
sent = end.get("sum_sent", {})
|
||||
recv = end.get("sum_received", {})
|
||||
sent_mbps = sent.get("bits_per_second", 0) / 1e6
|
||||
recv_mbps = recv.get("bits_per_second", 0) / 1e6
|
||||
sessions.append({
|
||||
"client": meta.get("client", "?"),
|
||||
"server": meta.get("server", "?"),
|
||||
"sent_mbps": sent_mbps,
|
||||
"recv_mbps": recv_mbps,
|
||||
})
|
||||
return sessions
|
||||
|
||||
def load_congestion(path):
|
||||
if not os.path.exists(path):
|
||||
return {}
|
||||
with open(path) as f:
|
||||
return json.load(f)
|
||||
|
||||
def print_sessions(label, sessions):
|
||||
# Filter out incomplete sessions (killed at teardown, no valid data)
|
||||
valid = [s for s in sessions if s["recv_mbps"] > 0]
|
||||
incomplete = len(sessions) - len(valid)
|
||||
if not valid:
|
||||
print(f" {label}: no completed iperf3 sessions ({incomplete} incomplete)")
|
||||
return 0
|
||||
print(f" {label}:")
|
||||
total_sent = 0
|
||||
total_recv = 0
|
||||
for s in valid:
|
||||
print(f" {s['client']:>4} -> {s['server']:<4} "
|
||||
f"sent={s['sent_mbps']:7.2f} Mbps recv={s['recv_mbps']:7.2f} Mbps")
|
||||
total_sent += s["sent_mbps"]
|
||||
total_recv += s["recv_mbps"]
|
||||
n = len(valid)
|
||||
print(f" {'':>14} avg sent={total_sent/n:7.2f} Mbps avg recv={total_recv/n:7.2f} Mbps")
|
||||
print(f" {'':>14} completed={n} incomplete={incomplete}")
|
||||
return total_recv / n
|
||||
|
||||
on_results = load_results("sim-results/ecn-ab-on/iperf3-results.json")
|
||||
off_results = load_results("sim-results/ecn-ab-off/iperf3-results.json")
|
||||
|
||||
on_sessions = extract_throughput(on_results)
|
||||
off_sessions = extract_throughput(off_results)
|
||||
|
||||
print("Throughput:")
|
||||
avg_on = print_sessions("ECN ON", on_sessions) or 0
|
||||
print()
|
||||
avg_off = print_sessions("ECN OFF", off_sessions) or 0
|
||||
print()
|
||||
|
||||
if avg_on and avg_off:
|
||||
delta_pct = ((avg_on - avg_off) / avg_off) * 100
|
||||
print(f" Delta: ECN ON vs OFF = {delta_pct:+.1f}% avg recv throughput")
|
||||
print()
|
||||
|
||||
# Congestion counters
|
||||
print("Congestion Counters (final snapshot):")
|
||||
for label, path in [("ECN ON", "sim-results/ecn-ab-on/congestion-snapshot-final.json"),
|
||||
("ECN OFF", "sim-results/ecn-ab-off/congestion-snapshot-final.json")]:
|
||||
snap = load_congestion(path)
|
||||
if not snap:
|
||||
print(f" {label}: no snapshot")
|
||||
continue
|
||||
totals = {"ce_forwarded": 0, "ce_received": 0, "congestion_detected": 0, "kernel_drop_events": 0}
|
||||
for node_id, data in sorted(snap.items()):
|
||||
cong = data.get("congestion", {})
|
||||
for k in totals:
|
||||
totals[k] += cong.get(k, 0)
|
||||
print(f" {label}: ce_fwd={totals['ce_forwarded']} ce_recv={totals['ce_received']} "
|
||||
f"cong_detect={totals['congestion_detected']} kern_drops={totals['kernel_drop_events']}")
|
||||
PYEOF
|
||||
@@ -8,6 +8,12 @@
|
||||
|
||||
set -e
|
||||
|
||||
# Enable TCP ECN negotiation for both IPv4 and IPv6 connections.
|
||||
# Despite the "ipv4" name, this sysctl controls ECN for all TCP.
|
||||
# Without this, IPv6 packets traversing the FIPS mesh carry ECN=0b00
|
||||
# (Not-ECT), and mark_ipv6_ecn_ce() is a no-op per RFC 3168.
|
||||
sysctl -w net.ipv4.tcp_ecn=1 >/dev/null 2>&1 || true
|
||||
|
||||
# Start background services
|
||||
dnsmasq
|
||||
/usr/sbin/sshd
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
# Congestion Stress: exercise both MMP-based and kernel-drop congestion detection
|
||||
#
|
||||
# Topology: 10-node tree with all links bandwidth-limited to 1 Mbps
|
||||
# (egress HTB + ingress policing). Heavy iperf3 traffic (8 concurrent
|
||||
# sessions, 8 parallel streams) combined with 5-10% netem loss.
|
||||
#
|
||||
# Congestion detection signals exercised:
|
||||
# 1. MMP loss detection: netem loss exceeds the 5% loss_threshold,
|
||||
# triggering detect_congestion() via MMP metrics on transit nodes.
|
||||
# 2. Kernel socket drops: small recv_buf_size (8KB) combined with
|
||||
# traffic saturation causes SO_RXQ_OVFL on the UDP socket,
|
||||
# triggering the transport drop detection path.
|
||||
# 3. Ingress policing: tc policer on the receive side drops excess
|
||||
# inbound packets, creating bursty arrival patterns.
|
||||
#
|
||||
# ECN is explicitly enabled via fips_overrides.
|
||||
#
|
||||
# Success criteria (verified via post-run congestion snapshot):
|
||||
# - congestion_detected > 0 on at least one forwarding node
|
||||
# - kernel_drop_events > 0 on at least one node
|
||||
# - ce_forwarded > 0 on transit nodes
|
||||
# - ce_received > 0 on destination nodes
|
||||
#
|
||||
# Topology:
|
||||
#
|
||||
# n01 (root)
|
||||
# / \
|
||||
# n02 n03
|
||||
# / \ / \
|
||||
# n04 n05 n06 n07
|
||||
# | |
|
||||
# n08 n09
|
||||
# |
|
||||
# n10
|
||||
|
||||
scenario:
|
||||
name: "congestion-stress"
|
||||
seed: 7
|
||||
duration_secs: 120
|
||||
|
||||
topology:
|
||||
algorithm: explicit
|
||||
num_nodes: 10
|
||||
params:
|
||||
adjacency:
|
||||
- [n01, n02]
|
||||
- [n01, n03]
|
||||
- [n02, n04]
|
||||
- [n02, n05]
|
||||
- [n03, n06]
|
||||
- [n03, n07]
|
||||
- [n04, n08]
|
||||
- [n05, n09]
|
||||
- [n08, n10]
|
||||
subnet: "172.20.0.0/24"
|
||||
ip_start: 10
|
||||
|
||||
bandwidth:
|
||||
enabled: true
|
||||
tiers_mbps: [1]
|
||||
|
||||
ingress:
|
||||
enabled: true
|
||||
tiers_kbps: [1000]
|
||||
burst_bytes: 16000
|
||||
|
||||
netem:
|
||||
enabled: true
|
||||
default_policy:
|
||||
delay_ms: [2, 5]
|
||||
jitter_ms: [0, 1]
|
||||
loss_pct: [5, 10]
|
||||
mutation:
|
||||
interval_secs: {min: 9999, max: 9999}
|
||||
fraction: 0.0
|
||||
policies:
|
||||
baseline:
|
||||
delay_ms: [2, 5]
|
||||
loss_pct: [5, 10]
|
||||
|
||||
link_flaps:
|
||||
enabled: false
|
||||
|
||||
traffic:
|
||||
enabled: true
|
||||
max_concurrent: 8
|
||||
interval_secs: {min: 2, max: 5}
|
||||
duration_secs: {min: 30, max: 60}
|
||||
parallel_streams: 8
|
||||
|
||||
node_churn:
|
||||
enabled: false
|
||||
|
||||
logging:
|
||||
rust_log: "info"
|
||||
output_dir: "./sim-results"
|
||||
|
||||
fips_overrides:
|
||||
node:
|
||||
ecn:
|
||||
enabled: true
|
||||
transports:
|
||||
udp:
|
||||
recv_buf_size: 4096
|
||||
@@ -0,0 +1,80 @@
|
||||
# ECN A/B Test: ECN DISABLED
|
||||
#
|
||||
# Identical to ecn-ab-on.yaml except ecn.enabled = false.
|
||||
# Run both scenarios and compare iperf3 throughput results.
|
||||
#
|
||||
# Same conditions: 10 Mbps egress, 1000 kbps ingress policing, 10ms
|
||||
# link delay, 8KB recv buffer. Without ECN, the only congestion signal
|
||||
# TCP gets is packet loss (timeouts/triple dup-ack).
|
||||
#
|
||||
# n01 (root)
|
||||
# / \
|
||||
# n02 n03
|
||||
# / \ \
|
||||
# n04 n05 n06
|
||||
|
||||
scenario:
|
||||
name: "ecn-ab-off"
|
||||
seed: 42
|
||||
duration_secs: 120
|
||||
|
||||
topology:
|
||||
algorithm: explicit
|
||||
num_nodes: 6
|
||||
params:
|
||||
adjacency:
|
||||
- [n01, n02]
|
||||
- [n01, n03]
|
||||
- [n02, n04]
|
||||
- [n02, n05]
|
||||
- [n03, n06]
|
||||
subnet: "172.20.0.0/24"
|
||||
ip_start: 10
|
||||
|
||||
bandwidth:
|
||||
enabled: true
|
||||
tiers_mbps: [10]
|
||||
|
||||
ingress:
|
||||
enabled: true
|
||||
tiers_kbps: [1000]
|
||||
burst_bytes: 16000
|
||||
|
||||
netem:
|
||||
enabled: true
|
||||
default_policy:
|
||||
delay_ms: [10, 10]
|
||||
jitter_ms: [0, 0]
|
||||
loss_pct: [0, 0]
|
||||
mutation:
|
||||
interval_secs: {min: 9999, max: 9999}
|
||||
fraction: 0.0
|
||||
policies:
|
||||
baseline:
|
||||
delay_ms: [10, 10]
|
||||
loss_pct: [0, 0]
|
||||
|
||||
link_flaps:
|
||||
enabled: false
|
||||
|
||||
traffic:
|
||||
enabled: true
|
||||
max_concurrent: 4
|
||||
interval_secs: {min: 3, max: 5}
|
||||
duration_secs: {min: 20, max: 30}
|
||||
parallel_streams: 4
|
||||
|
||||
node_churn:
|
||||
enabled: false
|
||||
|
||||
logging:
|
||||
rust_log: "info"
|
||||
output_dir: "./sim-results/ecn-ab-off"
|
||||
|
||||
fips_overrides:
|
||||
node:
|
||||
ecn:
|
||||
enabled: false
|
||||
transports:
|
||||
udp:
|
||||
recv_buf_size: 8192
|
||||
@@ -0,0 +1,100 @@
|
||||
# ECN A/B Test: ECN ENABLED
|
||||
#
|
||||
# Identical to ecn-ab-off.yaml except ecn.enabled = true.
|
||||
# Run both scenarios and compare iperf3 throughput results.
|
||||
#
|
||||
# Design rationale:
|
||||
# - Ingress policing at 1000 kbps: unlike HTB (which paces smoothly),
|
||||
# the ingress policer *drops* excess inbound packets at the kernel
|
||||
# level. This creates real packet loss visible to the FIPS UDP
|
||||
# socket, triggering detect_congestion() via transport_drops.
|
||||
# - 10 Mbps egress bandwidth: lets traffic flow freely outbound so
|
||||
# that the ingress policer is the bottleneck. Multiple flows
|
||||
# competing through bottleneck nodes generate excess traffic that
|
||||
# the policer drops.
|
||||
# - 10ms link delay: creates realistic RTT and causes TCP to batch
|
||||
# packets into bursts, making ingress drops more likely.
|
||||
# - 8KB recv_buf_size: small buffer amplifies the effect of bursty
|
||||
# arrivals — dropped packets at ingress + buffer overflow trigger
|
||||
# detect_congestion() via both MMP loss and transport_drops paths.
|
||||
# - tcp_ecn=1 in entrypoint.sh: TCP negotiates ECN so CE marks
|
||||
# reach TCP congestion control, enabling proactive backoff.
|
||||
# - 4 concurrent sessions with 4 parallel streams: moderate load
|
||||
# that creates real contention without overwhelming everything.
|
||||
#
|
||||
# With ECN enabled, transit nodes detect congestion early and set CE
|
||||
# flags, causing TCP to reduce its window proactively. Without ECN,
|
||||
# TCP only learns about congestion from packet drops (timeouts/triple
|
||||
# dup-ack), which is slower and causes more throughput loss.
|
||||
#
|
||||
# n01 (root)
|
||||
# / \
|
||||
# n02 n03
|
||||
# / \ \
|
||||
# n04 n05 n06
|
||||
|
||||
scenario:
|
||||
name: "ecn-ab-on"
|
||||
seed: 42
|
||||
duration_secs: 120
|
||||
|
||||
topology:
|
||||
algorithm: explicit
|
||||
num_nodes: 6
|
||||
params:
|
||||
adjacency:
|
||||
- [n01, n02]
|
||||
- [n01, n03]
|
||||
- [n02, n04]
|
||||
- [n02, n05]
|
||||
- [n03, n06]
|
||||
subnet: "172.20.0.0/24"
|
||||
ip_start: 10
|
||||
|
||||
bandwidth:
|
||||
enabled: true
|
||||
tiers_mbps: [10]
|
||||
|
||||
ingress:
|
||||
enabled: true
|
||||
tiers_kbps: [1000]
|
||||
burst_bytes: 16000
|
||||
|
||||
netem:
|
||||
enabled: true
|
||||
default_policy:
|
||||
delay_ms: [10, 10]
|
||||
jitter_ms: [0, 0]
|
||||
loss_pct: [0, 0]
|
||||
mutation:
|
||||
interval_secs: {min: 9999, max: 9999}
|
||||
fraction: 0.0
|
||||
policies:
|
||||
baseline:
|
||||
delay_ms: [10, 10]
|
||||
loss_pct: [0, 0]
|
||||
|
||||
link_flaps:
|
||||
enabled: false
|
||||
|
||||
traffic:
|
||||
enabled: true
|
||||
max_concurrent: 4
|
||||
interval_secs: {min: 3, max: 5}
|
||||
duration_secs: {min: 20, max: 30}
|
||||
parallel_streams: 4
|
||||
|
||||
node_churn:
|
||||
enabled: false
|
||||
|
||||
logging:
|
||||
rust_log: "info"
|
||||
output_dir: "./sim-results/ecn-ab-on"
|
||||
|
||||
fips_overrides:
|
||||
node:
|
||||
ecn:
|
||||
enabled: true
|
||||
transports:
|
||||
udp:
|
||||
recv_buf_size: 8192
|
||||
@@ -101,3 +101,42 @@ def snapshot_all_mmp(topology: SimTopology) -> dict[str, dict]:
|
||||
else:
|
||||
log.warning("No MMP data from %s", node_id)
|
||||
return result
|
||||
|
||||
|
||||
def query_routing(container: str) -> dict | None:
|
||||
"""Query a node's routing stats (includes congestion counters)."""
|
||||
return query_node(container, "show_routing")
|
||||
|
||||
|
||||
def query_transports(container: str) -> dict | None:
|
||||
"""Query a node's transport state (includes kernel drop counters)."""
|
||||
return query_node(container, "show_transports")
|
||||
|
||||
|
||||
def snapshot_all_congestion(topology: SimTopology) -> dict[str, dict]:
|
||||
"""Query show_routing on all nodes to capture congestion counters.
|
||||
|
||||
Returns {node_id: {"congestion": {...}, "kernel_drops": [...]}}.
|
||||
Nodes that fail to respond are omitted from the result.
|
||||
"""
|
||||
result = {}
|
||||
for node_id in sorted(topology.nodes):
|
||||
container = topology.container_name(node_id)
|
||||
routing = query_routing(container)
|
||||
transports = query_transports(container)
|
||||
if routing is not None:
|
||||
entry = {"congestion": routing.get("congestion", {})}
|
||||
if transports is not None:
|
||||
drops = []
|
||||
for t in transports.get("transports", []):
|
||||
stats = t.get("stats", {})
|
||||
drops.append({
|
||||
"transport_id": t.get("transport_id"),
|
||||
"name": t.get("name"),
|
||||
"kernel_drops": stats.get("kernel_drops"),
|
||||
})
|
||||
entry["kernel_drops"] = drops
|
||||
result[node_id] = entry
|
||||
else:
|
||||
log.warning("No routing data from %s", node_id)
|
||||
return result
|
||||
|
||||
@@ -26,6 +26,8 @@ class AnalysisResult:
|
||||
mmp_session_metrics: list[tuple[str, str]] = field(default_factory=list)
|
||||
handshake_timeouts: list[tuple[str, str]] = field(default_factory=list)
|
||||
panics: list[tuple[str, str]] = field(default_factory=list)
|
||||
congestion_detected: list[tuple[str, str]] = field(default_factory=list)
|
||||
kernel_drop_events: list[tuple[str, str]] = field(default_factory=list)
|
||||
|
||||
def summary(self) -> str:
|
||||
lines = [
|
||||
@@ -41,6 +43,8 @@ class AnalysisResult:
|
||||
f"Handshake timeouts: {len(self.handshake_timeouts)}",
|
||||
f"MMP link samples: {len(self.mmp_link_metrics)}",
|
||||
f"MMP session samples: {len(self.mmp_session_metrics)}",
|
||||
f"Congestion events: {len(self.congestion_detected)}",
|
||||
f"Kernel drop events: {len(self.kernel_drop_events)}",
|
||||
]
|
||||
|
||||
if self.panics:
|
||||
@@ -134,6 +138,11 @@ def analyze_logs(logs: dict[str, str]) -> AnalysisResult:
|
||||
result.mmp_link_metrics.append((container, line))
|
||||
if "MMP session metrics" in line:
|
||||
result.mmp_session_metrics.append((container, line))
|
||||
# Congestion events
|
||||
if "Congestion detected" in line:
|
||||
result.congestion_detected.append((container, line))
|
||||
if "Kernel recv drops first observed" in line:
|
||||
result.kernel_drop_events.append((container, line))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
@@ -8,6 +8,13 @@ u32 filters match destination IP to direct traffic to the right class.
|
||||
├── class 1:1 → netem 11: (peer 1) ← u32 filter: dst=<peer1_ip>
|
||||
├── class 1:2 → netem 12: (peer 2) ← u32 filter: dst=<peer2_ip>
|
||||
└── class 1:99 → pfifo (default, no impairment)
|
||||
|
||||
Optional ingress policing adds rate-based packet dropping on the
|
||||
receive side via tc ingress qdisc + policer filters:
|
||||
|
||||
eth0 ingress (ffff:)
|
||||
├── u32 filter: src=<peer1_ip> → police rate <R>kbit burst <B> drop
|
||||
└── u32 filter: src=<peer2_ip> → police rate <R>kbit burst <B> drop
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -17,7 +24,7 @@ import random
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .docker_exec import docker_exec_quiet, is_container_running
|
||||
from .scenario import BandwidthConfig, LinkPolicyOverride, NetemConfig, NetemPolicy
|
||||
from .scenario import BandwidthConfig, IngressConfig, LinkPolicyOverride, NetemConfig, NetemPolicy
|
||||
from .topology import SimTopology, veth_interface_name
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -65,6 +72,9 @@ class LinkNetemState:
|
||||
netem_handle: str # e.g., "11:"
|
||||
params: NetemParams = field(default_factory=NetemParams)
|
||||
rate_mbit: int = 0 # 0 = unlimited (1gbit default)
|
||||
ingress_rate_kbps: int = 0 # 0 = no ingress policing
|
||||
ingress_burst_bytes: int = 32000
|
||||
ingress_filter_prio: int = 0 # u32 filter priority for this peer
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -85,6 +95,7 @@ class NetemManager:
|
||||
config: NetemConfig,
|
||||
rng: random.Random,
|
||||
bandwidth: BandwidthConfig | None = None,
|
||||
ingress: IngressConfig | None = None,
|
||||
):
|
||||
self.topology = topology
|
||||
self.config = config
|
||||
@@ -102,6 +113,14 @@ class NetemManager:
|
||||
rate = rng.choice(bandwidth.tiers_mbps)
|
||||
self._edge_rates[(a, b)] = rate
|
||||
self._edge_rates[(b, a)] = rate
|
||||
# Per-edge ingress policing: (node_a, node_b) -> rate in kbps
|
||||
self._ingress_config = ingress
|
||||
self._ingress_rates: dict[tuple[str, str], int] = {}
|
||||
if ingress and ingress.enabled:
|
||||
for a, b in topology.edges:
|
||||
rate = rng.choice(ingress.tiers_kbps)
|
||||
self._ingress_rates[(a, b)] = rate
|
||||
self._ingress_rates[(b, a)] = rate
|
||||
# Per-edge policy overrides: canonical "nXX-nYY" -> NetemPolicy
|
||||
# Build a set of canonical edge strings for validation
|
||||
topo_edge_strs = {"-".join(sorted([a, b])) for a, b in topology.edges}
|
||||
@@ -194,6 +213,9 @@ class NetemManager:
|
||||
f"prio {idx} u32 match ip dst {dest_ip}/32 flowid {class_id}"
|
||||
)
|
||||
|
||||
ingress_rate = self._ingress_rates.get((node_id, peer_id), 0)
|
||||
ingress_burst = self._ingress_config.burst_bytes if self._ingress_config else 32000
|
||||
|
||||
state = LinkNetemState(
|
||||
container=container,
|
||||
dest_ip=dest_ip,
|
||||
@@ -201,16 +223,33 @@ class NetemManager:
|
||||
netem_handle=netem_handle,
|
||||
params=params,
|
||||
rate_mbit=rate_mbit,
|
||||
ingress_rate_kbps=ingress_rate,
|
||||
ingress_burst_bytes=ingress_burst,
|
||||
ingress_filter_prio=idx,
|
||||
)
|
||||
container_states[dest_ip] = state
|
||||
|
||||
# Ingress policing: add ingress qdisc + per-peer policer filters
|
||||
if self._ingress_rates:
|
||||
cmds.append(f"tc qdisc add dev {IFACE} ingress 2>/dev/null || true")
|
||||
for idx, (peer_id, dest_ip) in enumerate(ip_peers.items(), start=1):
|
||||
ingress_rate = self._ingress_rates.get((node_id, peer_id), 0)
|
||||
if ingress_rate > 0:
|
||||
ingress_burst = self._ingress_config.burst_bytes if self._ingress_config else 32000
|
||||
cmds.append(
|
||||
f"tc filter add dev {IFACE} parent ffff: protocol ip "
|
||||
f"prio {idx} u32 match ip src {dest_ip}/32 "
|
||||
f"police rate {ingress_rate}kbit burst {ingress_burst} drop"
|
||||
)
|
||||
|
||||
full_cmd = " && ".join(cmds)
|
||||
result = docker_exec_quiet(container, full_cmd, timeout=30)
|
||||
if result is not None:
|
||||
log.info(
|
||||
"Configured per-link netem on %s (%d IP peers)",
|
||||
"Configured per-link netem on %s (%d IP peers%s)",
|
||||
container,
|
||||
len(ip_peers),
|
||||
", ingress policing" if self._ingress_rates else "",
|
||||
)
|
||||
else:
|
||||
log.warning("Failed to configure netem on %s", container)
|
||||
@@ -282,13 +321,27 @@ class NetemManager:
|
||||
f"prio {prio} u32 match ip dst {dest_ip}/32 flowid {state.class_id}"
|
||||
)
|
||||
|
||||
# Re-apply ingress policing
|
||||
has_ingress = any(s.ingress_rate_kbps > 0 for s in container_states.values())
|
||||
if has_ingress:
|
||||
cmds.append(f"tc qdisc add dev {IFACE} ingress 2>/dev/null || true")
|
||||
for dest_ip, state in container_states.items():
|
||||
if state.ingress_rate_kbps > 0:
|
||||
cmds.append(
|
||||
f"tc filter add dev {IFACE} parent ffff: protocol ip "
|
||||
f"prio {state.ingress_filter_prio} u32 match ip src {dest_ip}/32 "
|
||||
f"police rate {state.ingress_rate_kbps}kbit "
|
||||
f"burst {state.ingress_burst_bytes} drop"
|
||||
)
|
||||
|
||||
full_cmd = " && ".join(cmds)
|
||||
result = docker_exec_quiet(container, full_cmd, timeout=30)
|
||||
if result is not None:
|
||||
log.info(
|
||||
"Re-applied IP netem on %s (%d peers)",
|
||||
"Re-applied IP netem on %s (%d peers%s)",
|
||||
container,
|
||||
len(container_states),
|
||||
", ingress policing" if has_ingress else "",
|
||||
)
|
||||
else:
|
||||
log.warning("Failed to re-apply IP netem on %s", container)
|
||||
|
||||
@@ -12,7 +12,7 @@ import time
|
||||
|
||||
from .compose import generate_compose
|
||||
from .config_gen import write_configs
|
||||
from .control import snapshot_all_mmp, snapshot_all_trees
|
||||
from .control import snapshot_all_congestion, snapshot_all_mmp, snapshot_all_trees
|
||||
from .docker_exec import docker_compose
|
||||
from .links import LinkManager
|
||||
from .logs import AnalysisResult, analyze_logs, collect_logs, write_sim_metadata
|
||||
@@ -140,7 +140,8 @@ class SimRunner:
|
||||
# 7. Initialize managers
|
||||
if s.netem.enabled:
|
||||
bw = s.bandwidth if s.bandwidth.enabled else None
|
||||
self.netem_mgr = NetemManager(self.topology, s.netem, self.rng, bandwidth=bw)
|
||||
ig = s.ingress if s.ingress.enabled else None
|
||||
self.netem_mgr = NetemManager(self.topology, s.netem, self.rng, bandwidth=bw, ingress=ig)
|
||||
self.netem_mgr.down_nodes = self._down_nodes
|
||||
log.info("Applying initial per-link netem...")
|
||||
self.netem_mgr.setup_initial()
|
||||
@@ -254,6 +255,15 @@ class SimRunner:
|
||||
log.info("Restoring stopped nodes...")
|
||||
self.node_mgr.restore_all()
|
||||
|
||||
# Collect iperf3 throughput results before containers stop
|
||||
if self.traffic_mgr:
|
||||
iperf_results = self.traffic_mgr.collect_results()
|
||||
if iperf_results:
|
||||
iperf_path = os.path.join(self.output_dir, "iperf3-results.json")
|
||||
with open(iperf_path, "w") as f:
|
||||
json.dump(iperf_results, f, indent=2)
|
||||
log.info("Saved %d iperf3 results to %s", len(iperf_results), iperf_path)
|
||||
|
||||
# Take final tree snapshot while nodes are still running
|
||||
self._take_snapshot("final")
|
||||
|
||||
@@ -298,27 +308,33 @@ class SimRunner:
|
||||
return result
|
||||
|
||||
def _take_snapshot(self, label: str):
|
||||
"""Query all nodes via control socket and save tree/MMP snapshots."""
|
||||
"""Query all nodes via control socket and save tree/MMP/congestion snapshots."""
|
||||
if not self.topology:
|
||||
return
|
||||
log.info("Taking %s snapshot...", label)
|
||||
tree_snap = snapshot_all_trees(self.topology)
|
||||
mmp_snap = snapshot_all_mmp(self.topology)
|
||||
congestion_snap = snapshot_all_congestion(self.topology)
|
||||
|
||||
tree_path = os.path.join(self.output_dir, f"tree-snapshot-{label}.json")
|
||||
mmp_path = os.path.join(self.output_dir, f"mmp-snapshot-{label}.json")
|
||||
congestion_path = os.path.join(self.output_dir, f"congestion-snapshot-{label}.json")
|
||||
os.makedirs(self.output_dir, exist_ok=True)
|
||||
with open(tree_path, "w") as f:
|
||||
json.dump(tree_snap, f, indent=2)
|
||||
with open(mmp_path, "w") as f:
|
||||
json.dump(mmp_snap, f, indent=2)
|
||||
with open(congestion_path, "w") as f:
|
||||
json.dump(congestion_snap, f, indent=2)
|
||||
log.info(
|
||||
"Snapshot %s: %d/%d tree, %d/%d mmp responses",
|
||||
"Snapshot %s: %d/%d tree, %d/%d mmp, %d/%d congestion responses",
|
||||
label,
|
||||
len(tree_snap),
|
||||
len(self.topology.nodes),
|
||||
len(mmp_snap),
|
||||
len(self.topology.nodes),
|
||||
len(congestion_snap),
|
||||
len(self.topology.nodes),
|
||||
)
|
||||
|
||||
def _schedule_next(self, now: float, interval) -> float:
|
||||
|
||||
@@ -120,6 +120,22 @@ class BandwidthConfig:
|
||||
tiers_mbps: list[int] = field(default_factory=lambda: [1, 10, 100, 1000])
|
||||
|
||||
|
||||
@dataclass
|
||||
class IngressConfig:
|
||||
"""Per-link ingress policing via tc ingress qdisc + policer.
|
||||
|
||||
When enabled, each link gets an ingress policer that drops inbound
|
||||
packets exceeding the configured rate. Unlike egress HTB (which
|
||||
paces packets smoothly), the policer drops excess packets at the
|
||||
kernel level, creating bursty arrival patterns that can overflow
|
||||
small socket buffers and trigger SO_RXQ_OVFL kernel drops.
|
||||
"""
|
||||
|
||||
enabled: bool = False
|
||||
tiers_kbps: list[int] = field(default_factory=lambda: [1000])
|
||||
burst_bytes: int = 32000
|
||||
|
||||
|
||||
@dataclass
|
||||
class LoggingConfig:
|
||||
rust_log: str = "info"
|
||||
@@ -137,6 +153,7 @@ class Scenario:
|
||||
traffic: TrafficConfig = field(default_factory=TrafficConfig)
|
||||
node_churn: NodeChurnConfig = field(default_factory=NodeChurnConfig)
|
||||
bandwidth: BandwidthConfig = field(default_factory=BandwidthConfig)
|
||||
ingress: IngressConfig = field(default_factory=IngressConfig)
|
||||
logging: LoggingConfig = field(default_factory=LoggingConfig)
|
||||
# Raw YAML dict appended to each generated FIPS node config.
|
||||
# Allows scenarios to override any FIPS config parameter
|
||||
@@ -271,6 +288,16 @@ def load_scenario(path: str) -> Scenario:
|
||||
raise ValueError("bandwidth.tiers_mbps must be a non-empty list")
|
||||
s.bandwidth.tiers_mbps = [int(t) for t in tiers]
|
||||
|
||||
# Ingress section
|
||||
ig = raw.get("ingress", {})
|
||||
s.ingress.enabled = ig.get("enabled", False)
|
||||
if "tiers_kbps" in ig:
|
||||
tiers = ig["tiers_kbps"]
|
||||
if not isinstance(tiers, list) or not tiers:
|
||||
raise ValueError("ingress.tiers_kbps must be a non-empty list")
|
||||
s.ingress.tiers_kbps = [int(t) for t in tiers]
|
||||
s.ingress.burst_bytes = int(ig.get("burst_bytes", 32000))
|
||||
|
||||
# Logging section
|
||||
lg = raw.get("logging", {})
|
||||
s.logging.rust_log = lg.get("rust_log", "info")
|
||||
@@ -376,3 +403,9 @@ def _validate(s: Scenario):
|
||||
for tier in s.bandwidth.tiers_mbps:
|
||||
if tier <= 0:
|
||||
raise ValueError(f"bandwidth.tiers_mbps: all values must be > 0, got {tier}")
|
||||
if s.ingress.enabled:
|
||||
for tier in s.ingress.tiers_kbps:
|
||||
if tier <= 0:
|
||||
raise ValueError(f"ingress.tiers_kbps: all values must be > 0, got {tier}")
|
||||
if s.ingress.burst_bytes <= 0:
|
||||
raise ValueError(f"ingress.burst_bytes must be > 0, got {s.ingress.burst_bytes}")
|
||||
|
||||
@@ -7,6 +7,7 @@ entrypoint).
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
@@ -26,6 +27,7 @@ class TrafficSession:
|
||||
started_at: float
|
||||
duration_secs: int
|
||||
container: str
|
||||
result_file: str = ""
|
||||
|
||||
|
||||
class TrafficManager:
|
||||
@@ -43,6 +45,7 @@ class TrafficManager:
|
||||
self.rng = rng
|
||||
self.down_nodes = down_nodes or set()
|
||||
self.active_sessions: list[TrafficSession] = []
|
||||
self.completed_results: list[dict] = []
|
||||
|
||||
@property
|
||||
def active_count(self) -> int:
|
||||
@@ -74,10 +77,14 @@ class TrafficManager:
|
||||
)
|
||||
streams = self.config.parallel_streams
|
||||
|
||||
# Start iperf3 in background (nohup, stdout to /dev/null)
|
||||
# Result file inside the container for JSON capture
|
||||
ts = int(time.time())
|
||||
result_file = f"/tmp/iperf3-{client}-{server}-{ts}.json"
|
||||
|
||||
# Start iperf3 in background with JSON output
|
||||
cmd = (
|
||||
f"nohup iperf3 -c {server_npub}.fips -t {duration} "
|
||||
f"-P {streams} > /dev/null 2>&1 &"
|
||||
f"-P {streams} --json > {result_file} 2>&1 &"
|
||||
)
|
||||
result = docker_exec_quiet(container, cmd)
|
||||
if result is not None:
|
||||
@@ -87,6 +94,7 @@ class TrafficManager:
|
||||
started_at=time.time(),
|
||||
duration_secs=duration,
|
||||
container=container,
|
||||
result_file=result_file,
|
||||
)
|
||||
self.active_sessions.append(session)
|
||||
log.info(
|
||||
@@ -103,16 +111,49 @@ class TrafficManager:
|
||||
"""Remove sessions that have completed (based on time)."""
|
||||
now = time.time()
|
||||
grace = 5 # seconds after expected completion
|
||||
before = len(self.active_sessions)
|
||||
self.active_sessions = [
|
||||
s
|
||||
for s in self.active_sessions
|
||||
if now - s.started_at < s.duration_secs + grace
|
||||
]
|
||||
removed = before - len(self.active_sessions)
|
||||
still_active = []
|
||||
for s in self.active_sessions:
|
||||
if now - s.started_at >= s.duration_secs + grace:
|
||||
self._collect_result(s)
|
||||
else:
|
||||
still_active.append(s)
|
||||
removed = len(self.active_sessions) - len(still_active)
|
||||
self.active_sessions = still_active
|
||||
if removed > 0:
|
||||
log.debug("Cleaned up %d expired traffic sessions", removed)
|
||||
|
||||
def _collect_result(self, session: TrafficSession):
|
||||
"""Retrieve iperf3 JSON result from container."""
|
||||
if not session.result_file:
|
||||
return
|
||||
if session.client_node in self.down_nodes:
|
||||
return
|
||||
stdout = docker_exec_quiet(
|
||||
session.container,
|
||||
f"cat {session.result_file} 2>/dev/null; rm -f {session.result_file}",
|
||||
)
|
||||
if stdout is None:
|
||||
return
|
||||
try:
|
||||
data = json.loads(stdout.strip())
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
log.debug("Could not parse iperf3 result for %s -> %s",
|
||||
session.client_node, session.server_node)
|
||||
return
|
||||
data["_meta"] = {
|
||||
"client": session.client_node,
|
||||
"server": session.server_node,
|
||||
"duration_secs": session.duration_secs,
|
||||
}
|
||||
self.completed_results.append(data)
|
||||
|
||||
def collect_results(self) -> list[dict]:
|
||||
"""Return all completed iperf3 JSON results."""
|
||||
# Collect any remaining active sessions that may have finished
|
||||
for s in self.active_sessions:
|
||||
self._collect_result(s)
|
||||
return self.completed_results
|
||||
|
||||
def stop_all(self):
|
||||
"""Kill all iperf3 client processes in running containers."""
|
||||
seen = set()
|
||||
@@ -124,4 +165,7 @@ class TrafficManager:
|
||||
"killall iperf3 2>/dev/null; true",
|
||||
)
|
||||
seen.add(session.container)
|
||||
# Collect results before clearing (iperf3 writes partial JSON on kill)
|
||||
for s in self.active_sessions:
|
||||
self._collect_result(s)
|
||||
self.active_sessions.clear()
|
||||
|
||||
Reference in New Issue
Block a user