subetha-cxc 0.1.4

MMF-backed cross-process IPC primitives for SubEtha: SharedRing, SharedHashMap, SharedRWLock, SharedSemaphore, SharedLRUCache, OwnerLease, HeartbeatTable, plus 30+ more. One byte layout serves cross-thread, cross-process, and disk-persistent.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
//! Adaptive control for the sliding-window RLC code: turn the fused channel
//! assessment ([`crate::fusion::SensorSnapshot`]) into RLC coding parameters -
//! the window size, the repair cadence (code rate), and the coefficient
//! density - then hold them steady with immediate-up / conservative-down
//! hysteresis so the knobs do not flap on a noisy channel.
//!
//! This is the RLC counterpart of [`crate::fusion::raw_target`], which maps the
//! same assessment to block-RS `(parity_r, interleave_depth)`. Both consume
//! [`crate::fusion::effective_loss`] and [`crate::fusion::is_clean`], so the two
//! codes judge the channel identically and differ only in the parameter map:
//!
//!  - **Rate from loss (`R = T / (T + B)`).** One repair every `step` source
//!    symbols is `T = step` source to `B = 1` repair, code rate
//!    `R = step / (step + 1)` and redundancy `1 / (step + 1)`. To cover an
//!    effective loss `p` with a safety margin `m`, the redundancy must satisfy
//!    `1 / (step + 1) >= m * p`, i.e. `step <= 1 / (m * p) - 1`. Higher loss
//!    shrinks `step` (more repairs, lower rate); lighter loss grows it toward
//!    the lightest-protection cap. This is the QUIC-FEC adaptive-rate result:
//!    the redundancy tracks the measured loss instead of a fixed code rate.
//!  - **Window from burst length.** A repair over a window of `w` source
//!    symbols, emitted every `step`, gives `w / step` repairs covering any one
//!    symbol. Recovering a burst of `b` consecutive losses needs at least `b`
//!    independent repairs spanning it, so `w >= b * step`. The mean burst length
//!    is `burstiness * 16` (the [`crate::burst_model_sensor`] Gilbert-Elliott
//!    fit, normalized into the snapshot), so the window scales with the real
//!    fitted burst, not a fixed depth.
//!  - **Density from burstiness / congestion.** A denser coefficient vector
//!    (higher `dt`) makes each repair touch more source symbols, raising the
//!    recovery probability per repair at more compute cost. Bursty or congested
//!    loss leans dense; light isolated loss can stay sparser.
//!  - **Disable-on-clean.** A provably-clean link ([`crate::fusion::is_clean`])
//!    turns coding off entirely: QUIC-FEC found FEC *hurts* a clean / bulk path
//!    (the redundancy is pure overhead and competes with the data for the
//!    bottleneck), so the RLC path ships data only and leans on the ARQ floor
//!    until the controller re-arms on the first sign of loss.

use crate::fusion::{is_clean, SensorSnapshot};
use crate::rlc_fec::DEFAULT_DT;

/// Safety margin on the rate law: provision redundancy for `RATE_MARGIN` times
/// the measured effective loss, so a momentary spike above the mean is still
/// covered rather than NAK'd.
/// Base redundancy margin on the rate law at near-zero round trip (a loopback /
/// IPC path where a NAK is nearly free, so a light code that leans on ARQ is
/// fine).
const BASE_RATE_MARGIN: f32 = 1.4;
/// How much the rate margin grows per millisecond of round trip. On a real
/// network a NAK costs a round trip, so heavier FEC (a smaller step) that
/// recovers in-window without a retransmit is worth the extra redundancy - the
/// streaming-codes "provision FEC to the latency budget" result, which only
/// shows up off-loopback (validated on the LAN: at near-zero RTT a light step=6
/// beats static, but at LAN RTT it loses to the heavier static step=4).
const RTT_MARGIN_SLOPE: f32 = 0.8;
/// Ceiling on the rate margin, so a very high round trip does not drive the code
/// to its heaviest rate for a modest loss.
const MAX_RATE_MARGIN: f32 = 2.8;

/// The rate-law redundancy margin for a measured round trip: base at near-zero
/// RTT, growing with RTT (an expensive NAK is worth avoiding with more FEC).
fn rate_margin_for_rtt(rtt_ms: f32) -> f32 {
    (BASE_RATE_MARGIN + RTT_MARGIN_SLOPE * rtt_ms.max(0.0)).min(MAX_RATE_MARGIN)
}

/// Lightest protection: at most one repair per [`STEP_MAX`] source symbols
/// (code rate `STEP_MAX / (STEP_MAX + 1)` ~= 0.94). Reached as loss approaches
/// zero (but not clean, where coding turns off entirely).
const STEP_MAX: u16 = 16;
/// Heaviest protection: one repair per source symbol (code rate 1/2). Reached
/// under very high effective loss.
const STEP_MIN: u16 = 1;

/// Smallest coding window: enough overlap for isolated-loss recovery.
const WINDOW_MIN: u16 = 8;
/// Largest coding window. Bounded so the decoder's Gaussian-elimination cost
/// per repair stays small and the receiver's recovery horizon comfortably
/// exceeds it.
const WINDOW_MAX: u16 = 64;
/// Window provisioning factor over the bare burst-span estimate `mean_burst *
/// step`, so a burst slightly longer than the fitted mean is still in scope.
const WINDOW_BURST_SAFETY: f32 = 1.5;

/// Floor on effective loss for the rate law, so a barely-lossy-but-not-clean
/// channel does not divide by zero (it clamps to the lightest protection
/// anyway).
const MIN_EFFECTIVE_LOSS: f32 = 1.0e-3;

/// The RLC coding configuration the controller selects.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RlcDecision {
    /// Whether the RLC code is active at all. `false` is the disable-on-clean
    /// state: ship source symbols only and lean on the ARQ floor.
    pub coding_on: bool,
    /// Sliding-window size: how many recent source symbols a repair spans.
    pub window: u16,
    /// Source symbols between repairs. Code rate is `step / (step + 1)`.
    pub step: u16,
    /// Coefficient density threshold (0..=15): each coefficient is nonzero with
    /// probability `(dt + 1) / 16`.
    pub dt: u8,
}

/// The RLC parameters the sensors call for at the base (near-zero-RTT) rate
/// margin, before any hysteresis.
pub fn rlc_target(s: &SensorSnapshot) -> RlcDecision {
    rlc_target_with_margin(s, BASE_RATE_MARGIN)
}

/// The RLC parameters at an explicit rate-law `margin` (the controller scales it
/// by the measured round trip; see `rate_margin_for_rtt`).
pub fn rlc_target_with_margin(s: &SensorSnapshot, margin: f32) -> RlcDecision {
    if is_clean(s) {
        // Disable-on-clean: coding off, but keep nominal knobs so re-arming
        // (which copies these) starts from a sane window / cadence.
        return RlcDecision {
            coding_on: false,
            window: WINDOW_MIN,
            step: STEP_MAX,
            dt: DEFAULT_DT,
        };
    }
    // The RLC FEC provisions its RATE against the measured LOSS RATE - the one
    // channel signal that is reliably measured here (the Gilbert-Elliott fit's
    // marginal loss). In principle congestion loss should bias the rate LIGHTER
    // (QUIC-FEC: FEC on a congested/bulk path steals bottleneck bandwidth and
    // deepens the queue), but acting on that needs a TRUSTWORTHY queue signal:
    // this transport's one-way-trip time folds in the receiver's own decode
    // backlog (it stamps arrival at process time and decodes synchronously), so
    // under loss it reads a false congestion, and a false positive that lightened
    // the rate would collapse recovery. So the congestion share is measured and
    // reported but does NOT drive the rate; the rate tracks the loss rate, which
    // is robust. Density and window carry the burst signal, which IS reliable.
    let loss_rate = s.loss.max(MIN_EFFECTIVE_LOSS);
    // Rate law R = T/(T+B): redundancy 1/(step+1) >= margin * loss_rate, so
    // step <= 1/(margin * loss_rate) - 1. Heavier loss -> smaller step; a larger
    // margin (a more expensive round trip) also shrinks step.
    let step = ((1.0 / (margin.max(1.0) * loss_rate) - 1.0).floor() as i32)
        .clamp(STEP_MIN as i32, STEP_MAX as i32) as u16;
    // Window spans the fitted burst: w >= mean_burst * step, with a margin, and
    // at least 2*step so consecutive repairs overlap.
    let mean_burst = (s.burstiness * 16.0).max(1.0);
    let span = (mean_burst * step as f32 * WINDOW_BURST_SAFETY).ceil() as u32;
    let window = span
        .max(2 * step as u32)
        .clamp(WINDOW_MIN as u32, WINDOW_MAX as u32) as u16;
    // Denser coefficients under BURSTY loss: a base 0.5 density plus half the
    // burstiness, mapped onto 0..=15 and floored at 4 (a too-sparse repair over a
    // small window can miss the lost symbol entirely). Density tracks burstiness
    // alone - congestion does not make a loss more recoverable, and denser
    // repairs over a congested path are the wrong response (less FEC, not more).
    let density = (0.5 + 0.5 * s.burstiness).clamp(0.0, 1.0);
    let dt = ((density * 15.0).round() as u8).clamp(4, 15);
    RlcDecision { coding_on: true, window, step, dt }
}

/// Whether `a` is strictly more protective than `b`: coding turning on, a
/// smaller step (more repairs), a larger window (longer reach), or denser
/// coefficients. Escalations in any of these directions are applied at once;
/// only de-escalations wait out the hold.
fn more_protective(a: &RlcDecision, b: &RlcDecision) -> bool {
    // `a` must have coding on to be more protective at all; then it is more
    // protective if `b`'s coding is off, or any knob in `a` is set more
    // aggressively (smaller step, larger window, denser coefficients).
    a.coding_on
        && (!b.coding_on || a.step < b.step || a.window > b.window || a.dt > b.dt)
}

/// Immediate-up / conservative-down controller for the RLC parameters. It
/// raises protection (coding on, smaller step, larger window, denser
/// coefficients) the instant the sensors call for it - cheap insurance against
/// loss - but only lowers it after `hold` consecutive ticks that all want less,
/// so a brief quiet spell does not strip protection. Turning coding OFF
/// entirely (disable-on-clean) is the riskiest de-escalation, so it needs the
/// longer `clean_hold` sustained-clean window, exactly as the block-RS policy
/// guards the drop to Passthrough.
#[derive(Debug, Clone)]
pub struct RlcController {
    state: RlcDecision,
    down_streak: u32,
    hold: u32,
    clean_hold: u32,
    /// Measured round trip (ms); scales the rate-law margin so an expensive NAK
    /// buys heavier FEC. Zero (the default) is the near-zero-RTT base margin.
    rtt_ms: f32,
    /// Latency-priority floor: never relax FEC below this baseline protection
    /// (and never disable coding). disable-on-clean and an over-light rate save
    /// redundancy bandwidth on a quiet assessment but pay an ARQ round trip on
    /// the next loss, which head-of-line-stalls in-order delivery. For the
    /// latency-priority code (RLC in the unified transport) the step is clamped at
    /// `floor_step` (the configured baseline) so an isolated loss always recovers
    /// in-window; the controller may still escalate HEAVIER under high loss. Zero
    /// disables the floor (the default bulk behaviour with disable-on-clean).
    floor_step: u16,
    /// Latency-priority floor on the coding WINDOW: never shrink the window below
    /// this baseline span. A clean assessment otherwise collapses the window to
    /// the minimum, and a short window cannot span a loss CLUSTER even at heavy
    /// redundancy (too few repairs reach back over the burst), so clustered losses
    /// fall to ARQ. Keeping the baseline span lets the in-window repairs cover a
    /// burst. Zero disables the floor.
    floor_window: u16,
}

impl RlcController {
    /// A controller starting from active coding at the given parameters, with
    /// `hold` lower-demand ticks required before relaxing a knob and
    /// `4 * hold` sustained-clean ticks before disabling coding entirely.
    pub fn new(window: u16, step: u16, dt: u8, hold: u32) -> Self {
        let hold = hold.max(1);
        Self {
            state: RlcDecision { coding_on: true, window, step, dt },
            down_streak: 0,
            hold,
            clean_hold: hold.saturating_mul(4).max(hold),
            rtt_ms: 0.0,
            floor_step: 0,
            floor_window: 0,
        }
    }

    /// Like [`new`](Self::new) but with an explicit `clean_hold`.
    pub fn with_holds(window: u16, step: u16, dt: u8, hold: u32, clean_hold: u32) -> Self {
        let hold = hold.max(1);
        Self {
            state: RlcDecision { coding_on: true, window, step, dt },
            down_streak: 0,
            hold,
            clean_hold: clean_hold.max(hold),
            rtt_ms: 0.0,
            floor_step: 0,
            floor_window: 0,
        }
    }

    /// Set the measured round trip (ms), which scales the rate-law margin.
    pub fn set_rtt_ms(&mut self, rtt_ms: f32) {
        self.rtt_ms = rtt_ms.max(0.0);
    }

    /// Enable the latency-priority floor at the current baseline step: the
    /// controller never relaxes FEC lighter than this (nor disables coding), so an
    /// isolated loss always has an in-window repair and never falls to an ARQ
    /// round trip that stalls in-order delivery. The controller may still escalate
    /// HEAVIER under high loss. Call once at construction, before any feedback.
    pub fn set_latency_floor(&mut self) {
        self.floor_step = self.state.step.max(1);
        self.floor_window = self.state.window;
    }

    /// The current coding configuration.
    pub fn current(&self) -> RlcDecision {
        self.state
    }

    /// Fold one channel assessment and return the configuration to apply.
    pub fn decide(&mut self, s: &SensorSnapshot) -> RlcDecision {
        let mut t = rlc_target_with_margin(s, rate_margin_for_rtt(self.rtt_ms));
        // Latency-priority floor: a clean / low-loss assessment would disable
        // coding or relax the rate too light, so the next loss falls to an ARQ
        // round trip that head-of-line-stalls the in-order stream. Hold coding on
        // and clamp the step at the baseline (never lighter), so an isolated loss
        // always recovers in-window; escalation to a HEAVIER step under high loss
        // still applies (the clamp only caps the light side).
        if self.floor_step > 0 {
            t.coding_on = true;
            t.step = t.step.min(self.floor_step);
            t.window = t.window.max(self.floor_window);
        }
        if more_protective(&t, &self.state) {
            // Escalate every knob that wants more protection, immediately.
            self.state.coding_on |= t.coding_on;
            if t.coding_on {
                self.state.step = self.state.step.min(t.step);
                self.state.window = self.state.window.max(t.window);
                self.state.dt = self.state.dt.max(t.dt);
            }
            self.down_streak = 0;
        } else if t == self.state {
            self.down_streak = 0;
        } else {
            // Lower demand: only relax after a sustained quiet run. Disabling
            // coding (the riskiest step) needs the longer clean_hold window;
            // relaxing a knob while coding stays on uses the shorter hold.
            self.down_streak += 1;
            let threshold = if !t.coding_on { self.clean_hold } else { self.hold };
            if self.down_streak >= threshold {
                self.state = t;
                self.down_streak = 0;
            }
        }
        self.state
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn clean() -> SensorSnapshot {
        SensorSnapshot::default()
    }

    fn lossy(loss: f32, burstiness: f32) -> SensorSnapshot {
        SensorSnapshot { loss, burstiness, ..SensorSnapshot::default() }
    }

    #[test]
    fn clean_link_disables_coding() {
        let d = rlc_target(&clean());
        assert!(!d.coding_on, "a provably-clean link must disable RLC coding");
    }

    #[test]
    fn heavier_loss_shrinks_step_toward_more_repairs() {
        // The rate law: more loss -> smaller step (more repairs, lower rate).
        let light = rlc_target(&lossy(0.02, 0.0));
        let heavy = rlc_target(&lossy(0.30, 0.0));
        assert!(light.coding_on && heavy.coding_on);
        assert!(
            heavy.step < light.step,
            "heavier loss must lower step (more repairs): heavy {} vs light {}",
            heavy.step,
            light.step,
        );
        // 30% loss provisioned at 1.4x margin needs redundancy ~0.42 -> step ~1.
        assert!(heavy.step <= 2, "30% loss should be near the heaviest rate, got step {}", heavy.step);
    }

    #[test]
    fn rate_law_covers_effective_loss() {
        // The provisioned redundancy 1/(step+1) must be at least the loss it is
        // covering (the whole point of the rate law), at a representative loss.
        for &loss in &[0.05f32, 0.10, 0.20] {
            let d = rlc_target(&lossy(loss, 0.0));
            let redundancy = 1.0 / (d.step as f32 + 1.0);
            assert!(
                redundancy >= loss,
                "redundancy {redundancy} must cover loss {loss} (step {})",
                d.step,
            );
        }
    }

    #[test]
    fn longer_bursts_grow_the_window() {
        // burstiness encodes mean_burst / 16: a longer fitted burst must widen
        // the window so enough repairs span the burst.
        let short = rlc_target(&lossy(0.10, 0.1)); // mean burst ~1.6
        let long = rlc_target(&lossy(0.10, 0.6)); // mean burst ~9.6
        assert!(
            long.window > short.window,
            "longer bursts must widen the window: long {} vs short {}",
            long.window,
            short.window,
        );
    }

    #[test]
    fn density_rises_with_burstiness() {
        // Clustered (bursty) loss calls for denser repairs; congestion does not
        // touch density (it lightens the rate instead).
        let mild = rlc_target(&lossy(0.10, 0.0));
        let bursty = rlc_target(&lossy(0.10, 0.9));
        assert!(bursty.dt > mild.dt, "burstiness must raise density: {} vs {}", bursty.dt, mild.dt);
    }

    #[test]
    fn higher_rtt_provisions_heavier_fec() {
        // The same loss, at a higher round trip, gets a smaller step (heavier
        // FEC): an expensive NAK is worth avoiding with more in-window recovery.
        let s = lossy(0.10, 0.0);
        let light = rlc_target_with_margin(&s, rate_margin_for_rtt(0.0)); // ~loopback
        let heavy = rlc_target_with_margin(&s, rate_margin_for_rtt(2.0)); // real network
        assert!(
            heavy.step < light.step,
            "a higher round trip must shrink step (heavier FEC): {} vs {}",
            heavy.step,
            light.step,
        );
        // And the margin is monotonic in RTT, capped.
        assert!(rate_margin_for_rtt(5.0) >= rate_margin_for_rtt(1.0));
        assert!(rate_margin_for_rtt(1000.0) <= MAX_RATE_MARGIN + 1e-6);
    }

    #[test]
    fn rate_tracks_loss_not_congestion_classification() {
        // The rate provisions against the measured loss rate and is robust to the
        // congestion classification (whose timing signal is unreliable on this
        // transport): the same measured loss yields the same rate whether classed
        // wireless or congestion, so a false-high congestion reading cannot
        // collapse the code.
        let wireless = rlc_target(&SensorSnapshot {
            loss: 0.20,
            congestion_fraction: 0.0,
            ..SensorSnapshot::default()
        });
        let congested = rlc_target(&SensorSnapshot {
            loss: 0.20,
            congestion_fraction: 0.9,
            ..SensorSnapshot::default()
        });
        assert_eq!(
            congested.step, wireless.step,
            "the rate must track the loss rate, not the congestion classification",
        );
    }

    #[test]
    fn controller_escalates_immediately_on_loss() {
        let mut c = RlcController::new(16, STEP_MAX, DEFAULT_DT, 8);
        // A loss spike must shrink step (raise protection) on the very first tick.
        let before = c.current().step;
        let d = c.decide(&lossy(0.25, 0.5));
        assert!(d.coding_on, "must keep coding on under loss");
        assert!(
            d.step < before,
            "must raise protection (smaller step) at once: {} -> {}",
            before,
            d.step,
        );
    }

    #[test]
    fn controller_holds_protection_through_a_blip() {
        let mut c = RlcController::with_holds(16, 4, 15, 4, 16);
        c.decide(&lossy(0.25, 0.5)); // escalate
        let escalated = c.current();
        // One clean tick must NOT immediately relax protection.
        let d = c.decide(&clean());
        assert_eq!(d.step, escalated.step, "a single clean tick must not relax step");
        assert!(d.coding_on, "a single clean tick must not disable coding");
    }

    #[test]
    fn controller_disables_coding_only_after_sustained_clean() {
        let mut c = RlcController::with_holds(16, 4, 15, 2, 6);
        c.decide(&lossy(0.25, 0.5)); // escalate, coding on
        // Fewer than clean_hold clean ticks: coding stays on.
        for i in 0..5 {
            let d = c.decide(&clean());
            assert!(d.coding_on, "coding dropped too early at clean tick {i}");
        }
        // The clean_hold-th sustained-clean tick disables coding.
        let d = c.decide(&clean());
        assert!(!d.coding_on, "sustained clean must finally disable coding");
    }

    #[test]
    fn latency_floor_holds_baseline_through_clean() {
        // The latency-priority floor holds FEC on AND never relaxes the step
        // lighter than the baseline (4) through a sustained-clean run - the fix
        // for the ARQ-latency cliff where the controller would otherwise relax to
        // an over-light rate (or disable coding) and pay an ARQ round trip on the
        // next loss.
        let mut c = RlcController::with_holds(16, 4, 15, 2, 4);
        c.set_latency_floor();
        c.decide(&lossy(0.25, 0.5)); // escalate under loss (step drops below 4)
        for i in 0..20 {
            let d = c.decide(&clean());
            assert!(d.coding_on, "floor must hold FEC on at clean tick {i}");
            assert!(d.step <= 4, "floor must not relax lighter than baseline 4 (got {})", d.step);
            assert!(d.window >= 16, "floor must not shrink window below baseline 16 (got {})", d.window);
        }
        // The floor still escalates HEAVIER than the baseline under high loss.
        let heavy = c.decide(&lossy(0.30, 0.5));
        assert!(heavy.step < 4, "floor must still escalate heavier under loss (got {})", heavy.step);
        // The same sustained-clean run disables coding WITHOUT the floor (baseline).
        let mut base = RlcController::with_holds(16, 4, 15, 2, 4);
        base.decide(&lossy(0.25, 0.5));
        let mut disabled = false;
        for _ in 0..20 {
            if !base.decide(&clean()).coding_on {
                disabled = true;
                break;
            }
        }
        assert!(disabled, "default controller must disable-on-clean (the baseline)");
    }
}