summaryrefslogtreecommitdiff
path: root/src/ipcpd
diff options
context:
space:
mode:
Diffstat (limited to 'src/ipcpd')
-rw-r--r--src/ipcpd/unicast/ca.c6
-rw-r--r--src/ipcpd/unicast/ca.h6
-rw-r--r--src/ipcpd/unicast/ca/mb-ecn.c371
-rw-r--r--src/ipcpd/unicast/ca/mb-ecn.h3
-rw-r--r--src/ipcpd/unicast/ca/nop.c6
-rw-r--r--src/ipcpd/unicast/ca/nop.h3
-rw-r--r--src/ipcpd/unicast/ca/ops.h4
-rw-r--r--src/ipcpd/unicast/ca/tests/CMakeLists.txt34
-rw-r--r--src/ipcpd/unicast/ca/tests/mb_ecn_lab_test.c1294
-rw-r--r--src/ipcpd/unicast/ca/tests/mb_ecn_test.c723
-rw-r--r--src/ipcpd/unicast/cap.c212
-rw-r--r--src/ipcpd/unicast/cap.h16
-rw-r--r--src/ipcpd/unicast/dt.c35
-rw-r--r--src/ipcpd/unicast/tests/cap_test.c434
14 files changed, 2112 insertions, 1035 deletions
diff --git a/src/ipcpd/unicast/ca.c b/src/ipcpd/unicast/ca.c
index f2a65fff..290c817e 100644
--- a/src/ipcpd/unicast/ca.c
+++ b/src/ipcpd/unicast/ca.c
@@ -194,7 +194,7 @@ time_t ca_ctx_update_snd(void * _ctx,
{
struct ca_ctx * ctx = _ctx;
- return ca.ops->ctx_update_snd(ctx->pol, len, lecn, ftag);
+ return ca.ops->ctx_update_snd(ctx->pol, len, lecn, ctx->refs, ftag);
}
bool ca_ctx_update_rcv(void * _ctx,
@@ -244,9 +244,9 @@ void ca_ctx_rtt(void * _ctx,
int ca_calc_ecn(size_t queued,
uint8_t * ecn,
qoscube_t qc,
- size_t len)
+ size_t mean)
{
- return ca.ops->calc_ecn(queued, ecn, qc, len);
+ return ca.ops->calc_ecn(queued, ecn, qc, mean);
}
bool ca_marks_ecn(void)
diff --git a/src/ipcpd/unicast/ca.h b/src/ipcpd/unicast/ca.h
index fc6de445..188fb08a 100644
--- a/src/ipcpd/unicast/ca.h
+++ b/src/ipcpd/unicast/ca.h
@@ -67,10 +67,14 @@ void ca_ctx_rtt(void * ctx,
uint64_t now,
uint64_t rtt);
+/*
+ * Marks congestion from the egress queue. Both queued and mean are
+ * in bytes, so their ratio is the queue depth in packets.
+ */
int ca_calc_ecn(size_t queued,
uint8_t * ecn,
qoscube_t qc,
- size_t len);
+ size_t mean);
bool ca_marks_ecn(void);
diff --git a/src/ipcpd/unicast/ca/mb-ecn.c b/src/ipcpd/unicast/ca/mb-ecn.c
index a4c9f29e..59f1cae5 100644
--- a/src/ipcpd/unicast/ca/mb-ecn.c
+++ b/src/ipcpd/unicast/ca/mb-ecn.c
@@ -43,90 +43,152 @@
* Multi-bit ECN congestion avoidance: a rate-based controller. The
* sender paces a token bucket at a rate steered by graded ECN
* feedback, so the backoff is proportional to the congestion. A
- * backlogged flow ramps in slow start to find the path capacity, then
- * settles into AIMD around its fair share. There is no sliding window
- * and no per-flow timer; the control runs on sends.
+ * backlogged flow ramps in slow start to find the path capacity,
+ * then settles into AIMD around its fair share. There is no sliding
+ * window and no per-flow timer; the control runs on sends.
*
- * Every rate step is scaled by elapsed wall-clock time (Δt), not by
- * packet count, so the per-second dynamics are RTT-independent. The
- * receiver's averaging window and the sender's feedback staleness both
- * stretch with the flow's byte rate, so a slow flow is measured and
- * controlled like a fast one; CA_RATE_MIN only bounds those horizons
- * (window <= CA_TW_ABSMAX, TTL ~8 s).
+ * Rate law, per control step of dt seconds (r bytes/s, m the mark
+ * in ece units, m_ref = CA_ECE_REF, ai the additive slope):
*
- * The ramp clock ss_tc = 2 * RTT holds slow-start overshoot near 1.65x
- * (e^1/2): it seeds from the declared max_rtt and then tracks the
- * heartbeat's measured RTT. Feedback silence past the staleness horizon
- * leaves slow start; a sustained run of it restarts at the floor.
+ * slow start dr = r * dt / ss_tc
+ * increase dr = (ai + r / T_probe) * dt
+ * decrease dr = -r * (min(m, CA_ECE_MAX) / m_ref) * dt + L,
+ * cut capped at r/2
+ * lead L = -dm * r / (m_ref * CA_MD_KD_DIV)
*
- * The floor and the AI slope scale with the path: forwarders stamp
- * their measured link capacity into the PCI (cap.c), the receiver
- * feeds the path MIN back with the ece, and the sender derives
- * rate_min = ai_rate = C / 32, clamped to [CA_RATE_MIN, CA_RMIN_MAX],
- * falling back to those defaults when the signal goes stale.
+ * dm is the mark's step since the last decrease, clamped to
+ * +-m_ref. On a rise L joins the cut before the r/2 cap; on a
+ * fall it returns after that cap, bounded on its own to
+ * +-r / CA_MD_KD_DIV, so a full cut is never handed back in one
+ * step.
+ *
+ * Every step scales by elapsed wall-clock time, not by packet
+ * count, so the per-second dynamics are RTT-independent.
+ *
+ * Pacer: a virtual clock vt advances at r; a packet's start tag is
+ * max(tag, vt) and it waits (tag - vt) / r.
+ *
+ * Receiver: ece is the time integral of ecn over a pricing window,
+ * ece = integral(ecn dt) / T. The window is a per-layer constant so
+ * every flow prices one bottleneck alike; it stretches only for a
+ * flow too slow to fill it with samples.
+ *
+ * Marking (mb_ecn_calc_ecn): ecn is the quarter-log2 of the queue
+ * measured in mark units U (U = CA_MARK_KNEE * mean), so the mark is
+ * a log-scale queue depth. Equilibrium is where increase balances
+ * decrease:
+ *
+ * ecn* = (m_ref / 32) * (ai * n / C + 1 / T_probe) = n + 2
+ *
+ * for n backlogged flows, i.e. a standing queue of 2^((n+2)/4) * U.
+ * This is the zero-delay fixpoint; feedback delay raises the real
+ * standing queue above it.
*/
+/* ECE fixed point */
#define CA_SHFT 5 /* ece fixed point: 32 * ecn */
-#define CA_TW_MIN (1ULL << 20) /* min mean window ~1.05 ms */
-#define CA_TW_INIT (1ULL << 26) /* initial mean window ~67ms */
+
+/* Receiver averaging window */
+#define CA_TW (1ULL << 26) /* pricing window ~67 ms */
+#define CA_TW_MIN (4ULL * MILLION) /* pricing window floor 4 ms */
+#define CA_TW_RTT_MUL 2 /* T_w = 2 * layer RTT */
#define CA_TW_ABSMAX (1ULL << 32) /* window ceiling ~4.3 s */
/* Quiet horizon, in windows (1 << shift): gap restart and the TTLs. */
#define CA_TW_GAP_SHFT 2
-#define CA_N_TARGET 16 /* target packets per window */
-#define CA_RX_WBYTES (CA_N_TARGET * 1000ULL) /* target bytes/window */
+#define CA_RX_WBYTES 16000ULL /* 16 pkts x 1000 B a window */
#define CA_RX_WCLOSE (2 * CA_RX_WBYTES) /* byte-triggered early close */
#define CA_TW_SM_SHFT 2 /* window EWMA weight 1/4 */
-#define CA_MARK_Q 4 /* mark quantum (packets) */
+/* Congestion marking */
+#define CA_MARK_KNEE 1 /* mark onset (packets) */
+
+/* Rate machine */
#define CA_RATE_MIN (1ULL << 13) /* 8 KiB/s rate floor */
#define CA_RATE_INIT (1ULL << 16) /* slow start seed 64 KiB/s */
/* Rate cap; also keeps rate * dt and rate * rise below 2^64. */
#define CA_RATE_MAX (1ULL << 37)
#define CA_INV_SHFT 32 /* reciprocal-rate fixp */
-#define CA_AI_RATE (1ULL << 16) /* 64 KiB/s^2 additive inc */
+#define CA_AI_RATE (1ULL << 17) /* 128 KiB/s^2 additive inc */
#define CA_PROBE_TC (8ULL * BILLION) /* proportional probe TC 8s */
#define CA_ECE_REF (16 << CA_SHFT) /* full congestion: ecn 16 */
-#define CA_MD_KD_DIV 4 /* one-sided lead gain 1/4 */
+/* Decrease saturation, and the level below which the hold clears. */
+#define CA_ECE_MAX (2 * CA_ECE_REF) /* ecn 32 */
+#define CA_MD_KD_DIV 16 /* lead gain 1/16 */
+
+/* Control cadence */
#define CA_DT_CTRL (BILLION / 1000) /* min rate-update spacing */
#define CA_DT_CAP (BILLION / 20) /* idle-resume Δt clamp 50ms */
-/* Floor of the rate-relative feedback staleness (ctx->ece_ttl). */
-#define CA_ECE_TTL ((1 << CA_TW_GAP_SHFT) * CA_TW_INIT)
+#define CA_IDLE_PKTS 4 /* idle: gap over 4 packets */
+/* Feedback staleness floor; ctx->ece_ttl rides above it by rate. */
+#define CA_ECE_TTL (1ULL << 28) /* ~268 ms */
+
+/* Slow start */
#define CA_SS_RTT_MUL 2 /* ss_tc = 2 * layer RTT */
#define CA_SS_TC_MIN (BILLION / 1000) /* ramp floor 1 ms */
#define CA_SS_TC_MAX (4ULL * BILLION) /* ramp ceiling 4 s */
-#define CA_HB_MIN (40 * MILLION) /* heartbeat interval floor */
-#define CA_HB_LOSS 4 /* stale horizons -> restart */
#define CA_RTT_SHFT 2 /* ss_tc EWMA weight 1/4 */
+#define CA_SS_TC_GRW 1 /* ramp climb cap 2x a sample */
#define CA_SS_RTT_DEF 200 /* default layer RTT (ms) */
-#define CA_WASH_BKT (BILLION / 32) /* washout bucket ~31 ms */
-#define CA_WASH_SHFT 2 /* damp 1/4 of bucket change */
+/* Heartbeat */
+#define CA_HB_MIN (40 * MILLION) /* heartbeat interval floor */
+#define CA_HB_LOSS 4 /* stale horizons -> restart */
+
+/* Path capacity */
#define CA_CAP_SHFT 5 /* floor = capacity / 32 */
#define CA_CAP_SM_SHFT 1 /* capacity EWMA weight 1/2 */
/* Outlives ece_ttl 16x: onset-fresh fcap re-seeds each episode. */
#define CA_CAP_TTL_SHFT 4
#define CA_RMIN_MAX (1ULL << 32) /* derived floor ceiling */
-#define CA_SND_WIN CA_TW_INIT /* sender util window ~67ms */
+/* Sender utilisation */
+#define CA_SND_WIN (1ULL << 26) /* sender util window ~67 ms */
#define CA_USE_NUM 3 /* backlogged: offered >= */
#define CA_USE_DEN 4 /* 3/4 * window-start rate */
-#define CA_HDRM_MARKS 4 /* ceiling ~2x offered load */
#define CA_SND_DEC_SHFT 4 /* offered max-filter 1/16 */
#define CA_SND_DEC_CAP 16 /* bound gapped-close decay */
#define CA_SND_BYT_MAX (1ULL << 33) /* offered-byte saturation */
+#define CA_PAC_DEN 4 /* backlogged: 1/4 deferred */
/*
* Retuning invariants (pinned by the unit tests):
- * - (1 << CA_TW_GAP_SHFT) * CA_TW_INIT > S * BILLION / CA_RATE_MIN, or
+ * - (1 << CA_TW_GAP_SHFT) * CA_TW > S * BILLION / CA_RATE_MIN, or
* a floor-rate flow's onset restart-loops (S ~ one MTU; both ns).
* - CA_RX_WBYTES * BILLION / CA_RATE_MIN < CA_TW_ABSMAX: the
* floor-rate window must clear the ceiling.
+ * - CA_TW < CA_RX_WBYTES * BILLION / CA_RATE_MIN: at the rate
+ * floor the sample budget, not the horizon, sizes the window.
+ * - CA_TW << CA_TW_GAP_SHFT <= CA_ECE_TTL: the estimator must
+ * not call a gap fresh that the sender still counts as live.
+ * - CA_ECE_TTL > S * BILLION / CA_RATE_MIN: the idle cap clears a
+ * floor-rate flow's inter-send gap, so pacing never reads as idle.
+ * - CA_DT_CAP < CA_ECE_TTL: the idle clamp needs the TTL above it,
+ * or every slow flow reads idle on every send.
* - CA_RATE_MAX * CA_DT_CAP, the folded lead * inv_rate at
* CA_RATE_MIN, and owed * BILLION (owed clamped in mb_ecn_snd) all
* keep the pacer arithmetic below 2^64.
- * - CA_DT_CTRL < CA_WASH_BKT < CA_DT_CAP: control cadence under the
- * washout bucket under the sparse-step cutoff.
* - CA_RATE_MIN <= CA_RATE_INIT and CA_RMIN_MAX < CA_RATE_MAX.
+ * - cap_enc(16 * mean) - cap_enc(mean) == CA_ECE_REF >> CA_SHFT: a
+ * queue of 16 packets is what reads as full congestion.
+ * - CA_MD_KD_DIV sets the lead gain. The term acts both ways (cut on
+ * a rise, give back on a fall), which cancels the DC bias a
+ * one-sided term would rectify into a standing rate difference
+ * between flows pricing one queue; that is what lets the gain run
+ * at 1/16 instead of the deadzone below 1/8.
+ * - T_w = clamp(CA_TW_RTT_MUL * RTT, CA_TW_MIN, CA_TW) scales only
+ * the receiver pricing window; CA_ECE_TTL, CA_SND_WIN, CA_DT_CAP
+ * and CA_DT_CTRL are absolute and must not be derived from it.
+ * - The gap-restart horizon is floored at CA_ECE_TTL, so a
+ * floor-rate flow's inter-packet gap never reads as an onset.
+ * - The ai_hold release threshold equals the decrease saturation
+ * clamp: a standing mark that is a legal equilibrium must be able
+ * to clear the hold.
+ *
+ * Structural invariants (not exercised by the unit tests):
+ * - CA_MARK_KNEE <= 4: the full decrease range must fit the ring
+ * (SSM_RBUFF_SIZE, not visible from this file).
+ * - ecn* = 2 + n holds for n <= 29 (the decrease clamp) and only
+ * with live capacity feedback.
*/
struct mb_ecn_ctx {
@@ -148,8 +210,6 @@ struct mb_ecn_ctx {
uint64_t ai_rate; /* additive-increase slope (B/s^2) */
uint64_t ece_ttl; /* how long feedback stays valid (ns) */
uint64_t ss_tc; /* slow-start time constant (ns) */
- uint64_t r_bkt; /* rate snapshot at last washout bucket */
- uint64_t wash_acc; /* washout bucket time accumulator (ns) */
uint64_t dec_acc; /* sub-ms decrease time carried (ns) */
uint64_t inv_rate; /* fixed-point 1/rate for pacing */
uint64_t vt; /* virtual service clock (bytes) */
@@ -159,12 +219,15 @@ struct mb_ecn_ctx {
uint64_t last_fb; /* last congestion feedback (ns) */
uint64_t last_sig; /* last liveness signal, incl. hb (ns) */
uint64_t n_fb; /* feedback updates received */
+ uint64_t n_rtt; /* heartbeat RTT samples folded */
uint64_t last_hb; /* last heartbeat emitted (ns) */
uint64_t last_res; /* last resume from idle (ns) */
uint64_t last_loc; /* last local mark seen (ns) */
uint64_t last_cap; /* last capacity applied (ns) */
uint64_t snd_byt; /* bytes offered this window (capped) */
+ size_t snd_flows; /* flows sharing the ctx, >= 1 */
+ uint64_t snd_pac; /* bytes the pacer held back this win */
uint64_t snd_win; /* utilisation window start (ns) */
uint64_t snd_r0; /* rate at window start */
uint64_t snd_rate; /* max-filter of offered rate (B/s) */
@@ -180,7 +243,6 @@ struct mb_ecn_ctx {
uint64_t n_ttl; /* feedback aged out (TTL) */
uint64_t n_cap; /* capacity updates applied */
uint64_t n_loss; /* signal-loss cuts (collapse) */
- uint64_t n_rtt; /* heartbeat RTT samples folded */
uint64_t ss_peak; /* peak rate in slow start (bytes/s) */
};
@@ -188,6 +250,9 @@ struct mb_ecn_ctx {
static uint64_t mb_ecn_ss_tc = (uint64_t) CA_SS_RTT_MUL *
CA_SS_RTT_DEF * MILLION;
+/* Layer pricing window (ns), from the declared RTT. */
+static uint64_t mb_ecn_tw = CA_TW;
+
struct ca_ops mb_ecn_ca_ops = {
.ctx_create = mb_ecn_ctx_create,
.ctx_destroy = mb_ecn_ctx_destroy,
@@ -224,6 +289,8 @@ static uint64_t mb_ecn_ece_ttl(uint64_t rate)
void mb_ecn_init(uint32_t rtt_ms)
{
uint64_t tc;
+ uint64_t rtt;
+ uint64_t tw;
if (rtt_ms == 0) /* unspecified: safe default */
rtt_ms = CA_SS_RTT_DEF;
@@ -233,6 +300,17 @@ void mb_ecn_init(uint32_t rtt_ms)
tc = CA_SS_TC_MIN;
mb_ecn_ss_tc = tc;
+
+ rtt = (uint64_t) rtt_ms * MILLION;
+
+ tw = (uint64_t) CA_TW_RTT_MUL * rtt;
+ if (tw < CA_TW_MIN)
+ tw = CA_TW_MIN;
+
+ if (tw > CA_TW)
+ tw = CA_TW;
+
+ mb_ecn_tw = tw;
}
void * mb_ecn_ctx_create(void)
@@ -256,11 +334,10 @@ void * mb_ecn_ctx_create(void)
ctx->ai_rate = CA_AI_RATE;
ctx->ss_tc = mb_ecn_ss_tc;
ctx->ece_ttl = mb_ecn_ece_ttl(CA_RATE_INIT);
- ctx->r_bkt = CA_RATE_INIT;
ctx->inv_rate = mb_ecn_rate_inv(CA_RATE_INIT);
ctx->rx_ts = t;
ctx->rx_win = t;
- ctx->rx_tw = CA_TW_INIT;
+ ctx->rx_tw = mb_ecn_tw;
ctx->last_ts = t;
ctx->last_ctrl = t;
ctx->last_fb = t;
@@ -271,6 +348,7 @@ void * mb_ecn_ctx_create(void)
/* snd_win/last_ts re-seeded lazily on the first real send. */
ctx->snd_r0 = CA_RATE_INIT;
ctx->snd_rate = CA_RATE_INIT;
+ ctx->snd_flows = 1;
ctx->backlogged = true;
return (void *) ctx;
@@ -300,8 +378,6 @@ static void mb_ecn_slow_start(struct mb_ecn_ctx * ctx,
{
if (ctx->backlogged)
ctx->rate += ctx->rate * dta / ctx->ss_tc;
-
- ctx->r_bkt = ctx->rate;
}
/* Additive increase plus a rate-independent proportional probe. */
@@ -321,16 +397,18 @@ static void mb_ecn_increase(struct mb_ecn_ctx * ctx,
/*
* Multiplicative decrease: cut proportional to mark x elapsed time,
- * plus a one-sided lead that cuts extra while the mark is rising.
+ * plus a lead term on the mark's step, clamped and acting both ways.
*/
static void mb_ecn_decrease(struct mb_ecn_ctx * ctx,
uint64_t dtc)
{
uint64_t dtm;
uint64_t mark;
- uint64_t rise;
+ uint64_t step;
+ uint64_t lead;
uint64_t cut;
uint16_t m;
+ bool up;
m = ctx->tx_ece > 0 ? ctx->tx_ece
: (uint16_t) (ctx->tx_loc << CA_SHFT);
@@ -340,13 +418,20 @@ static void mb_ecn_decrease(struct mb_ecn_ctx * ctx,
return;
}
- mark = MIN(m, CA_ECE_REF);
+ mark = MIN(m, CA_ECE_MAX);
+
+ /* Lead on the mark step; the clamp bounds it to rate/KD. */
+ up = m > ctx->tx_ecp;
+ step = up ? m - ctx->tx_ecp : ctx->tx_ecp - m;
+ step = MIN(step, CA_ECE_REF);
+ lead = ctx->rate * step / (CA_ECE_REF * CA_MD_KD_DIV);
- /* One-sided lead: cut extra while the mark is still rising. */
- rise = m > ctx->tx_ecp ? MIN(m - ctx->tx_ecp, CA_ECE_REF) : 0;
- cut = ctx->rate * rise / (CA_ECE_REF * CA_MD_KD_DIV);
+ cut = up ? lead : 0;
- /* Honest elapsed ms; the sub-ms remainder carries over. */
+ /*
+ * Bank the remainder: at a 1 ms control cadence, truncating
+ * to whole milliseconds would drop up to half of every cut.
+ */
ctx->dec_acc += dtc;
dtm = ctx->dec_acc / MILLION;
ctx->dec_acc -= dtm * MILLION;
@@ -359,42 +444,16 @@ static void mb_ecn_decrease(struct mb_ecn_ctx * ctx,
cut = ctx->rate / 2;
ctx->rate -= cut;
- ctx->tx_ecp = m;
-}
-
-/*
- * Washout: once per wall-clock bucket, damp a fixed fraction of the
- * rate change over that bucket. Bucketed (not per-step) so it stays
- * cadence-independent; bounded so it cannot reverse a ramp. A sparse
- * step resets it, so a starved sender keeps its cut.
- */
-static void mb_ecn_washout(struct mb_ecn_ctx * ctx,
- uint64_t dtc,
- uint64_t dta)
-{
- if (dtc > (uint64_t) CA_DT_CAP) {
- ctx->r_bkt = ctx->rate;
- ctx->wash_acc = 0;
- return;
- }
-
- ctx->wash_acc += dta;
- if (ctx->wash_acc < (uint64_t) CA_WASH_BKT)
- return;
- if (ctx->rate > ctx->r_bkt)
- ctx->rate -= (ctx->rate - ctx->r_bkt) >> CA_WASH_SHFT;
- else
- ctx->rate += (ctx->r_bkt - ctx->rate) >> CA_WASH_SHFT;
+ if (!up)
+ ctx->rate += lead;
- ctx->r_bkt = ctx->rate;
- ctx->wash_acc = 0;
+ ctx->tx_ecp = m;
}
/* Offered-load ceiling backstop while source-limited. */
static void mb_ecn_ceiling(struct mb_ecn_ctx * ctx)
{
- unsigned code;
uint64_t hi;
if (ctx->backlogged) {
@@ -402,22 +461,16 @@ static void mb_ecn_ceiling(struct mb_ecn_ctx * ctx)
return;
}
- code = (unsigned) cap_enc(ctx->snd_rate) + CA_HDRM_MARKS;
- if (code > UINT8_MAX) /* keep the cast lossless */
- code = UINT8_MAX;
-
- hi = cap_dec((uint8_t) code);
- if (hi > CA_RATE_MAX)
- hi = CA_RATE_MAX;
-
+ /* Land on the backlog level; a ceiling above it never clears. */
+ hi = ctx->snd_rate > CA_RATE_MAX / CA_USE_DEN * CA_USE_NUM
+ ? (uint64_t) CA_RATE_MAX
+ : ctx->snd_rate * CA_USE_DEN / CA_USE_NUM;
if (hi < CA_RATE_MIN)
hi = CA_RATE_MIN;
ctx->src_limited = ctx->rate > hi;
- if (ctx->src_limited) {
- ctx->rate = hi;
- ctx->r_bkt = ctx->rate;
- }
+ if (ctx->src_limited)
+ ctx->rate = hi;
}
static void mb_ecn_ctrl(struct mb_ecn_ctx * ctx,
@@ -436,7 +489,6 @@ static void mb_ecn_ctrl(struct mb_ecn_ctx * ctx,
if (ctx->tx_cav) {
mb_ecn_increase(ctx, dta);
mb_ecn_decrease(ctx, dtc);
- mb_ecn_washout(ctx, dtc, dta);
} else {
mb_ecn_slow_start(ctx, dta);
}
@@ -475,6 +527,34 @@ static void mb_ecn_offered(struct mb_ecn_ctx * ctx,
ctx->snd_rate -= (ctx->snd_rate - offered) >> CA_SND_DEC_SHFT;
}
+/* Open a fresh utilisation window at t. */
+static void mb_ecn_win_open(struct mb_ecn_ctx * ctx,
+ uint64_t t)
+{
+ ctx->snd_win = t;
+ ctx->snd_byt = 0;
+ ctx->snd_pac = 0;
+ ctx->snd_r0 = ctx->rate;
+}
+
+/*
+ * Note the flow count; a window spanning two populations measures
+ * neither, so a change opens a fresh one.
+ */
+static void mb_ecn_flows(struct mb_ecn_ctx * ctx,
+ size_t flows,
+ uint64_t t)
+{
+ size_t n = flows > 0 ? flows : 1;
+
+ if (n == ctx->snd_flows)
+ return;
+
+ ctx->snd_flows = n;
+
+ mb_ecn_win_open(ctx, t);
+}
+
/*
* Close the utilisation window: set backlogged from the level test,
* fold offered into the max filter, then reset the window.
@@ -486,9 +566,20 @@ static void mb_ecn_win(struct mb_ecn_ctx * ctx,
uint64_t offered;
bool was = ctx->backlogged;
- offered = ctx->snd_byt * BILLION / elapsed;
+ /*
+ * snd_byt is the whole ctx's offered bytes but rate is what one
+ * flow may send, so share it out before either is compared.
+ */
+ offered = ctx->snd_byt * BILLION / elapsed / ctx->snd_flows;
- ctx->backlogged = offered * CA_USE_DEN >= ctx->snd_r0 * CA_USE_NUM;
+ /*
+ * Offered load is counted past the pacer, so it cannot tell a
+ * quiet source from one the pacer is holding back, and idle
+ * flows on the context drag it down. A window the pacer had to
+ * defer is rate-limited whatever the bytes say.
+ */
+ ctx->backlogged = offered * CA_USE_DEN >= ctx->snd_r0 * CA_USE_NUM
+ || ctx->snd_pac * CA_PAC_DEN >= ctx->snd_byt;
if (!was && ctx->backlogged) /* resume: fresh liveness baseline */
ctx->last_res = t;
@@ -498,9 +589,7 @@ static void mb_ecn_win(struct mb_ecn_ctx * ctx,
if (ctx->backlogged)
ctx->src_limited = false;
- ctx->snd_win = t;
- ctx->snd_byt = 0;
- ctx->snd_r0 = ctx->rate;
+ mb_ecn_win_open(ctx, t);
}
/* Age out congestion, local-mark and capacity signals once stale. */
@@ -520,7 +609,6 @@ static void mb_ecn_loss(struct mb_ecn_ctx * ctx,
if (ctx->rate < (uint64_t) CA_RATE_MIN)
ctx->rate = CA_RATE_MIN;
- ctx->r_bkt = ctx->rate;
ctx->inv_rate = mb_ecn_rate_inv(ctx->rate);
ctx->ece_ttl = mb_ecn_ece_ttl(ctx->rate);
ctx->last_sig = t;
@@ -602,6 +690,7 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx,
{
uint64_t dt;
uint64_t dtc;
+ uint64_t idle;
uint64_t s;
/* Lazy warm-up seed: packet #1 is never an idle resume. */
@@ -616,8 +705,16 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx,
dt = t - ctx->last_ts;
ctx->last_ts = t;
- /* Idle gap clears backlog before aging: no false loss on resume. */
- if (dt > (uint64_t) CA_DT_CAP)
+ /*
+ * Idle gap clears backlog before aging: no false loss on resume.
+ * Measured against the pacer's own spacing, so a flow paced
+ * slower than CA_DT_CAP per packet does not read as idle on
+ * every send, and bounded by the staleness horizon.
+ */
+ idle = CA_IDLE_PKTS * len * BILLION / ctx->rate;
+ idle = MAX(idle, (uint64_t) CA_DT_CAP);
+ idle = MIN(idle, (uint64_t) CA_ECE_TTL);
+ if (dt > idle)
ctx->backlogged = false;
mb_ecn_age(ctx, t);
@@ -643,6 +740,9 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx,
s = *ftag > ctx->vt ? *ftag : ctx->vt;
*ftag = s + len;
+ if (s > ctx->vt)
+ ctx->snd_pac += len;
+
ctx->lead = s - ctx->vt;
/* Reciprocal pacing; folded so any lead * rate stays in range. */
@@ -656,6 +756,7 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx,
time_t mb_ecn_ctx_update_snd(void * _ctx,
size_t len,
uint8_t lecn,
+ size_t flows,
uint64_t * ftag)
{
struct timespec now;
@@ -666,22 +767,32 @@ time_t mb_ecn_ctx_update_snd(void * _ctx,
t = TS_TO_UINT64(now);
+ mb_ecn_flows(ctx, flows, t);
+
mb_ecn_loc(ctx, lecn, t);
return mb_ecn_snd(ctx, len, t, ftag);
}
-/* Estimator idle, or a gap past ~4 current windows: restart fresh. */
+/* Estimator idle, or a quiet gap past the horizon: restart fresh. */
static bool mb_ecn_rcv_fresh(const struct mb_ecn_ctx * ctx,
uint64_t dt)
{
+ uint64_t gap;
+
if (ctx->rx_ece == 0 && ctx->rx_acc == 0)
return true;
- return dt > ctx->rx_tw << CA_TW_GAP_SHFT;
+ gap = ctx->rx_tw << CA_TW_GAP_SHFT;
+
+ return dt > MAX(gap, (uint64_t) CA_ECE_TTL);
}
-/* Size the next averaging window to ~CA_N_TARGET packets at this rate. */
+/*
+ * Size the next averaging window to ~16 packets at this rate, floored
+ * at the price horizon: a flow fast enough to fill the horizon
+ * integrates over CA_TW, a slower one stretches for its samples.
+ */
static void mb_ecn_resize(struct mb_ecn_ctx * ctx,
uint64_t win)
{
@@ -692,8 +803,8 @@ static void mb_ecn_resize(struct mb_ecn_ctx * ctx,
else
ctx->rx_tw -= (ctx->rx_tw - tw) >> CA_TW_SM_SHFT;
- if (ctx->rx_tw < CA_TW_MIN)
- ctx->rx_tw = CA_TW_MIN;
+ if (ctx->rx_tw < mb_ecn_tw)
+ ctx->rx_tw = mb_ecn_tw;
if (ctx->rx_tw > CA_TW_ABSMAX)
ctx->rx_tw = CA_TW_ABSMAX;
@@ -731,18 +842,15 @@ static bool mb_ecn_rcv(struct mb_ecn_ctx * ctx,
/* Dwell clamp: one packet weighs at most one window of mark. */
ctx->rx_acc += ecn * MIN(dt, ctx->rx_tw);
ctx->rx_byt += len;
- ctx->rx_cap = cap_min(ctx->rx_cap, cap);
-
- *ece = ctx->rx_ece;
+ ctx->rx_cap = cap_min(ctx->rx_cap, cap);
win = t - ctx->rx_win;
if (win < ctx->rx_tw) {
/* Early close once 2x target bytes arrive (speed-up). */
- if (ctx->rx_byt < CA_RX_WCLOSE)
- return false;
-
- if (win < CA_TW_MIN)
+ if (ctx->rx_byt < CA_RX_WCLOSE || win < mb_ecn_tw) {
+ *ece = ctx->rx_ece;
return false;
+ }
}
/* Time-integral mean over the actual window elapsed (never rx_tw). */
@@ -788,8 +896,8 @@ static void mb_ecn_ece(struct mb_ecn_ctx * ctx,
ctx->tx_ece = ece;
ctx->tx_cav = true; /* closed-loop feedback: leave slow start */
- /* A clean (unsaturated) signal means the queue drained: resume. */
- if (ece < (uint16_t) CA_ECE_REF)
+ /* An unsaturated signal means the queue drained: resume. */
+ if (ece < (uint16_t) CA_ECE_MAX)
ctx->ai_hold = false;
ctx->last_fb = t;
@@ -812,7 +920,7 @@ static void mb_ecn_ece(struct mb_ecn_ctx * ctx,
ctx->rate_min -= (ctx->rate_min - tgt)
>> CA_CAP_SM_SHFT;
- ctx->ai_rate = ctx->rate_min;
+ ctx->ai_rate = 2 * ctx->rate_min;
ctx->tx_cap = cap;
ctx->last_cap = t;
ctx->n_cap++;
@@ -871,6 +979,15 @@ void mb_ecn_ctx_rtt(void * _ctx,
if (tgt > (uint64_t) CA_SS_TC_MAX) /* at the real RTT, not the */
tgt = CA_SS_TC_MAX; /* declared worst case */
+ /*
+ * A control packet stuck behind a stalled reader returns an RTT
+ * worth seconds on a path worth milliseconds. Cap how far one
+ * sample carries the ramp, so a stall costs a step and a rise
+ * that holds still arrives within a few samples.
+ */
+ if (tgt > ctx->ss_tc << CA_SS_TC_GRW)
+ tgt = ctx->ss_tc << CA_SS_TC_GRW;
+
ctx->ss_tc += (tgt >> CA_RTT_SHFT) - (ctx->ss_tc >> CA_RTT_SHFT);
ctx->last_sig = now; /* liveness only: never ages the ece signal */
@@ -880,16 +997,28 @@ void mb_ecn_ctx_rtt(void * _ctx,
int mb_ecn_calc_ecn(size_t queued,
uint8_t * ecn,
qoscube_t qc,
- size_t len)
+ size_t mean)
{
- size_t q;
- uint8_t mark;
+ uint64_t u;
+ int q;
+ uint8_t mark;
- (void) len;
(void) qc;
- /* Saturate: a queue past 255 quanta must not wrap to a low mark. */
- q = queued / CA_MARK_Q;
+ if (queued == 0 || mean == 0)
+ return 0;
+
+ u = (uint64_t) CA_MARK_KNEE * mean;
+
+ /*
+ * Difference of two quarter-log2 codes is a log-scale ratio:
+ * the same queue in units of U marks the same on any link.
+ */
+ q = (int) cap_enc(queued) - (int) cap_enc(u);
+ if (q <= 0)
+ return 0;
+
+ /* Saturate: a deeper queue must not wrap to a low mark. */
mark = q > 255 ? (uint8_t) 255 : (uint8_t) q;
if (mark > *ecn)
diff --git a/src/ipcpd/unicast/ca/mb-ecn.h b/src/ipcpd/unicast/ca/mb-ecn.h
index 781e25a8..08bb542d 100644
--- a/src/ipcpd/unicast/ca/mb-ecn.h
+++ b/src/ipcpd/unicast/ca/mb-ecn.h
@@ -34,6 +34,7 @@ void mb_ecn_ctx_destroy(void * ctx);
time_t mb_ecn_ctx_update_snd(void * ctx,
size_t len,
uint8_t lecn,
+ size_t flows,
uint64_t * ftag);
bool mb_ecn_ctx_update_rcv(void * ctx,
@@ -57,7 +58,7 @@ void mb_ecn_ctx_rtt(void * ctx,
int mb_ecn_calc_ecn(size_t queued,
uint8_t * ecn,
qoscube_t qc,
- size_t len);
+ size_t mean);
ssize_t mb_ecn_print_stats(void * ctx,
char * buf,
diff --git a/src/ipcpd/unicast/ca/nop.c b/src/ipcpd/unicast/ca/nop.c
index e6965297..7a2f72db 100644
--- a/src/ipcpd/unicast/ca/nop.c
+++ b/src/ipcpd/unicast/ca/nop.c
@@ -48,11 +48,13 @@ void nop_ctx_destroy(void * ctx)
time_t nop_ctx_update_snd(void * ctx,
size_t len,
uint8_t lecn,
+ size_t flows,
uint64_t * ftag)
{
(void) ctx;
(void) len;
(void) lecn;
+ (void) flows;
(void) ftag;
return 0;
@@ -88,10 +90,10 @@ void nop_ctx_update_ece(void * ctx,
int nop_calc_ecn(size_t queued,
uint8_t * ecn,
qoscube_t qc,
- size_t len)
+ size_t mean)
{
(void) queued;
- (void) len;
+ (void) mean;
(void) ecn;
(void) qc;
diff --git a/src/ipcpd/unicast/ca/nop.h b/src/ipcpd/unicast/ca/nop.h
index 6ca206df..386a5310 100644
--- a/src/ipcpd/unicast/ca/nop.h
+++ b/src/ipcpd/unicast/ca/nop.h
@@ -32,6 +32,7 @@ void nop_ctx_destroy(void * ctx);
time_t nop_ctx_update_snd(void * ctx,
size_t len,
uint8_t lecn,
+ size_t flows,
uint64_t * ftag);
bool nop_ctx_update_rcv(void * ctx,
@@ -48,7 +49,7 @@ void nop_ctx_update_ece(void * ctx,
int nop_calc_ecn(size_t queued,
uint8_t * ecn,
qoscube_t qc,
- size_t len);
+ size_t mean);
extern struct ca_ops nop_ca_ops;
diff --git a/src/ipcpd/unicast/ca/ops.h b/src/ipcpd/unicast/ca/ops.h
index b01e8573..835fe0c5 100644
--- a/src/ipcpd/unicast/ca/ops.h
+++ b/src/ipcpd/unicast/ca/ops.h
@@ -33,6 +33,7 @@ struct ca_ops {
time_t (* ctx_update_snd)(void * ctx,
size_t len,
uint8_t lecn,
+ size_t flows,
uint64_t * ftag);
bool (* ctx_update_rcv)(void * ctx,
@@ -54,10 +55,11 @@ struct ca_ops {
uint64_t now,
uint64_t rtt);
+ /* queued and mean are bytes; their ratio is packets. */
int (* calc_ecn)(size_t queued,
uint8_t * ecn,
qoscube_t qc,
- size_t len);
+ size_t mean);
/* True if calc_ecn inspects the queue; gates the lookup. */
bool marks_ecn;
diff --git a/src/ipcpd/unicast/ca/tests/CMakeLists.txt b/src/ipcpd/unicast/ca/tests/CMakeLists.txt
index 6e42163d..20e2349d 100644
--- a/src/ipcpd/unicast/ca/tests/CMakeLists.txt
+++ b/src/ipcpd/unicast/ca/tests/CMakeLists.txt
@@ -42,3 +42,37 @@ target_link_libraries(${PARENT_DIR}_test PRIVATE ouroboros-common)
add_dependencies(build_tests ${PARENT_DIR}_test)
ouroboros_register_tests(TARGET ${PARENT_DIR}_test TESTS ${${PARENT_DIR}_tests})
+
+# The lab includes mb-ecn.c for its statics, so it needs its own binary
+create_test_sourcelist(${PARENT_DIR}_lab_tests test_lab_suite.c
+ mb_ecn_lab_test.c
+ )
+
+add_executable(${PARENT_DIR}_lab_test ${${PARENT_DIR}_lab_tests}
+ ${UNICAST_SOURCE_DIR}/cap.c
+ )
+
+target_include_directories(${PARENT_DIR}_lab_test PRIVATE
+ ${CMAKE_CURRENT_SOURCE_DIR}
+ ${CMAKE_CURRENT_BINARY_DIR}
+ ${CURRENT_SOURCE_PARENT_DIR}
+ ${CURRENT_BINARY_PARENT_DIR}
+ ${UNICAST_SOURCE_DIR}
+ ${UNICAST_BINARY_DIR}
+ ${CMAKE_SOURCE_DIR}/include
+ ${CMAKE_BINARY_DIR}/include
+ ${CMAKE_SOURCE_DIR}/src/ipcpd
+ ${CMAKE_BINARY_DIR}/src/ipcpd
+)
+
+disable_test_logging_for_target(${PARENT_DIR}_lab_test)
+target_link_libraries(${PARENT_DIR}_lab_test PRIVATE ouroboros-common)
+
+if(MB_ECN_LAB_FULL)
+ target_compile_definitions(${PARENT_DIR}_lab_test PRIVATE MB_ECN_LAB_FULL)
+endif()
+
+add_dependencies(build_tests ${PARENT_DIR}_lab_test)
+
+ouroboros_register_tests(TARGET ${PARENT_DIR}_lab_test
+ TESTS ${${PARENT_DIR}_lab_tests})
diff --git a/src/ipcpd/unicast/ca/tests/mb_ecn_lab_test.c b/src/ipcpd/unicast/ca/tests/mb_ecn_lab_test.c
new file mode 100644
index 00000000..dac5e8ac
--- /dev/null
+++ b/src/ipcpd/unicast/ca/tests/mb_ecn_lab_test.c
@@ -0,0 +1,1294 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Shared-bottleneck lab for multi-bit ECN congestion avoidance
+ *
+ * Dimitri Staessens <dimitri@ouroboros.rocks>
+ * Sander Vrijders <sander@ouroboros.rocks>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#include "mb-ecn.c"
+#include <test/test.h>
+
+#define MS (MILLION) /* one millisecond in ns */
+#define LEN 1000 /* default packet size (bytes) */
+
+/* Create a context with the clock zeroed for deterministic time steps. */
+static struct mb_ecn_ctx * mk_ctx(void)
+{
+ struct mb_ecn_ctx * ctx;
+
+ ctx = mb_ecn_ctx_create();
+ if (ctx == NULL)
+ return NULL;
+
+ ctx->rx_ts = 0;
+ ctx->rx_win = 0;
+ ctx->last_ts = 0;
+ ctx->last_ctrl = 0;
+ ctx->last_fb = 0;
+ ctx->last_sig = 0;
+ ctx->last_loc = 0;
+ ctx->last_cap = 0;
+
+ ctx->snd_byt = 0;
+ ctx->snd_win = 0;
+ ctx->snd_r0 = CA_RATE_INIT;
+ ctx->snd_rate = CA_RATE_INIT;
+ ctx->backlogged = true;
+ ctx->src_limited = false;
+ ctx->started = false;
+ ctx->ss_tc = 20 * MS; /* fixed slope for deterministic SS */
+
+ return ctx;
+}
+
+/*
+ * ------------------------------------------------------------------
+ * Lab: packet-level shared-bottleneck simulator.
+ *
+ * Exact-time FIFO link of capacity cap: a packet departs at
+ * max(enqueue, previous departure) + len / cap. Packets are marked
+ * at enqueue from the instantaneous byte queue by mb_ecn_calc_ecn,
+ * the same function the forwarding path calls. Delivered packets
+ * drive a per-flow receiver estimator (mb_ecn_rcv); every window
+ * close is fed back to the sender as ece after a one-way lag,
+ * including the ece 0 release (fa.c). The sender sees its own
+ * previous packet's mark as the local fallback (fa.c l_ecn) and
+ * heartbeat pongs keep liveness. Greedy flows send whenever the
+ * pacer allows; CBR flows follow an absolute schedule. A tick every
+ * LAB_SAMPLE drains the link between sends, so feedback queued by a
+ * departure is due on time even while every flow sits idle. With
+ * cfg.shared every flow runs on one ctx, as a production build does,
+ * and the flow count follows t0, t1 and the churn period.
+ *
+ * The fixpoint tests assert; the sweep always returns success: an
+ * instrument, not a regression test.
+ * ------------------------------------------------------------------
+ */
+
+#define LAB_MAXF 8 /* most flows on one link */
+#define LAB_FIFO 16384 /* bottleneck ring, packets */
+#define LAB_FBQ 64 /* pending feedback ring */
+#define LAB_NONE UINT64_MAX /* no pending event */
+#define LAB_SAMPLE (5 * MS) /* service tick */
+
+struct lab_pkt {
+ uint64_t dep; /* departure time (ns) */
+ uint8_t ecn;
+ uint8_t f; /* flow index */
+};
+
+struct lab_fb {
+ uint64_t t;
+ uint16_t ece;
+ uint8_t fcap;
+};
+
+struct lab_flow {
+ struct mb_ecn_ctx * snd;
+ struct mb_ecn_ctx * rcv;
+ uint64_t t_snd; /* next send attempt (ns) */
+ uint64_t last; /* ctx clock high-water (ns) */
+ uint64_t ftag;
+ uint64_t ia; /* app interval, 0 = greedy */
+ uint64_t app; /* next app slot (ns) */
+ uint64_t lag; /* feedback one-way lag (ns) */
+ uint8_t lecn; /* own previous packet's mark */
+ struct lab_fb fbq[LAB_FBQ];
+ size_t fb_h;
+ size_t fb_n;
+ uint64_t hb_t; /* pong due, LAB_NONE = none */
+ uint64_t hb_rtt;
+ /* metrics, accumulated past warmup */
+ uint64_t m_t; /* last accounting time */
+ uint64_t dlv; /* delivered bytes */
+ uint64_t dlv2; /* delivered in score window */
+ uint64_t r_int; /* integral of rate dt */
+ uint64_t r_lo;
+ uint64_t r_hi;
+ uint64_t lead_B; /* lead-term cut volume */
+ uint64_t prop_B; /* proportional cut volume */
+ uint64_t cuts; /* >45% single-event cuts */
+ uint64_t hold_ns; /* time with ai_hold set */
+ uint64_t lim_ns; /* time src_limited (latch) */
+ uint64_t idl_ns; /* time the backlog test off */
+};
+
+struct lab_link {
+ uint64_t cap; /* bytes/s */
+ uint8_t cc; /* stamped capacity code */
+ uint64_t t_srv; /* line busy until (ns) */
+ uint64_t q; /* queued bytes */
+ uint64_t qmax; /* blocking threshold (bytes) */
+ struct lab_pkt pk[LAB_FIFO];
+ size_t h;
+ size_t n;
+ /* metrics */
+ uint64_t q_t; /* last q-change time */
+ uint64_t q_int; /* integral of q dt */
+ uint64_t mk_int; /* integral of ece(q) dt */
+ uint64_t e_from; /* empty-dwell start */
+ uint64_t e_ns; /* empty time past warmup */
+ size_t e_eps; /* empty episodes past warmup */
+ uint64_t dlv; /* delivered bytes */
+};
+
+struct lab_cfg {
+ const char * name;
+ uint64_t cap; /* bytes/s */
+ uint64_t dur; /* run length (ns) */
+ uint64_t wu; /* warmup excluded (ns) */
+ size_t len; /* packet size (bytes) */
+ uint64_t qmax; /* bytes */
+ size_t n; /* flows, up to LAB_MAXF */
+ uint64_t ia[LAB_MAXF]; /* app interval, 0 = greedy */
+ uint64_t lag[LAB_MAXF]; /* one-way feedback lag (ns) */
+ bool no_loc; /* disable local-mark path */
+ uint64_t st_d; /* service stall length (ns) */
+ uint64_t st_p; /* stall period, 0 = never */
+ unsigned st_f; /* stalled-flow mask, 0 = all */
+ uint64_t t0[LAB_MAXF]; /* flow start offsets (ns) */
+ uint64_t t1[LAB_MAXF]; /* flow stop, 0 = runs to end */
+ uint64_t r0[LAB_MAXF]; /* seed rate, 0 = slow start */
+ uint64_t sc_lo; /* score window (ns), as the */
+ uint64_t sc_hi; /* integration test scores */
+ bool shared; /* one ctx for every flow */
+ uint64_t ch_p; /* churn period, 0 = never */
+ unsigned ch_f; /* churning flow mask */
+};
+
+static struct lab_link lab_lnk;
+static struct lab_flow lab_fl[LAB_MAXF];
+static uint64_t lab_sc_lo;
+static uint64_t lab_sc_hi;
+static size_t lab_len;
+
+/*
+ * Does flow i hold the ctx at t? A churning flow holds it for the
+ * first half of every ch_p and is gone for the second.
+ */
+static bool lab_up(const struct lab_cfg * c,
+ size_t i,
+ uint64_t t)
+{
+ if (t < c->t0[i])
+ return false;
+
+ if (c->t1[i] > 0 && t >= c->t1[i])
+ return false;
+
+ if (c->ch_p == 0 || ((c->ch_f >> i) & 1) == 0)
+ return true;
+
+ return t % c->ch_p < c->ch_p / 2;
+}
+
+/* Flows holding the ctx at t, the count ca_ctx_get refcounts. */
+static size_t lab_live(const struct lab_cfg * c,
+ uint64_t t)
+{
+ size_t n = 0;
+ size_t i;
+
+ for (i = 0; i < c->n; i++)
+ if (lab_up(c, i, t))
+ n++;
+
+ return n > 0 ? n : 1;
+}
+
+/* The bottleneck marks with the production function, nothing else. */
+static uint8_t lab_mark(uint64_t q)
+{
+ uint8_t e = 0;
+
+ if (q == 0)
+ return 0;
+
+ mb_ecn_calc_ecn(q, &e, QOS_CUBE_BE, lab_len);
+
+ return e;
+}
+
+/* Track the queue integral, the mark integral and empty dwells. */
+static void lab_q_acct(struct lab_link * l,
+ uint64_t now,
+ uint64_t wu)
+{
+ uint64_t dt = now - l->q_t;
+
+ if (l->q_t >= wu && dt > 0) {
+ l->q_int += l->q * dt;
+ l->mk_int += (uint64_t) lab_mark(l->q) * 32 * dt;
+ }
+
+ if (l->q == 0) {
+ if (l->e_from == LAB_NONE)
+ l->e_from = l->q_t;
+ } else if (l->e_from != LAB_NONE) {
+ if (now >= wu) {
+ uint64_t f = l->e_from > wu ? l->e_from : wu;
+ l->e_ns += l->q_t > f ? l->q_t - f : 0;
+ l->e_eps++;
+ }
+ l->e_from = LAB_NONE;
+ }
+
+ l->q_t = now;
+}
+
+/* Deliver everything due; receiver estimator feeds the fb ring. */
+static void lab_service(struct lab_link * l,
+ uint64_t now,
+ uint64_t wu)
+{
+ uint16_t ece;
+ uint8_t fcap;
+
+ while (l->n > 0 && l->pk[l->h].dep <= now) {
+ struct lab_pkt * p = &l->pk[l->h];
+ struct lab_flow * f = &lab_fl[p->f];
+
+ lab_q_acct(l, p->dep, wu);
+ l->q -= lab_len;
+
+ if (p->dep >= wu) {
+ l->dlv += lab_len;
+ f->dlv += lab_len;
+ }
+
+ if (p->dep >= lab_sc_lo && p->dep < lab_sc_hi)
+ f->dlv2 += lab_len;
+
+ if (mb_ecn_rcv(f->rcv, lab_len, p->ecn, l->cc, &ece, &fcap,
+ p->dep) &&
+ f->fb_n < LAB_FBQ) {
+ size_t i = (f->fb_h + f->fb_n++) % LAB_FBQ;
+ f->fbq[i].t = p->dep + f->lag;
+ f->fbq[i].ece = ece;
+ f->fbq[i].fcap = fcap;
+ }
+
+ l->h = (l->h + 1) % LAB_FIFO;
+ l->n--;
+ }
+}
+
+/* Integrate rate, regime dwell and extrema between a flow's events. */
+static void lab_f_acct(struct lab_flow * f,
+ uint64_t now,
+ uint64_t wu)
+{
+ struct mb_ecn_ctx * c = f->snd;
+ uint64_t dt;
+ uint16_t m;
+
+ if (now < f->m_t)
+ now = f->m_t;
+
+ dt = now - f->m_t;
+ if (f->m_t >= wu && dt > 0) {
+ f->r_int += c->rate * dt;
+
+ if (c->ai_hold)
+ f->hold_ns += dt;
+
+ if (c->src_limited)
+ f->lim_ns += dt;
+
+ if (!c->backlogged)
+ f->idl_ns += dt;
+
+ m = c->tx_ece > 0 ? c->tx_ece
+ : (uint16_t) (c->tx_loc << CA_SHFT);
+
+ if (m > CA_ECE_MAX)
+ m = CA_ECE_MAX;
+ f->prop_B += c->rate / CA_ECE_REF * m * dt / BILLION;
+
+ if (c->rate < f->r_lo)
+ f->r_lo = c->rate;
+
+ if (c->rate > f->r_hi)
+ f->r_hi = c->rate;
+ }
+
+ f->m_t = now;
+}
+
+/* One send attempt; returns false when blocked on a full buffer. */
+static bool lab_send(struct lab_link * l,
+ struct lab_flow * f,
+ size_t fi,
+ size_t nf,
+ uint64_t wu,
+ bool no_loc)
+{
+ uint64_t t = f->t_snd;
+ uint64_t r0;
+ uint64_t dep;
+ uint8_t ecn;
+ time_t w;
+
+ lab_service(l, t, wu);
+
+ if (l->q + lab_len > l->qmax) { /* blocking write */
+ f->t_snd = l->pk[l->h].dep;
+ return false;
+ }
+
+ lab_f_acct(f, t, wu);
+
+ ecn = lab_mark(l->q);
+
+ r0 = f->snd->rate;
+
+ mb_ecn_flows(f->snd, nf, t);
+
+ if (!no_loc)
+ mb_ecn_loc(f->snd, f->lecn, t);
+
+ w = mb_ecn_snd(f->snd, lab_len, t, &f->ftag);
+
+ if (f->snd->rate * 100 < r0 * 55)
+ f->cuts++;
+
+ f->lecn = ecn;
+ f->last = t;
+
+ lab_q_acct(l, t, wu);
+
+ dep = (t > l->t_srv ? t : l->t_srv) + lab_len * BILLION / l->cap;
+ l->t_srv = dep;
+ l->pk[(l->h + l->n) % LAB_FIFO].dep = dep;
+ l->pk[(l->h + l->n) % LAB_FIFO].ecn = ecn;
+ l->pk[(l->h + l->n) % LAB_FIFO].f = (uint8_t) fi;
+ l->n++;
+ l->q += lab_len;
+
+ if (mb_ecn_ctx_hb_due(f->snd, t) && f->hb_t == LAB_NONE) {
+ f->hb_rtt = l->q * BILLION / l->cap + 2 * f->lag;
+ f->hb_t = t + f->hb_rtt;
+ }
+
+ if (f->ia == 0) {
+ f->t_snd = t + (w > 0 ? (uint64_t) w : 1);
+ } else {
+ f->app += f->ia;
+ f->t_snd = f->app > t + (uint64_t) w ? f->app
+ : t + (uint64_t) w;
+ }
+
+ return true;
+}
+
+/* Apply one queued feedback to the sender, with lead accounting. */
+static void lab_fb_apply(struct lab_flow * f,
+ uint64_t wu)
+{
+ struct lab_fb * fb = &f->fbq[f->fb_h];
+ uint64_t t = fb->t > f->last ? fb->t : f->last;
+ uint64_t r0 = f->snd->rate;
+ uint16_t step;
+ bool up;
+
+ lab_f_acct(f, t, wu);
+
+ if (t >= wu) {
+ up = fb->ece > f->snd->tx_ecp;
+ step = up ? fb->ece - f->snd->tx_ecp
+ : f->snd->tx_ecp - fb->ece;
+
+ if (step > CA_ECE_REF)
+ step = CA_ECE_REF;
+
+ if (up)
+ f->lead_B += r0 * step
+ / (CA_ECE_REF * CA_MD_KD_DIV);
+ }
+
+ mb_ecn_ece(f->snd, fb->ece, fb->fcap, t);
+
+ if (f->snd->rate * 100 < r0 * 55)
+ f->cuts++;
+
+ f->last = t;
+ f->fb_h = (f->fb_h + 1) % LAB_FBQ;
+ f->fb_n--;
+}
+
+static void lab_run(const struct lab_cfg * c)
+{
+ struct lab_link * l = &lab_lnk;
+ uint64_t smp = 0;
+ uint64_t st_t;
+ uint64_t span;
+ double secs;
+ size_t i;
+
+ memset(l, 0, sizeof(*l));
+
+ l->cap = c->cap;
+ l->cc = cap_enc(c->cap);
+ l->qmax = c->qmax;
+ l->e_from = LAB_NONE;
+
+ memset(lab_fl, 0, sizeof(lab_fl));
+
+ for (i = 0; i < c->n; i++) {
+ struct lab_flow * f = &lab_fl[i];
+
+ /* Production interns one ctx per (peer, qos cube). */
+ if (c->shared && i > 0) {
+ f->snd = lab_fl[0].snd;
+ f->rcv = lab_fl[0].rcv;
+ } else {
+ f->snd = mk_ctx();
+ f->rcv = mk_ctx();
+ }
+
+ if (f->snd == NULL || f->rcv == NULL) {
+ printf("lab: no memory.\n");
+ goto fail_ctx;
+ }
+
+ f->ia = c->ia[i];
+ f->lag = c->lag[i];
+ f->hb_t = LAB_NONE;
+ f->r_lo = UINT64_MAX;
+ f->t_snd = c->t0[i];
+ f->app = c->t0[i];
+ f->m_t = c->t0[i];
+ /* Layer-declared RTT seed; pongs then track truth. */
+ f->snd->ss_tc = 2 * CA_SS_RTT_DEF * MILLION;
+
+ /* Seeded: start in AIMD, so the sweep probes the
+ attractor and not the ramp. */
+ if (c->r0[i] == 0)
+ continue;
+
+ f->snd->rate = c->r0[i];
+ f->snd->inv_rate = mb_ecn_rate_inv(c->r0[i]);
+ f->snd->snd_r0 = c->r0[i];
+ f->snd->snd_rate = c->r0[i];
+ f->snd->tx_cav = true;
+ }
+
+ lab_len = c->len;
+ lab_sc_lo = c->sc_lo;
+ lab_sc_hi = c->sc_hi;
+
+ st_t = c->st_p > 0 ? c->st_p : LAB_NONE;
+
+ while (true) {
+ uint64_t nxt = LAB_NONE;
+ int ev = -1; /* flow * 4 + kind */
+
+ /*
+ * Sender-side service stall: the scheduler feeding
+ * the transmit queue pauses for st_d, the queue
+ * drains clean, and the resume bursts the backlog
+ * through the marker (dsched untrack/starve model).
+ * Jitter the period so it cannot phase-lock.
+ */
+ if (st_t != LAB_NONE && smp >= st_t) {
+ uint64_t end = st_t + c->st_d;
+ unsigned msk = c->st_f == 0 ? 3 : c->st_f;
+
+ for (i = 0; i < c->n; i++)
+ if (((msk >> i) & 1) && lab_fl[i].t_snd < end)
+ lab_fl[i].t_snd = end;
+ st_t += c->st_p + (st_t / c->st_p % 3) * c->st_p / 5;
+ }
+
+ for (i = 0; i < c->n; i++) {
+ struct lab_flow * f = &lab_fl[i];
+
+ if (c->t1[i] > 0 && f->t_snd >= c->t1[i]) {
+ f->t_snd = LAB_NONE;
+ } else if (c->ch_p > 0 && !lab_up(c, i, f->t_snd)) {
+ /* Gone: the app resumes next period. */
+ f->t_snd = (f->t_snd / c->ch_p + 1) * c->ch_p;
+ f->app = f->t_snd;
+ }
+
+ if (f->t_snd < nxt) {
+ nxt = f->t_snd;
+ ev = (int) i * 4;
+ }
+ if (f->fb_n > 0 && f->fbq[f->fb_h].t < nxt) {
+ nxt = f->fbq[f->fb_h].t;
+ ev = (int) i * 4 + 1;
+ }
+ if (f->hb_t < nxt) {
+ nxt = f->hb_t;
+ ev = (int) i * 4 + 2;
+ }
+ }
+
+ if (smp < nxt) {
+ nxt = smp;
+ ev = -2;
+ }
+
+ if (nxt >= c->dur)
+ break;
+
+ if (ev == -2) {
+ lab_service(l, smp, c->wu);
+ smp += LAB_SAMPLE;
+ continue;
+ }
+
+ i = (size_t) (ev / 4);
+ switch (ev % 4) {
+ case 0:
+ (void) lab_send(l, &lab_fl[i], i,
+ c->shared ? lab_live(c, nxt) : 1,
+ c->wu, c->no_loc);
+ break;
+ case 1:
+ lab_fb_apply(&lab_fl[i], c->wu);
+ break;
+ default:
+ lab_f_acct(&lab_fl[i], lab_fl[i].hb_t, c->wu);
+ if (lab_fl[i].hb_t > lab_fl[i].last)
+ lab_fl[i].last = lab_fl[i].hb_t;
+ mb_ecn_ctx_rtt(lab_fl[i].snd, lab_fl[i].last,
+ lab_fl[i].hb_rtt);
+ lab_fl[i].hb_t = LAB_NONE;
+ break;
+ }
+ }
+
+ lab_service(l, c->dur, c->wu);
+ lab_q_acct(l, c->dur, c->wu);
+
+ span = c->dur - c->wu;
+ secs = (double) span / BILLION;
+
+ printf("%-14s C %5.2f MB/s n %zu | util %5.1f%% "
+ "q %6.1f pkt mk %5.1f e%% %4.1f eps %3zu\n",
+ c->name, (double) c->cap / MILLION, c->n,
+ 100.0 * (double) l->dlv / ((double) c->cap * secs),
+ (double) l->q_int / ((double) span * c->len),
+ (double) l->mk_int / (double) span,
+ 100.0 * (double) l->e_ns / (double) span,
+ l->e_eps);
+
+ for (i = 0; i < c->n; i++) {
+ struct lab_flow * f = &lab_fl[i];
+ struct mb_ecn_ctx * s = f->snd;
+
+ lab_f_acct(f, c->dur, c->wu);
+
+ if (c->sc_hi > c->sc_lo)
+ printf(" f%zu score %.3f Mb/s in [%llu,%llu)s\n",
+ i, 8.0 * (double) f->dlv2 /
+ ((double) (c->sc_hi - c->sc_lo) / BILLION
+ * MILLION),
+ (unsigned long long) (c->sc_lo / BILLION),
+ (unsigned long long) (c->sc_hi / BILLION));
+ printf(" f%zu %s dlv %5.3f Mb/s rate mean %8.0f "
+ "lo %8" PRIu64 " hi %8" PRIu64 "\n"
+ " lead %8" PRIu64 " prop %8" PRIu64
+ " cuts %4" PRIu64 " hold %4.1f%% lim %4.1f%%"
+ " idl %4.1f%% loss %" PRIu64 "\n",
+ i, f->ia == 0 ? "gdy" : "cbr",
+ 8.0 * (double) f->dlv / ((double) secs * MILLION),
+ (double) f->r_int / (double) span,
+ f->r_lo == UINT64_MAX ? 0 : f->r_lo, f->r_hi,
+ f->lead_B, f->prop_B, f->cuts,
+ 100.0 * (double) f->hold_ns / (double) span,
+ 100.0 * (double) f->lim_ns / (double) span,
+ 100.0 * (double) f->idl_ns / (double) span,
+ s->n_loss);
+
+ if (c->shared && i > 0)
+ continue;
+
+ mb_ecn_ctx_destroy(f->snd);
+ mb_ecn_ctx_destroy(f->rcv);
+ }
+
+ return;
+ fail_ctx:
+ for (i = 0; i < c->n; i++) {
+ if (c->shared && i > 0)
+ break;
+
+ mb_ecn_ctx_destroy(lab_fl[i].snd);
+ mb_ecn_ctx_destroy(lab_fl[i].rcv);
+ }
+}
+
+/*
+ * Faithful test_cbr_protection / test_single_flow_slow_link protocol:
+ * 3-node chain, bottleneck one hop past the sender (no local mark),
+ * ~2 ms feedback path, CBR from t = 0, greedy joining at 300 ms,
+ * scored over the integration test's own window (slow start and
+ * convergence included, as the real assertion sees them).
+ */
+static void lab_cfg_std(struct lab_cfg * c,
+ const char * name,
+ uint64_t cap,
+ size_t n)
+{
+ size_t i;
+
+ memset(c, 0, sizeof(*c));
+
+ c->name = name;
+ c->cap = cap;
+ c->n = n;
+ c->len = LEN;
+ /* 1024 = SSM_RBUFF_SIZE (cmake/config/lib/ssm.cmake). */
+ c->qmax = 1024 * c->len;
+ c->no_loc = true;
+
+ for (i = 0; i < n; i++)
+ c->lag[i] = 2 * MS;
+
+ if (n == 2) { /* cbr_protection */
+ c->ia[1] = c->len * BILLION / 375000;
+ c->t0[0] = 300 * MS;
+ c->dur = 35ULL * BILLION;
+ c->wu = 30ULL * BILLION;
+ c->sc_lo = 5ULL * BILLION;
+ c->sc_hi = 30ULL * BILLION;
+ } else { /* single_flow_slow_link */
+ c->dur = 95ULL * BILLION;
+ c->wu = 55ULL * BILLION;
+ c->sc_lo = 20ULL * BILLION;
+ c->sc_hi = 50ULL * BILLION;
+ }
+}
+
+/* Spread the seeds may end on and still count as one attractor. */
+#define LAB_FIX_TOL 0.10
+
+/*
+ * Two flows on one bottleneck reach the same split whatever they start
+ * from: the difference mode contracts, so the seed cannot survive in
+ * the answer. A second attractor shows as seeds that disagree, a bias
+ * as agreement away from 1.
+ */
+static int test_mb_ecn_lab_fixpoint(uint64_t cap,
+ uint64_t lag)
+{
+ static struct lab_cfg c;
+ static const unsigned num[] = { 1, 1, 4 };
+ static const unsigned den[] = { 4, 1, 1 };
+ uint64_t kb = cap * 8 / 1000;
+ uint64_t ms = lag / MS;
+ double r[3];
+ double lo;
+ double hi;
+ size_t i;
+
+ TEST_START("(%" PRIu64 " kb/s, lag %" PRIu64 " ms)", kb, ms);
+
+ for (i = 0; i < 3; i++) {
+ lab_cfg_std(&c, "fixpoint", cap, 2);
+
+ /* 10 Gb/s carries 9000 B frames; below 1 Gb/s, 1000 B. */
+ c.len = cap >= 125000000 ? 9000 : LEN;
+ c.qmax = 1024 * c.len;
+
+ c.ia[1] = 0; /* both greedy */
+ c.t0[0] = 0;
+ c.lag[0] = lag;
+ c.lag[1] = lag;
+ c.dur = 60ULL * BILLION;
+ c.wu = 30ULL * BILLION;
+ c.sc_lo = 30ULL * BILLION;
+ c.sc_hi = 60ULL * BILLION;
+ c.r0[0] = cap * num[i] / (num[i] + den[i]);
+ c.r0[1] = cap * den[i] / (num[i] + den[i]);
+
+ lab_run(&c);
+
+ if (lab_fl[1].dlv2 == 0) {
+ printf("seed %u:%u starved a flow.\n", num[i], den[i]);
+ goto fail;
+ }
+
+ r[i] = (double) lab_fl[0].dlv2 / (double) lab_fl[1].dlv2;
+ }
+
+ lo = hi = r[0];
+
+ for (i = 1; i < 3; i++) {
+ if (r[i] < lo)
+ lo = r[i];
+
+ if (r[i] > hi)
+ hi = r[i];
+ }
+
+ if (hi > lo * (1.0 + LAB_FIX_TOL)) {
+ printf("seeds disagree: %.3f %.3f %.3f.\n", r[0], r[1], r[2]);
+ goto fail;
+ }
+
+ if (lo < 1.0 - LAB_FIX_TOL || hi > 1.0 + LAB_FIX_TOL) {
+ printf("split %.3f..%.3f is not fair.\n", lo, hi);
+ goto fail;
+ }
+
+ TEST_SUCCESS("(%" PRIu64 " kb/s, lag %" PRIu64 " ms)", kb, ms);
+
+ return TEST_RC_SUCCESS;
+ fail:
+ TEST_FAIL("(%" PRIu64 " kb/s, lag %" PRIu64 " ms)", kb, ms);
+ return TEST_RC_FAIL;
+}
+
+static int test_mb_ecn_lab_fixpoint_all(void)
+{
+#ifdef MB_ECN_LAB_FULL
+ static const uint64_t cap[] = { 1250000000, 12500000,
+ 1250000, 62500 };
+#else
+ static const uint64_t cap[] = { 1250000, 62500 };
+#endif
+ static const uint64_t lag[] = { 2 * MS, 42 * MS };
+ int ret = 0;
+ size_t i;
+ size_t j;
+
+ for (i = 0; i < sizeof(cap) / sizeof(cap[0]); i++)
+ for (j = 0; j < sizeof(lag) / sizeof(lag[0]); j++)
+ ret |= test_mb_ecn_lab_fixpoint(cap[i], lag[j]);
+
+ return ret;
+}
+
+/* Seed weights: even, graded, and graded reversed. */
+static const unsigned lab_n8_w[3][LAB_MAXF] = {
+ { 1, 1, 1, 1, 1, 1, 1, 1 },
+ { 1, 2, 3, 4, 5, 6, 7, 8 },
+ { 8, 7, 6, 5, 4, 3, 2, 1 }
+};
+
+/* Spread across eight flows that still counts as one even split. */
+#define LAB_N8_TOL 0.10
+
+/*
+ * Below this, packet-size quantisation dominates the spread, so the
+ * split is not scored here; starvation (lo == 0) and the utilisation
+ * gate still apply, so the exemption is narrow.
+ */
+#define LAB_N8_FAIR 125000 /* bytes/s, 1 Mb/s */
+
+/* Aggregate the link has to carry for a run to say anything at all. */
+#define LAB_N8_UTIL 2 /* divisor: half of capacity */
+
+/*
+ * Eight flows on one bottleneck. The additive increase is per flow, so
+ * both the aggregate probe and the contraction rate scale with the flow
+ * count, and this is where that scaling shows.
+ */
+static int test_mb_ecn_lab_fixpoint_n8(uint64_t cap,
+ uint64_t lag)
+{
+ static struct lab_cfg c;
+ uint64_t kb = cap * 8 / 1000;
+ uint64_t ms = lag / MS;
+ double worst = 1.0;
+ uint64_t want;
+ uint64_t tot;
+ uint64_t sum;
+ uint64_t lo;
+ uint64_t hi;
+ size_t i;
+ size_t j;
+
+ TEST_START("(%" PRIu64 " kb/s, lag %" PRIu64 " ms)", kb, ms);
+
+ for (i = 0; i < 3; i++) {
+ lab_cfg_std(&c, "fixpoint-n8", cap, LAB_MAXF);
+
+ /* 10 Gb/s carries 9000 B frames; below 1 Gb/s, 1000 B. */
+ c.len = cap >= 125000000 ? 9000 : LEN;
+ c.qmax = 1024 * c.len;
+
+ sum = 0;
+
+ for (j = 0; j < LAB_MAXF; j++)
+ sum += lab_n8_w[i][j];
+
+ for (j = 0; j < LAB_MAXF; j++) {
+ c.ia[j] = 0; /* all greedy */
+ c.t0[j] = 0;
+ c.lag[j] = lag;
+ c.r0[j] = cap * lab_n8_w[i][j] / sum;
+ }
+
+ c.dur = 400ULL * BILLION;
+ c.wu = 200ULL * BILLION;
+ c.sc_lo = 200ULL * BILLION;
+ c.sc_hi = 400ULL * BILLION;
+
+ lab_run(&c);
+
+ tot = 0;
+ lo = lab_fl[0].dlv2;
+ hi = lab_fl[0].dlv2;
+
+ for (j = 0; j < LAB_MAXF; j++) {
+ tot += lab_fl[j].dlv2;
+ if (lab_fl[j].dlv2 < lo)
+ lo = lab_fl[j].dlv2;
+
+ if (lab_fl[j].dlv2 > hi)
+ hi = lab_fl[j].dlv2;
+ }
+
+ if (lo == 0) {
+ printf("seed %zu wedged a flow.\n", i);
+ goto fail;
+ }
+
+ want = cap * ((c.sc_hi - c.sc_lo) / BILLION);
+ if (tot < want / LAB_N8_UTIL) {
+ printf("seed %zu carried %" PRIu64 " of %" PRIu64
+ " bytes.\n", i, tot, want);
+ goto fail;
+ }
+
+ if ((double) hi / (double) lo > worst)
+ worst = (double) hi / (double) lo;
+ }
+
+ if (cap >= LAB_N8_FAIR && worst > 1.0 + LAB_N8_TOL) {
+ printf("widest split %.3f across eight flows.\n", worst);
+ goto fail;
+ }
+
+ TEST_SUCCESS("(%" PRIu64 " kb/s, lag %" PRIu64 " ms)", kb, ms);
+
+ return TEST_RC_SUCCESS;
+ fail:
+ TEST_FAIL("(%" PRIu64 " kb/s, lag %" PRIu64 " ms)", kb, ms);
+ return TEST_RC_FAIL;
+}
+
+static int test_mb_ecn_lab_fixpoint_n8_all(void)
+{
+#ifdef MB_ECN_LAB_FULL
+ static const uint64_t cap[] = { 1250000000, 12500000,
+ 1250000, 62500 };
+#else
+ static const uint64_t cap[] = { 1250000, 62500 };
+#endif
+ static const uint64_t lag[] = { 2 * MS, 42 * MS };
+ int ret = 0;
+ size_t i;
+ size_t j;
+
+ for (i = 0; i < sizeof(cap) / sizeof(cap[0]); i++)
+ for (j = 0; j < sizeof(lag) / sizeof(lag[0]); j++)
+ ret |= test_mb_ecn_lab_fixpoint_n8(cap[i], lag[j]);
+
+ return ret;
+}
+
+/* Jain's fairness index over the score bytes of flows [lo, hi). */
+static double lab_jain(size_t lo,
+ size_t hi)
+{
+ double s = 0.0;
+ double s2 = 0.0;
+ double x;
+ size_t i;
+
+ for (i = lo; i < hi; i++) {
+ x = (double) lab_fl[i].dlv2;
+ s += x;
+ s2 += x * x;
+ }
+
+ if (s2 == 0.0)
+ return 0.0;
+
+ return s * s / ((double) (hi - lo) * s2);
+}
+
+/* Greedy flows on one ctx, all from t = 0, short feedback path. */
+static void lab_cfg_shared(struct lab_cfg * c,
+ const char * name,
+ uint64_t cap,
+ size_t n)
+{
+ size_t i;
+
+ lab_cfg_std(c, name, cap, n);
+
+ c->shared = true;
+
+ for (i = 0; i < n; i++) {
+ c->ia[i] = 0;
+ c->t0[i] = 0;
+ c->lag[i] = 2 * MS;
+ }
+}
+
+/*
+ * ------------------------------------------------------------------
+ * Shared context: n flows, one struct mb_ecn_ctx, one ftag each.
+ *
+ * This is what a production build runs: ca_ctx_get interns one ctx
+ * per (peer, qos cube), so rate, vt and the mark are shared and the
+ * start tag is all a flow owns. The pacer then admits n * rate, so
+ * the attractor for rate is C / n, and the offered load the ctx
+ * measures is n flows' bytes against one flow's rate.
+ * ------------------------------------------------------------------
+ */
+
+/*
+ * Spread the shared attractor may sit in and still count as a
+ * per-flow share. The flows share one virtual clock, so the pacer
+ * fires them in one burst per tick; where that burst is a large part
+ * of the bandwidth-delay product the quarter-log2 mark prices it as
+ * a queue and the loop settles into a deep sawtooth, down to ~0.55
+ * of C / n around 10 Mb/s at 1 kB packets. The band carries that and
+ * is still an order of magnitude under the path rate C.
+ */
+#define LAB_SHR_LO 0.45
+#define LAB_SHR_HI 1.10
+
+/* Peak of the same sawtooth, over the settled window. */
+#define LAB_SHR_PK 1.35
+
+/* One even split across the flows sharing the ctx. */
+#define LAB_SHR_JN 0.98
+
+/*
+ * Greedy flows on one ctx, optionally with half of them joining and
+ * leaving again mid-run. Scores the epoch after the last change.
+ */
+static int test_mb_ecn_lab_shared(uint64_t cap,
+ size_t n,
+ bool churn)
+{
+ static struct lab_cfg c;
+ uint64_t kb = cap * 8 / 1000;
+ uint64_t tc = 20ULL * BILLION;
+ uint64_t fair;
+ uint64_t mean;
+ double jain;
+ size_t nl;
+ size_t i;
+
+ TEST_START("(%" PRIu64 " kb/s, %zu flows%s)", kb, n,
+ churn ? ", churn" : "");
+
+ lab_cfg_shared(&c, churn ? "shr-churn" : "shr-gdy", cap, n);
+
+ nl = churn ? n / 2 : n;
+ /* Half the flows join at tc and are gone again at 2 * tc. */
+ for (i = nl; i < n; i++) {
+ c.t0[i] = tc;
+ c.t1[i] = 2 * tc;
+ }
+
+ fair = cap / nl;
+
+ /*
+ * Warmup ends at the last change, so r_hi is the peak of the
+ * epoch that has to settle back to the new share.
+ */
+ c.dur = 2 * tc + 60ULL * BILLION;
+ c.wu = churn ? 2 * tc : 40ULL * BILLION;
+ c.sc_lo = c.wu + 20ULL * BILLION;
+ c.sc_hi = c.dur;
+
+ lab_run(&c);
+
+ mean = lab_fl[0].r_int / (c.dur - c.wu);
+ jain = lab_jain(0, nl);
+
+ if (mean < (uint64_t) (LAB_SHR_LO * (double) fair)
+ || mean > (uint64_t) (LAB_SHR_HI * (double) fair)) {
+ printf("rate %" PRIu64 " is not a %" PRIu64 " share.\n",
+ mean, fair);
+ goto fail;
+ }
+
+ if (lab_fl[0].r_hi > (uint64_t) (LAB_SHR_PK * (double) fair)) {
+ printf("rate peaked at %.2f of the share.\n",
+ (double) lab_fl[0].r_hi / (double) fair);
+ goto fail;
+ }
+
+ if (jain < LAB_SHR_JN) {
+ printf("fairness %.4f across %zu flows.\n", jain, nl);
+ goto fail;
+ }
+
+ TEST_SUCCESS("(%" PRIu64 " kb/s, %zu flows%s)", kb, n,
+ churn ? ", churn" : "");
+
+ return TEST_RC_SUCCESS;
+ fail:
+ TEST_FAIL("(%" PRIu64 " kb/s, %zu flows%s)", kb, n,
+ churn ? ", churn" : "");
+ return TEST_RC_FAIL;
+}
+
+/*
+ * Offered load per flow, as a divisor of the fair share. Low enough
+ * that the aggregate never fills the link, so nothing marks and the
+ * offered-load path is the only thing bounding the rate.
+ */
+#define LAB_SRC_DIV 4
+
+/*
+ * mb_ecn_ceiling admits twice the load ONE flow offers. The slack
+ * covers the additive increase banked between two window closes and
+ * the truncation in sharing the offered bytes out. Reading the whole
+ * ctx's load as one flow's puts the bound n times higher, so a wide
+ * slack still separates the two.
+ */
+#define LAB_SRC_CEIL 2.5
+
+/* Churn half-period; below CA_SND_WIN no window would ever close. */
+#define LAB_SRC_CHP (100 * MS)
+
+/*
+ * Source-limited flows on one ctx. Each offers a fixed rate well
+ * under its share, so mb_ecn_ceiling and the backlog level are all
+ * that bound the rate, and both read the offered load. With churn,
+ * the flows above n / 4 come and go every LAB_SRC_CHP, so a window
+ * that does not restart on the count change never measures one
+ * population.
+ */
+static int test_mb_ecn_lab_shared_load(uint64_t cap,
+ size_t n,
+ bool churn)
+{
+ static struct lab_cfg c;
+ uint64_t kb = cap * 8 / 1000;
+ uint64_t off = cap / (LAB_SRC_DIV * n);
+ uint64_t hi;
+ double jain;
+ size_t ns;
+ size_t i;
+
+ TEST_START("(%" PRIu64 " kb/s, %zu flows%s)", kb, n,
+ churn ? ", churn" : "");
+
+ lab_cfg_shared(&c, churn ? "src-churn" : "src-cbr", cap, n);
+
+ for (i = 0; i < n; i++)
+ c.ia[i] = c.len * BILLION / off;
+
+ if (churn) {
+ c.ch_p = 2 * LAB_SRC_CHP;
+ c.ch_f = ~0u << (n / 4);
+ }
+
+ c.dur = 120ULL * BILLION;
+ c.wu = 60ULL * BILLION;
+ c.sc_lo = 60ULL * BILLION;
+ c.sc_hi = 120ULL * BILLION;
+
+ lab_run(&c);
+
+ hi = lab_fl[0].r_hi;
+
+ /* Score the flows that shared the ctx over the same epochs. */
+ ns = churn ? n / 4 : 0;
+ jain = lab_jain(ns, n);
+
+ if (hi > (uint64_t) (LAB_SRC_CEIL * (double) off)) {
+ printf("rate peaked at %.2f of the %" PRIu64 " offered, "
+ "%.2f of the %" PRIu64 " path.\n",
+ (double) hi / (double) off, off,
+ (double) hi / (double) cap, cap);
+ goto fail;
+ }
+
+ if (jain < LAB_SHR_JN) {
+ printf("fairness %.4f across %zu flows.\n", jain, n - ns);
+ goto fail;
+ }
+
+ TEST_SUCCESS("(%" PRIu64 " kb/s, %zu flows%s)", kb, n,
+ churn ? ", churn" : "");
+
+ return TEST_RC_SUCCESS;
+ fail:
+ TEST_FAIL("(%" PRIu64 " kb/s, %zu flows%s)", kb, n,
+ churn ? ", churn" : "");
+ return TEST_RC_FAIL;
+}
+
+static int test_mb_ecn_lab(void)
+{
+ static const uint64_t gc_cap[] = { 625000, 1250000, 12500000 };
+ static const char * gc_nm[] = { "gc-5M", "gc-10M", "gc-100M" };
+ static const uint64_t sf_cap[] = { 62500, 125000, 1250000 };
+ static const char * sf_nm[] = { "sf-500k", "sf-1M", "sf-10M" };
+ static const uint64_t g2_cap[] = {
+ 1250000, 1250000, 1250000, 62500, 62500, 62500
+ };
+ static const uint64_t g2_lag[] = {
+ 2 * MS, 20 * MS, 42 * MS, 2 * MS, 20 * MS, 42 * MS
+ };
+ static const char * g2_nm[] = {
+ "g2-10M-2", "g2-10M-20", "g2-10M-42",
+ "g2-500k-2", "g2-500k-20", "g2-500k-42"
+ };
+ static const uint64_t ul_lag[] = { 20 * MS, 42 * MS };
+ static const char * ul_nm[] = { "g2-ul20", "g2-ul42" };
+ static struct lab_cfg c;
+ size_t i;
+
+ TEST_START();
+
+ /*
+ * cbr_protection over capacity: a 3 Mb/s CBR flow shares the
+ * link with a greedy flow joining at 300 ms, so the share the
+ * CBR has to hold runs 60%, 30% and 3% of the link.
+ */
+ for (i = 0; i < 3; i++) {
+ lab_cfg_std(&c, gc_nm[i], gc_cap[i], 2);
+ lab_run(&c);
+ }
+
+ /*
+ * single_flow_slow_link over capacity: one greedy flow alone.
+ * The marking quantum is fixed in bytes, so capacity alone
+ * decides how much queueing delay one ecn step prices.
+ */
+ for (i = 0; i < 3; i++) {
+ lab_cfg_std(&c, sf_nm[i], sf_cap[i], 1);
+ lab_run(&c);
+ }
+
+ /*
+ * Two greedy flows over capacity and equal feedback lag: the
+ * split they settle on and how a long loop degrades it.
+ */
+ for (i = 0; i < 6; i++) {
+ lab_cfg_std(&c, g2_nm[i], g2_cap[i], 2);
+
+ c.ia[1] = 0;
+ c.lag[0] = g2_lag[i];
+ c.lag[1] = g2_lag[i];
+ c.dur = 65ULL * BILLION;
+ c.wu = 35ULL * BILLION;
+
+ lab_run(&c);
+ }
+
+ /* Unequal lag: flow 0 keeps 2 ms, flow 1 reacts slower. */
+ for (i = 0; i < 2; i++) {
+ lab_cfg_std(&c, ul_nm[i], 1250000, 2);
+
+ c.ia[1] = 0;
+ c.lag[1] = ul_lag[i];
+ c.dur = 65ULL * BILLION;
+ c.wu = 35ULL * BILLION;
+
+ lab_run(&c);
+ }
+
+ /*
+ * Service stalls: the scheduler feeding the transmit queue
+ * pauses, the queue drains clean and the resume bursts the
+ * backlog through the marker.
+ */
+ lab_cfg_std(&c, "st-gc", 1250000, 2);
+
+ c.st_d = 60 * MS;
+ c.st_p = 400 * MS;
+
+ lab_run(&c);
+
+ lab_cfg_std(&c, "st-sf", 125000, 1);
+
+ c.st_d = 200 * MS;
+ c.st_p = BILLION;
+
+ lab_run(&c);
+
+ /* Per-flow starvation: only the sparse CBR flow stalls. */
+ lab_cfg_std(&c, "st-pf", 1250000, 2);
+
+ c.st_d = 100 * MS;
+ c.st_p = 300 * MS;
+ c.st_f = 2;
+
+ lab_run(&c);
+
+ /*
+ * Greedy joins 10 s in, once the CBR flow has settled: a step
+ * into contention rather than a shared ramp.
+ */
+ lab_cfg_std(&c, "gc-late", 1250000, 2);
+
+ c.t0[0] = 10ULL * BILLION;
+ c.dur = 45ULL * BILLION;
+ c.wu = 40ULL * BILLION;
+ c.sc_lo = 15ULL * BILLION;
+ c.sc_hi = 40ULL * BILLION;
+
+ lab_run(&c);
+
+ /*
+ * Second greedy flow joins 5 s in: the incumbent has to give
+ * back half to a newcomer that is still in slow start.
+ */
+ lab_cfg_std(&c, "g2-stag", 1250000, 2);
+
+ c.ia[1] = 0;
+ c.t0[0] = 0;
+ c.t0[1] = 5ULL * BILLION;
+ c.dur = 65ULL * BILLION;
+ c.wu = 35ULL * BILLION;
+
+ lab_run(&c);
+
+ TEST_SUCCESS();
+
+ return TEST_RC_SUCCESS;
+}
+
+int mb_ecn_lab_test(int argc,
+ char ** argv)
+{
+ int ret = 0;
+
+ (void) argc;
+ (void) argv;
+
+ ret |= test_mb_ecn_lab_shared(1250000, 5, false);
+ ret |= test_mb_ecn_lab_shared(1250000, 8, false);
+ ret |= test_mb_ecn_lab_shared(1250000, 8, true);
+ ret |= test_mb_ecn_lab_shared_load(1250000, 5, false);
+ ret |= test_mb_ecn_lab_shared_load(1250000, 8, false);
+ ret |= test_mb_ecn_lab_shared_load(1250000, 5, true);
+ ret |= test_mb_ecn_lab_shared_load(1250000, 8, true);
+ ret |= test_mb_ecn_lab_fixpoint_all();
+ ret |= test_mb_ecn_lab_fixpoint_n8_all();
+ ret |= test_mb_ecn_lab();
+
+ return ret;
+}
diff --git a/src/ipcpd/unicast/ca/tests/mb_ecn_test.c b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c
index 4bbc12aa..7186d3af 100644
--- a/src/ipcpd/unicast/ca/tests/mb_ecn_test.c
+++ b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c
@@ -21,6 +21,7 @@
*/
#include "mb-ecn.c"
+
#include <test/test.h>
#define MS (MILLION) /* one millisecond in ns */
@@ -121,109 +122,140 @@ static int test_mb_ecn_ctx_create_destroy(void)
return TEST_RC_FAIL;
}
-static int test_mb_ecn_calc_ecn(void)
+/* The pricing window derives from the declared RTT. */
+static int test_mb_ecn_init_window(void)
{
- uint8_t ecn;
-
TEST_START();
- /* A queue below one ECN quantum marks nothing. */
- ecn = 0;
- mb_ecn_calc_ecn(CA_MARK_Q - 1, &ecn, QOS_CUBE_BE, 0);
- if (ecn != 0) {
- printf("Sub-quantum queue marked %u.\n", ecn);
- goto fail;
- }
+ /* A fabric RTT lands on the floor, not below it. */
+ mb_ecn_init(1);
- /* Queue depth maps to ecn = queued / CA_MARK_Q. */
- ecn = 0;
- mb_ecn_calc_ecn(5 * CA_MARK_Q, &ecn, QOS_CUBE_BE, 0);
- if (ecn != 5) {
- printf("Expected ecn 5, got %u.\n", ecn);
+ if (mb_ecn_tw != CA_TW_MIN) {
+ printf("fabric window %" PRIu64 ".\n", mb_ecn_tw);
goto fail;
}
- /* MAX keeps the larger value; a smaller mark cannot raise it. */
- ecn = 0x80;
- mb_ecn_calc_ecn(CA_MARK_Q, &ecn, QOS_CUBE_BE, 0);
- if (ecn != 0x80) {
- printf("Expected ecn 0x80, got 0x%x.\n", ecn);
- goto fail;
- }
+ /* A WAN RTT caps the window. */
+ mb_ecn_init(200);
- ecn = 3;
- mb_ecn_calc_ecn(4 * CA_MARK_Q, &ecn, QOS_CUBE_BE, 0);
- if (ecn != 4) {
- printf("Expected ecn 4, got %u.\n", ecn);
+ if (mb_ecn_tw != CA_TW) {
+ printf("wan window %" PRIu64 ".\n", mb_ecn_tw);
goto fail;
}
- /* A queue past 255 quanta saturates, it does not wrap to a low mark. */
- ecn = 0;
- mb_ecn_calc_ecn(256 * CA_MARK_Q, &ecn, QOS_CUBE_BE, 0);
- if (ecn != 255) {
- printf("Deep queue wrapped: exp 255, got %u.\n", ecn);
+ /* An unspecified RTT takes the default and caps the window. */
+ mb_ecn_init(0);
+
+ if (mb_ecn_tw != CA_TW) {
+ printf("default window %" PRIu64 ".\n", mb_ecn_tw);
goto fail;
}
+ mb_ecn_init(CA_SS_RTT_DEF);
+
TEST_SUCCESS();
return TEST_RC_SUCCESS;
fail:
+ mb_ecn_init(CA_SS_RTT_DEF);
TEST_FAIL();
return TEST_RC_FAIL;
}
-/* The first mark after idle emits the raw value with zero latency. */
-static int test_mb_ecn_rcv_onset_immediate(void)
+/* Queue depth (packets) that reads as full congestion. */
+#define FULL_PKTS (CA_MARK_KNEE << ((CA_ECE_REF >> CA_SHFT) / 4))
+
+static int test_mb_ecn_calc_ecn(void)
{
- struct mb_ecn_ctx * ctx;
- uint16_t ece;
- uint8_t fcap;
+ uint8_t ecn;
TEST_START();
- ctx = mk_ctx();
- if (ctx == NULL) {
- printf("Failed to create context.\n");
+ /* One packet in the queue is the floor: it marks nothing. */
+ ecn = 0;
+
+ mb_ecn_calc_ecn(1400, &ecn, QOS_CUBE_BE, 1400);
+
+ if (ecn != 0) {
+ printf("Single packet marked %u.\n", ecn);
goto fail;
}
- if (!mb_ecn_rcv(ctx, LEN, 4, 0, &ece, &fcap, MS)) {
- printf("Onset did not update.\n");
- goto fail_ctx;
+ /* An unknown mean packet size cannot mark. */
+ ecn = 0;
+
+ mb_ecn_calc_ecn(1400, &ecn, QOS_CUBE_BE, 0);
+
+ if (ecn != 0) {
+ printf("Unknown mean marked %u.\n", ecn);
+ goto fail;
}
- if (ece != 4 << CA_SHFT) {
- printf("Onset ece: exp %u, got %u.\n", 4 << CA_SHFT, ece);
- goto fail_ctx;
+ /* Each doubling of the queue adds 4. */
+ ecn = 0;
+
+ mb_ecn_calc_ecn(2 * 1400, &ecn, QOS_CUBE_BE, 1400);
+
+ if (ecn != 4) {
+ printf("Expected ecn 4 at 2 packets, got %u.\n", ecn);
+ goto fail;
}
- if (mb_ecn_rcv(ctx, LEN, 4, 0, &ece, &fcap, 2 * MS)) {
- printf("Mid-window packet updated.\n");
- goto fail_ctx;
+ /* FULL_PKTS packets is full congestion. */
+ ecn = 0;
+
+ mb_ecn_calc_ecn(FULL_PKTS * 1400, &ecn, QOS_CUBE_BE, 1400);
+
+ if (ecn != (CA_ECE_REF >> CA_SHFT)) {
+ printf("Expected ecn %u at full, got %u.\n",
+ CA_ECE_REF >> CA_SHFT, ecn);
+ goto fail;
}
- mb_ecn_ctx_destroy(ctx);
+ /* The same packet count marks the same at any packet size. */
+ ecn = 0;
+
+ mb_ecn_calc_ecn(FULL_PKTS * 200, &ecn, QOS_CUBE_BE, 200);
+
+ if (ecn != (CA_ECE_REF >> CA_SHFT)) {
+ printf("Size dependence: exp %u, got %u.\n",
+ CA_ECE_REF >> CA_SHFT, ecn);
+ goto fail;
+ }
+
+ /* MAX keeps the larger value; a smaller mark cannot lower it. */
+ ecn = 0x80;
+
+ mb_ecn_calc_ecn(2 * 1400, &ecn, QOS_CUBE_BE, 1400);
+
+ if (ecn != 0x80) {
+ printf("Expected ecn 0x80, got 0x%x.\n", ecn);
+ goto fail;
+ }
+
+ ecn = 3;
+
+ mb_ecn_calc_ecn(4 * 1400, &ecn, QOS_CUBE_BE, 1400);
+
+ if (ecn != 8) {
+ printf("Expected ecn 8, got %u.\n", ecn);
+ goto fail;
+ }
TEST_SUCCESS();
return TEST_RC_SUCCESS;
- fail_ctx:
- mb_ecn_ctx_destroy(ctx);
fail:
TEST_FAIL();
return TEST_RC_FAIL;
}
-/* A 50% duty mark square wave emits the time mean, not the last peak. */
-static int test_mb_ecn_rcv_window_mean(void)
+/* The first mark after idle emits the raw value with zero latency. */
+static int test_mb_ecn_rcv_onset_immediate(void)
{
struct mb_ecn_ctx * ctx;
uint16_t ece;
uint8_t fcap;
- size_t upd;
- size_t i;
TEST_START();
@@ -233,24 +265,18 @@ static int test_mb_ecn_rcv_window_mean(void)
goto fail;
}
- mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, MS);
-
- /* The byte trigger closes every ~32 packets: two windows. */
- upd = 0;
- for (i = 1; i <= 68; i++) {
- time_t ecn = (i & 1) ? 8 : 0;
- time_t t = MS + i * MS;
- if (mb_ecn_rcv(ctx, LEN, ecn, 0, &ece, &fcap, t))
- upd++;
+ if (!mb_ecn_rcv(ctx, LEN, 4, 0, &ece, &fcap, MS)) {
+ printf("Onset did not update.\n");
+ goto fail_ctx;
}
- if (upd != 2) {
- printf("%zu updates in two windows.\n", upd);
+ if (ece != 4 << CA_SHFT) {
+ printf("Onset ece: exp %u, got %u.\n", 4 << CA_SHFT, ece);
goto fail_ctx;
}
- if (ece < 120 || ece > 136) {
- printf("window mean: exp ~128, got %u.\n", ece);
+ if (mb_ecn_rcv(ctx, LEN, 4, 0, &ece, &fcap, 2 * MS)) {
+ printf("Mid-window packet updated.\n");
goto fail_ctx;
}
@@ -466,7 +492,7 @@ static int test_mb_ecn_rcv_gap_restart(void)
mb_ecn_rcv(ctx, LEN, 6, 0, &ece, &fcap, MS);
mb_ecn_rcv(ctx, LEN, 6, 0, &ece, &fcap, 2 * MS);
- t = 2 * MS + 10 * CA_TW_INIT;
+ t = 2 * MS + 10 * CA_TW;
if (!mb_ecn_rcv(ctx, LEN, 5, 0, &ece, &fcap, t)) {
printf("gap restart did not update.\n");
goto fail_ctx;
@@ -477,7 +503,7 @@ static int test_mb_ecn_rcv_gap_restart(void)
goto fail_ctx;
}
- t += 10 * CA_TW_INIT;
+ t += 10 * CA_TW;
if (!mb_ecn_rcv(ctx, LEN, 0, 0, &ece, &fcap, t) || ece != 0) {
printf("gap with clean packet did not end: %u.\n", ece);
goto fail_ctx;
@@ -500,8 +526,14 @@ static int test_mb_ecn_rcv_gap_restart(void)
return TEST_RC_FAIL;
}
-/* Max marks at max gaps: exact ceiling, no overflow past the edge. */
-static int test_mb_ecn_rcv_accum_bounds(void)
+/*
+ * At a floored layer RTT, rx_tw sits at CA_TW_MIN, so 4 * rx_tw is
+ * well under CA_ECE_TTL. A gap in that band must still close the
+ * window as a diluted average, not restart fresh: a fresh restart
+ * always emits the raw undiluted mark (ecn << CA_SHFT), so an ece
+ * that low pins the CA_ECE_TTL floor in mb_ecn_rcv_fresh.
+ */
+static int test_mb_ecn_rcv_gap_floor(void)
{
struct mb_ecn_ctx * ctx;
uint16_t ece;
@@ -510,33 +542,39 @@ static int test_mb_ecn_rcv_accum_bounds(void)
TEST_START();
+ mb_ecn_init(2);
+
ctx = mk_ctx();
if (ctx == NULL) {
printf("Failed to create context.\n");
goto fail;
}
- mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, MS);
-
- /* Two packets at dt just under CA_TW_INIT straddle the boundary. */
- t = MS + CA_TW_INIT - 1;
- if (mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, t)) {
- printf("update before the window closed.\n");
+ if (ctx->rx_tw != CA_TW_MIN) {
+ printf("window not floored: %" PRIu64 ".\n", ctx->rx_tw);
goto fail_ctx;
}
- t += CA_TW_INIT - 1;
- if (!mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, t)) {
- printf("no update at the window boundary.\n");
+ /* Onset, then a second packet inside the window: mark banked. */
+ mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, MS);
+ mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, 2 * MS);
+
+ /* 20 ms gap: past 4 * rx_tw (16 ms), well under CA_ECE_TTL. */
+
+ t = 2 * MS + 20 * MS;
+ if (!mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t)) {
+ printf("window did not close.\n");
goto fail_ctx;
}
- if (ece != 15 << CA_SHFT) {
- printf("ceiling: exp %u, got %u.\n", 15 << CA_SHFT, ece);
+ /* A fresh restart would emit the raw mark 8 << CA_SHFT, undiluted. */
+ if (ece >= (8 << CA_SHFT)) {
+ printf("gap read as a fresh onset: ece %u.\n", ece);
goto fail_ctx;
}
mb_ecn_ctx_destroy(ctx);
+ mb_ecn_init(CA_SS_RTT_DEF);
TEST_SUCCESS();
@@ -544,78 +582,49 @@ static int test_mb_ecn_rcv_accum_bounds(void)
fail_ctx:
mb_ecn_ctx_destroy(ctx);
fail:
+ mb_ecn_init(CA_SS_RTT_DEF);
TEST_FAIL();
return TEST_RC_FAIL;
}
-/* Scale-free density: the window holds ~CA_N_TARGET packets at any rate. */
-static int test_mb_ecn_rcv_window_holds_target(void)
+/* Max marks at max gaps: exact ceiling, no overflow past the edge. */
+static int test_mb_ecn_rcv_accum_bounds(void)
{
struct mb_ecn_ctx * ctx;
uint16_t ece;
uint8_t fcap;
- uint16_t last_ece;
- uint64_t rates[4];
- uint64_t ia;
uint64_t t;
- size_t closes;
- size_t since;
- size_t count;
- size_t ri;
TEST_START();
- rates[0] = 5000000;
- rates[1] = 10000000;
- rates[2] = 50000000;
- rates[3] = 100000000;
-
- ctx = NULL;
- for (ri = 0; ri < 4; ri++) {
- ia = 8000ULL * BILLION / rates[ri];
- t = 0;
- closes = 0;
- since = 0;
- count = 0;
- last_ece = 0;
-
- ctx = mk_ctx();
- if (ctx == NULL) {
- printf("Failed to create context.\n");
- goto fail;
- }
+ ctx = mk_ctx();
+ if (ctx == NULL) {
+ printf("Failed to create context.\n");
+ goto fail;
+ }
- /* Warm past the ramp from CA_TW_INIT, then time one gap. */
- while (closes < 42) {
- t += ia;
- since++;
- if (!mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t))
- continue;
- closes++;
- if (closes == 41) {
- since = 0;
- } else if (closes == 42) {
- count = since - 1;
- last_ece = ece;
- }
- }
+ mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, MS);
- if (count < 8 || count > 32) {
- printf("rate %" PRIu64 ": %zu pkts/window.\n",
- rates[ri], count);
- goto fail_ctx;
- }
+ /* Two packets at dt just under CA_TW straddle the boundary. */
+ t = MS + CA_TW - 1;
+ if (mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, t)) {
+ printf("update before the window closed.\n");
+ goto fail_ctx;
+ }
- if (last_ece < 224 || last_ece > 288) {
- printf("rate %" PRIu64 ": ece %u ~256.\n",
- rates[ri], last_ece);
- goto fail_ctx;
- }
+ t += CA_TW - 1;
+ if (!mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, t)) {
+ printf("no update at the window boundary.\n");
+ goto fail_ctx;
+ }
- mb_ecn_ctx_destroy(ctx);
- ctx = NULL;
+ if (ece != 15 << CA_SHFT) {
+ printf("ceiling: exp %u, got %u.\n", 15 << CA_SHFT, ece);
+ goto fail_ctx;
}
+ mb_ecn_ctx_destroy(ctx);
+
TEST_SUCCESS();
return TEST_RC_SUCCESS;
@@ -627,7 +636,7 @@ static int test_mb_ecn_rcv_window_holds_target(void)
}
/*
- * The window floors at CA_TW_MIN, tracks the rate below the old knee,
+ * The window floors at CA_TW, tracks the rate below the knee,
* and only a pathological fold hits the CA_TW_ABSMAX ceiling.
*/
static int test_mb_ecn_rcv_window_clip_bounds(void)
@@ -641,7 +650,7 @@ static int test_mb_ecn_rcv_window_clip_bounds(void)
TEST_START();
- /* 1 GbE is above the high knee: the window floors at CA_TW_MIN. */
+ /* 1 GbE is above the high knee: the window floors at CA_TW. */
ctx = mk_ctx();
if (ctx == NULL) {
printf("Failed to create context.\n");
@@ -657,15 +666,15 @@ static int test_mb_ecn_rcv_window_clip_bounds(void)
closes++;
}
- if (ctx->rx_tw != CA_TW_MIN) {
+ if (ctx->rx_tw != CA_TW) {
printf("high-rate window: exp %" PRIu64 ", got %" PRIu64
- ".\n", (uint64_t) CA_TW_MIN, ctx->rx_tw);
+ ".\n", (uint64_t) CA_TW, ctx->rx_tw);
goto fail_ctx;
}
mb_ecn_ctx_destroy(ctx);
- /* 1 Mbps: past the old knee, ~16 pkts = 16 * 8 ms = 131 ms. */
+ /* 1 Mbps: below the knee, ~16 pkts = 16 * 8 ms = 131 ms. */
ctx = mk_ctx();
if (ctx == NULL) {
printf("Failed to create context.\n");
@@ -697,7 +706,7 @@ static int test_mb_ecn_rcv_window_clip_bounds(void)
}
mb_ecn_rcv(ctx, 10, 8, 0, &ece, &fcap, MS);
- mb_ecn_rcv(ctx, 10, 8, 0, &ece, &fcap, MS + CA_TW_INIT);
+ mb_ecn_rcv(ctx, 10, 8, 0, &ece, &fcap, MS + CA_TW);
if (ctx->rx_tw != CA_TW_ABSMAX) {
printf("window ceiling breached: %" PRIu64 ".\n",
@@ -782,9 +791,9 @@ static int test_mb_ecn_rcv_no_overflow_highrate(void)
/* Open a window, then inject a maximal byte count and span. */
mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, 0);
ctx->rx_byt = CA_RATE_MAX / 8;
- ctx->rx_ts = 2 * CA_TW_INIT - 2;
+ ctx->rx_ts = 2 * CA_TW - 2;
- ok = mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, 2 * CA_TW_INIT - 1);
+ ok = mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, 2 * CA_TW - 1);
if (!ok) {
printf("max-window close did not fire.\n");
@@ -797,7 +806,7 @@ static int test_mb_ecn_rcv_no_overflow_highrate(void)
}
/* A wrapped numerator drives rx_tw to MAX; it must descend. */
- if (ctx->rx_tw >= CA_TW_INIT || ctx->rx_tw < CA_TW_MIN) {
+ if (ctx->rx_tw > CA_TW || ctx->rx_tw < CA_TW) {
printf("window %" PRIu64 " did not descend.\n", ctx->rx_tw);
goto fail_ctx;
}
@@ -816,7 +825,7 @@ static int test_mb_ecn_rcv_no_overflow_highrate(void)
/*
* The sender holds a mark across the full inter-feedback gap (TTL >
- * 2 * CA_TW_INIT); a repeated mark adds only the proportional cut.
+ * 2 * CA_TW); a repeated mark adds only the proportional cut.
*/
static int test_mb_ecn_ece_ttl_covers_cadence(void)
{
@@ -840,10 +849,10 @@ static int test_mb_ecn_ece_ttl_covers_cadence(void)
mb_ecn_ece(ctx, 100, 0, MS);
mb_ecn_snd(ctx, LEN, MS, &ftag);
- /* Sends between feedbacks spaced 2 * CA_TW_INIT + 5 ms apart. */
+ /* Sends between feedbacks spaced 2 * CA_TW + 5 ms apart. */
t = MS;
for (i = 0; i < 4; i++) {
- t += (2 * CA_TW_INIT + 5 * MS) / 4;
+ t += (2 * CA_TW + 5 * MS) / 4;
mb_ecn_snd(ctx, LEN, t, &ftag);
if (ctx->tx_ece == 0) {
printf("mark cleared inside the feedback gap.\n");
@@ -949,10 +958,8 @@ static int test_mb_ecn_dt_scaling_invariant(void)
mb_ecn_ece(b, 0, 0, 0);
a->rate = (uint64_t) 10 << 20;
b->rate = (uint64_t) 10 << 20;
- a->r_bkt = a->rate;
- b->r_bkt = b->rate;
- /* a: one 30 ms step (under one washout bucket, so it stays out). */
+ /* a: one 30 ms step. */
mb_ecn_snd(a, LEN, 30 * MS, &fta);
/* b: thirty 1 ms steps over the same 30 ms. */
@@ -1036,6 +1043,66 @@ static int test_mb_ecn_multiplicative_decrease(void)
return TEST_RC_FAIL;
}
+/* The hold releases on any unsaturated mark, including above REF. */
+static int test_mb_ecn_ai_hold_release(void)
+{
+ struct mb_ecn_ctx * ctx;
+ uint64_t r0 = (uint64_t) 100 << 20;
+
+ TEST_START();
+
+ ctx = mk_ctx();
+ if (ctx == NULL) {
+ printf("Failed to create context.\n");
+ goto fail;
+ }
+
+ /* A saturated mark keeps the hold: the queue has not drained. */
+ ctx->ai_hold = true;
+
+ mb_ecn_ece(ctx, CA_ECE_MAX, 0, MS);
+
+ if (!ctx->ai_hold) {
+ printf("saturated feedback released the hold.\n");
+ goto fail_ctx;
+ }
+
+ /* A standing mark of 20 flows is unsaturated: release. */
+ ctx->ai_hold = true;
+
+ mb_ecn_ece(ctx, 22 << CA_SHFT, 0, 2 * MS);
+
+ if (ctx->ai_hold) {
+ printf("unsaturated feedback held the increase.\n");
+ goto fail_ctx;
+ }
+
+ /* The decrease still scales with the mark saturated at CA_ECE_MAX. */
+ ctx->rate = r0;
+ ctx->tx_cav = true;
+ ctx->tx_ece = CA_ECE_MAX;
+ ctx->tx_ecp = CA_ECE_MAX;
+ ctx->dec_acc = 0;
+
+ mb_ecn_decrease(ctx, MILLION);
+
+ if (ctx->rate > r0 - r0 / 700) {
+ printf("clamped decrease too weak: %" PRIu64 ".\n", ctx->rate);
+ goto fail_ctx;
+ }
+
+ mb_ecn_ctx_destroy(ctx);
+
+ TEST_SUCCESS();
+
+ return TEST_RC_SUCCESS;
+ fail_ctx:
+ mb_ecn_ctx_destroy(ctx);
+ fail:
+ TEST_FAIL();
+ return TEST_RC_FAIL;
+}
+
static int test_mb_ecn_rate_floor(void)
{
struct mb_ecn_ctx * ctx;
@@ -1119,17 +1186,15 @@ static int test_mb_ecn_fixed_point(void)
return TEST_RC_FAIL;
}
-/*
- * A rising ECE mark takes a one-sided lead cut on top of the
- * proportional term; a flat or falling mark takes only the small
- * proportional cut.
- */
-static int test_mb_ecn_lead_cut(void)
+/* The lead is two-sided: it cuts on a rise and gives back on a fall. */
+static int test_mb_ecn_lead_symmetric(void)
{
struct mb_ecn_ctx * ctx;
- uint64_t prev;
+ uint64_t r0 = (uint64_t) 100 << 20;
+ uint64_t net;
uint64_t drop;
- uint64_t ftag = 0;
+ uint64_t gain;
+ uint64_t kd2;
TEST_START();
@@ -1139,36 +1204,81 @@ static int test_mb_ecn_lead_cut(void)
goto fail;
}
- ctx->rate = (uint64_t) 100 << 20;
ctx->tx_cav = true;
- /* Rise 0 -> 256: the lead term cuts hard (~rate/8). */
- prev = ctx->rate;
- mb_ecn_ece(ctx, 256, 0, MS);
- mb_ecn_snd(ctx, LEN, MS, &ftag);
- drop = prev - ctx->rate;
- if (drop < prev / 16) {
- printf("rising mark under-cut: %" PRIu64 ".\n", drop);
+ /* Rise of one reference: cut rate / CA_MD_KD_DIV, no more. */
+ ctx->rate = r0;
+ ctx->tx_ece = CA_ECE_REF;
+ ctx->tx_ecp = 0;
+ ctx->dec_acc = 0;
+
+ mb_ecn_decrease(ctx, 0);
+
+ drop = r0 - ctx->rate;
+ if (drop != r0 / CA_MD_KD_DIV) {
+ printf("rise cut %" PRIu64 ", want %" PRIu64 ".\n",
+ drop, r0 / CA_MD_KD_DIV);
goto fail_ctx;
}
- /* Flat mark: no rise, only the proportional cut. */
- prev = ctx->rate;
- mb_ecn_ece(ctx, 256, 0, 2 * MS);
- mb_ecn_snd(ctx, LEN, 2 * MS, &ftag);
- drop = prev - ctx->rate;
- if (drop > prev / 100) {
- printf("flat mark over-cut: dropped %" PRIu64 ".\n", drop);
+ /* Fall of one reference: give the same fraction back. */
+ ctx->rate = r0;
+ ctx->tx_ece = 1;
+ ctx->tx_ecp = CA_ECE_REF + 1;
+ ctx->dec_acc = 0;
+
+ mb_ecn_decrease(ctx, 0);
+
+ gain = ctx->rate - r0;
+ if (gain != r0 / CA_MD_KD_DIV) {
+ printf("fall boost %" PRIu64 ", want %" PRIu64 ".\n",
+ gain, r0 / CA_MD_KD_DIV);
goto fail_ctx;
}
- /* Falling mark: no rise, no lead beyond the proportional term. */
- prev = ctx->rate;
- mb_ecn_ece(ctx, 64, 0, 3 * MS);
- mb_ecn_snd(ctx, LEN, 3 * MS, &ftag);
- drop = prev > ctx->rate ? prev - ctx->rate : 0;
- if (drop > prev / 100) {
- printf("falling mark cut: dropped %" PRIu64 ".\n", drop);
+ /* A collapse from deep saturation is clamped to the same. */
+ ctx->rate = r0;
+ ctx->tx_ece = 1;
+ ctx->tx_ecp = 255 << CA_SHFT;
+ ctx->dec_acc = 0;
+
+ mb_ecn_decrease(ctx, 0);
+
+ gain = ctx->rate - r0;
+ if (gain != r0 / CA_MD_KD_DIV) {
+ printf("unclamped fall boost %" PRIu64 ".\n", gain);
+ goto fail_ctx;
+ }
+
+ /* A cycle that stays marked nets out: no standing bias. */
+ ctx->rate = r0;
+ ctx->tx_ecp = 4 << CA_SHFT;
+ ctx->tx_ece = 12 << CA_SHFT;
+ ctx->dec_acc = 0;
+
+ mb_ecn_decrease(ctx, 0);
+
+ ctx->tx_ece = 4 << CA_SHFT;
+ ctx->dec_acc = 0;
+
+ mb_ecn_decrease(ctx, 0);
+
+ net = ctx->rate > r0 ? ctx->rate - r0 : r0 - ctx->rate;
+ /*
+ * The two lead steps compound to (1 - x)(1 + x), x = 1 /
+ * (2 * CA_MD_KD_DIV), so net ~= r0 / (4 * CA_MD_KD_DIV^2).
+ * Band it a factor of 2 either side so a materially weaker
+ * gain (e.g. KD off by a factor of 4) fails the floor.
+ */
+ kd2 = (uint64_t) CA_MD_KD_DIV * CA_MD_KD_DIV;
+ if (net > r0 / (2 * kd2)) {
+ printf("cycle bias %" PRIu64 " of %" PRIu64 ".\n", net, r0);
+ goto fail_ctx;
+ }
+
+ if (net < r0 / (8 * kd2)) {
+ printf("lead gain weaker than expected: net %" PRIu64
+ " of %" PRIu64 ".\n", net, r0);
goto fail_ctx;
}
@@ -1722,8 +1832,6 @@ static int test_mb_ecn_probe_scale_invariant(void)
mb_ecn_ece(b, 0, 0, 0);
a->rate = (uint64_t) 10 << 20;
b->rate = (uint64_t) 1000 << 20;
- a->r_bkt = a->rate;
- b->r_bkt = b->rate;
a->backlogged = true;
b->backlogged = true;
ra0 = a->rate;
@@ -1787,8 +1895,8 @@ static int test_mb_ecn_probe_time_constant(void)
/* Clean path, out of slow start, backlogged, below the ceiling. */
mb_ecn_ece(ctx, 0, 0, 0);
+
ctx->rate = (uint64_t) 1 << 30;
- ctx->r_bkt = ctx->rate;
ctx->backlogged = true;
r0 = ctx->rate;
@@ -1799,11 +1907,10 @@ static int test_mb_ecn_probe_time_constant(void)
mb_ecn_ece(ctx, 0, 0, t);
}
- /* Washout damps the probe to ~4/3 TC, so 8 s -> ~2.12x. */
+ /* Undamped probe: 8 s at TC 8 s is one full e-fold, ~2.72x. */
ratio = (double) ctx->rate / r0;
- if (ratio < 2.0 || ratio > 2.25) {
- printf("probe TC off: exp ~2.12, got %.3fx over 8 s.\n",
- ratio);
+ if (ratio < 2.6 || ratio > 2.85) {
+ printf("probe TC off: exp ~2.72, got %.3fx over 8 s.\n", ratio);
goto fail_ctx;
}
@@ -1851,7 +1958,7 @@ static int test_mb_ecn_rcv_cap_window_min(void)
mb_ecn_rcv(ctx, LEN, 8, 40, &ece, &fcap, 2 * MS);
mb_ecn_rcv(ctx, LEN, 8, 36, &ece, &fcap, 3 * MS);
- t = 3 * MS + CA_TW_INIT;
+ t = 3 * MS + CA_TW;
if (!mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t)) {
printf("Window did not close.\n");
goto fail_ctx;
@@ -1865,7 +1972,7 @@ static int test_mb_ecn_rcv_cap_window_min(void)
/* The next window starts unknown; follow the adapted rx_tw. */
upd = false;
for (i = 0; i < 128 && !upd; i++) {
- t += CA_TW_INIT;
+ t += CA_TW;
upd = mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t);
}
@@ -1920,7 +2027,7 @@ static int test_mb_ecn_rcv_cap_onset_fresh(void)
mb_ecn_rcv(ctx, LEN, 4, 50, &ece, &fcap, 2 * MS);
/* A gap restart must not fold in the stale window min. */
- t = 2 * MS + 5 * CA_TW_INIT;
+ t = 2 * MS + 5 * CA_TW;
if (!mb_ecn_rcv(ctx, LEN, 4, 90, &ece, &fcap, t)) {
printf("Gap restart did not update.\n");
goto fail_ctx;
@@ -1970,7 +2077,7 @@ static int test_mb_ecn_ece_cap_derives_rates(void)
goto fail_ctx;
}
- if (ctx->ai_rate != ctx->rate_min) {
+ if (ctx->ai_rate != 2 * ctx->rate_min) {
printf("AI slope did not track the floor.\n");
goto fail_ctx;
}
@@ -2162,8 +2269,8 @@ static int test_mb_ecn_ctrl_per_ctx_ai(void)
/* Leave slow start; raise the slope as capacity would. */
mb_ecn_ece(ctx, 0, 0, 0);
+
ctx->rate = (uint64_t) 10 << 20;
- ctx->r_bkt = ctx->rate;
ctx->ai_rate = 16 * CA_AI_RATE;
want = ctx->rate + ctx->ai_rate * (30 * MS) / BILLION;
@@ -2407,15 +2514,15 @@ static int test_mb_ecn_backlogged_paced(void)
}
/*
- * A source-limited flow is capped to the next quarter-log2 headroom
- * above the offered estimate, and NOT re-floored to a high capacity
- * rate_min.
+ * A source-limited flow is capped to the backlog level above the
+ * offered estimate, and NOT re-floored to a high capacity rate_min.
*/
static int test_mb_ecn_source_limited_ceiling(void)
{
struct mb_ecn_ctx * ctx;
uint64_t ftag = 0;
uint64_t t = 10 * MS;
+ uint64_t expect;
TEST_START();
@@ -2439,9 +2546,10 @@ static int test_mb_ecn_source_limited_ceiling(void)
mb_ecn_snd(ctx, LEN, t + 2 * MS, &ftag);
- if (ctx->rate != ((uint64_t) 2 << 20)) {
+ expect = 1398101; /* (1 << 20) * 4 / 3, truncated */
+ if (ctx->rate != expect) {
printf("ceiling: exp %" PRIu64 ", got %" PRIu64 ".\n",
- (uint64_t) 2 << 20, ctx->rate);
+ expect, ctx->rate);
goto fail_ctx;
}
@@ -2452,6 +2560,37 @@ static int test_mb_ecn_source_limited_ceiling(void)
mb_ecn_ctx_destroy(ctx);
+ /* Non-power-of-two case: verify the exact level, not a step. */
+
+ ctx = mk_ctx();
+ if (ctx == NULL) {
+ printf("Failed to create context.\n");
+ goto fail;
+ }
+
+ ctx->tx_cav = true;
+ ctx->started = true;
+ ctx->backlogged = false;
+ ctx->rate = (uint64_t) 100 << 20;
+ ctx->inv_rate = mb_ecn_rate_inv(ctx->rate);
+ ctx->rate_min = (uint64_t) 50 << 20;
+ ctx->snd_rate = (uint64_t) 303 << 12;
+ ctx->snd_r0 = ctx->rate;
+ ctx->snd_win = t;
+ ctx->last_ts = t;
+ ctx->last_ctrl = t;
+
+ mb_ecn_snd(ctx, LEN, t + 2 * MS, &ftag);
+
+ expect = 1654784; /* (303 << 12) * 4 / 3, exact */
+ if (ctx->rate != expect) {
+ printf("exact ceiling: exp %" PRIu64 ", got %" PRIu64
+ ".\n", expect, ctx->rate);
+ goto fail_ctx;
+ }
+
+ mb_ecn_ctx_destroy(ctx);
+
TEST_SUCCESS();
return TEST_RC_SUCCESS;
@@ -2526,6 +2665,7 @@ static int test_mb_ecn_idle_clears_backlogged(void)
ctx->started = true;
ctx->backlogged = true;
+ ctx->rate = snd_rate;
ctx->snd_rate = snd_rate;
ctx->snd_win = t;
ctx->last_ts = t;
@@ -2638,7 +2778,7 @@ static void shared_link_run(struct mb_ecn_ctx * a,
tgt = (k + 1) * TF_STEP;
ecn = 0;
- mb_ecn_calc_ecn(q / LEN, &ecn, QOS_CUBE_BE, LEN);
+ mb_ecn_calc_ecn(q, &ecn, QOS_CUBE_BE, LEN);
hist[k % TF_HIST] = (uint16_t) (ecn << CA_SHFT);
ea = k < lag_a ? 0 : hist[(k - lag_a) % TF_HIST];
@@ -2775,16 +2915,14 @@ static int test_mb_ecn_ramp_overshoot_grows_with_rtt(void)
}
/*
- * Washout damps a fixed 1/4 of the rate change once per wall-clock
- * bucket: a sub-bucket step banks time only, a full bucket removes a
- * quarter of the gap either way without crossing the snapshot, and a
- * sparse step (> CA_DT_CAP) resets it so a starved sender keeps its cut.
+ * Flows sharing a context offer bytes together but each may send at
+ * rate, so the window must be shared out before the backlog level is
+ * read. Four flows offering two thirds of a share each stay below it.
*/
-static int test_mb_ecn_washout_bucket(void)
+static int test_mb_ecn_shared_ctx_offered_per_flow(void)
{
struct mb_ecn_ctx * ctx;
- uint64_t r0 = (uint64_t) 100 << 20;
- uint64_t gap;
+ uint64_t rate = 1000000;
TEST_START();
@@ -2794,56 +2932,148 @@ static int test_mb_ecn_washout_bucket(void)
goto fail;
}
- gap = r0 >> 4;
+ ctx->rate = rate;
+ ctx->snd_flows = 4;
+ ctx->snd_r0 = rate;
+ ctx->snd_win = 0;
+ ctx->snd_byt = 4 * rate * 2 / 3;
- /* Sub-bucket: time banks, the rate does not move. */
- ctx->rate = r0;
- ctx->r_bkt = r0 - gap;
- ctx->wash_acc = 0;
- mb_ecn_washout(ctx, MS, MS);
- if (ctx->rate != r0 || ctx->wash_acc != MS) {
- printf("sub-bucket washout moved the rate: %" PRIu64 ".\n",
- ctx->rate);
+ mb_ecn_win(ctx, BILLION);
+
+ if (ctx->backlogged) {
+ printf("aggregate load read as one flow's backlog.\n");
goto fail_ctx;
}
- /* Full bucket, rate leads: cut a quarter of the gap, no crossing. */
- ctx->rate = r0;
- ctx->r_bkt = r0 - gap;
- ctx->wash_acc = 0;
- mb_ecn_washout(ctx, MS, CA_WASH_BKT);
- if (ctx->rate != r0 - (gap >> CA_WASH_SHFT)) {
- printf("bucket down: got %" PRIu64 ".\n", ctx->rate);
+ /* The close left a fresh window; a full share each clears it. */
+ ctx->snd_byt = 4 * rate;
+
+ mb_ecn_win(ctx, 2 * BILLION);
+
+ if (!ctx->backlogged) {
+ printf("per-flow share did not read as backlogged.\n");
+ goto fail_ctx;
+ }
+
+ mb_ecn_ctx_destroy(ctx);
+
+ TEST_SUCCESS();
+
+ return TEST_RC_SUCCESS;
+ fail_ctx:
+ mb_ecn_ctx_destroy(ctx);
+ fail:
+ TEST_FAIL();
+ return TEST_RC_FAIL;
+}
+
+/*
+ * A join or a leave opens a fresh window, so none divides the bytes
+ * one population offered by the count of another. An unchanged count
+ * leaves the running window alone.
+ */
+static int test_mb_ecn_flow_count_restarts_window(void)
+{
+ struct mb_ecn_ctx * ctx;
+ uint64_t rate = 1000000;
+
+ TEST_START();
+
+ ctx = mk_ctx();
+ if (ctx == NULL) {
+ printf("Failed to create context.\n");
+ goto fail;
+ }
+
+ ctx->rate = rate;
+ ctx->snd_flows = 2;
+ ctx->snd_win = MS;
+ ctx->snd_byt = 12345;
+ ctx->snd_r0 = 7;
+
+ mb_ecn_flows(ctx, 5, 8 * MS);
+
+ if (ctx->snd_flows != 5 || ctx->snd_byt != 0
+ || ctx->snd_win != 8 * MS || ctx->snd_r0 != rate) {
+ printf("count change left a stale window: flows=%zu "
+ "byt=%" PRIu64 " win=%" PRIu64 " r0=%" PRIu64
+ ".\n", ctx->snd_flows, ctx->snd_byt,
+ ctx->snd_win, ctx->snd_r0);
goto fail_ctx;
}
- if (ctx->rate <= r0 - gap) {
- printf("washout reversed a ramp: %" PRIu64 ".\n", ctx->rate);
+ ctx->snd_byt = 999;
+
+ mb_ecn_flows(ctx, 5, 20 * MS);
+
+ if (ctx->snd_byt != 999 || ctx->snd_win != 8 * MS) {
+ printf("unchanged count restarted the window.\n");
goto fail_ctx;
}
- if (ctx->r_bkt != ctx->rate || ctx->wash_acc != 0) {
- printf("washout did not reset the bucket.\n");
+ /* An empty context still measures a single sender. */
+ mb_ecn_flows(ctx, 0, 30 * MS);
+
+ if (ctx->snd_flows != 1) {
+ printf("zero flows did not floor at one: %zu.\n",
+ ctx->snd_flows);
goto fail_ctx;
}
- /* Full bucket, rate trails: add a quarter of the gap (symmetric). */
- ctx->rate = r0;
- ctx->r_bkt = r0 + gap;
- ctx->wash_acc = 0;
- mb_ecn_washout(ctx, MS, CA_WASH_BKT);
- if (ctx->rate != r0 + (gap >> CA_WASH_SHFT)) {
- printf("bucket up: got %" PRIu64 ".\n", ctx->rate);
+ mb_ecn_ctx_destroy(ctx);
+
+ TEST_SUCCESS();
+
+ return TEST_RC_SUCCESS;
+ fail_ctx:
+ mb_ecn_ctx_destroy(ctx);
+ fail:
+ TEST_FAIL();
+ return TEST_RC_FAIL;
+}
+
+/*
+ * The ceiling must land where a window of the delivered rate reads
+ * backlogged again: a context clamped above that level can never
+ * leave the clamp, and loses its capacity floor with it.
+ */
+static int test_mb_ecn_ceiling_clears_backlog(void)
+{
+ struct mb_ecn_ctx * ctx;
+ uint64_t x = 1 << 20;
+
+ TEST_START();
+
+ ctx = mk_ctx();
+ if (ctx == NULL) {
+ printf("Failed to create context.\n");
+ goto fail;
+ }
+
+ ctx->backlogged = false;
+ ctx->snd_rate = x;
+ ctx->rate = 100 * x;
+ ctx->rate_min = CA_RATE_MIN;
+
+ mb_ecn_ceiling(ctx);
+
+ if (ctx->rate >= 100 * x) {
+ printf("ceiling did not bind: %" PRIu64 ".\n", ctx->rate);
goto fail_ctx;
}
- /* Sparse step: reset to the current rate, keep the cut. */
- ctx->rate = r0;
- ctx->r_bkt = r0 - gap;
- ctx->wash_acc = CA_WASH_BKT / 2;
- mb_ecn_washout(ctx, CA_DT_CAP + 1, MS);
- if (ctx->rate != r0 || ctx->r_bkt != r0 || ctx->wash_acc != 0) {
- printf("sparse step did not reset the bucket.\n");
+ /* One window delivering x, with the pacer deferring nothing. */
+ ctx->snd_r0 = ctx->rate;
+ ctx->snd_flows = 1;
+ ctx->snd_win = 0;
+ ctx->snd_byt = x;
+ ctx->snd_pac = 0;
+
+ mb_ecn_win(ctx, BILLION);
+
+ if (!ctx->backlogged) {
+ printf("clamped at %" PRIu64 " cannot clear on %" PRIu64
+ ".\n", ctx->snd_r0, x);
goto fail_ctx;
}
@@ -2868,15 +3098,15 @@ int mb_ecn_test(int argc,
(void) argv;
ret |= test_mb_ecn_ctx_create_destroy();
+ ret |= test_mb_ecn_init_window();
ret |= test_mb_ecn_calc_ecn();
ret |= test_mb_ecn_rcv_onset_immediate();
- ret |= test_mb_ecn_rcv_window_mean();
ret |= test_mb_ecn_rcv_rate_independent();
ret |= test_mb_ecn_rcv_size_fair();
ret |= test_mb_ecn_rcv_release_exact_zero();
ret |= test_mb_ecn_rcv_gap_restart();
+ ret |= test_mb_ecn_rcv_gap_floor();
ret |= test_mb_ecn_rcv_accum_bounds();
- ret |= test_mb_ecn_rcv_window_holds_target();
ret |= test_mb_ecn_rcv_window_clip_bounds();
ret |= test_mb_ecn_rcv_slow_window();
ret |= test_mb_ecn_rcv_no_overflow_highrate();
@@ -2885,8 +3115,9 @@ int mb_ecn_test(int argc,
ret |= test_mb_ecn_dt_scaling_invariant();
ret |= test_mb_ecn_probe_scale_invariant();
ret |= test_mb_ecn_multiplicative_decrease();
+ ret |= test_mb_ecn_ai_hold_release();
ret |= test_mb_ecn_fixed_point();
- ret |= test_mb_ecn_lead_cut();
+ ret |= test_mb_ecn_lead_symmetric();
ret |= test_mb_ecn_slow_start_local_brake();
ret |= test_mb_ecn_slow_start_clean_ramp();
ret |= test_mb_ecn_starved_decrease_escape();
@@ -2917,7 +3148,9 @@ int mb_ecn_test(int argc,
ret |= test_mb_ecn_first_send_warmup();
ret |= test_mb_ecn_two_flow_converge();
ret |= test_mb_ecn_ramp_overshoot_grows_with_rtt();
- ret |= test_mb_ecn_washout_bucket();
+ ret |= test_mb_ecn_shared_ctx_offered_per_flow();
+ ret |= test_mb_ecn_flow_count_restarts_window();
+ ret |= test_mb_ecn_ceiling_clears_backlog();
return ret;
}
diff --git a/src/ipcpd/unicast/cap.c b/src/ipcpd/unicast/cap.c
index 0d823dc6..67b7967c 100644
--- a/src/ipcpd/unicast/cap.c
+++ b/src/ipcpd/unicast/cap.c
@@ -1,7 +1,7 @@
/*
* Ouroboros - Copyright (C) 2016 - 2026
*
- * Link capacity estimation
+ * Link capacity codes
*
* Dimitri Staessens <dimitri@ouroboros.rocks>
* Sander Vrijders <sander@ouroboros.rocks>
@@ -20,112 +20,15 @@
* Foundation, Inc., http://www.fsf.org/about/contact/.
*/
-#if defined(__linux__) || defined(__CYGWIN__)
-#define _DEFAULT_SOURCE
-#else
-#define _POSIX_C_SOURCE 200809L
-#endif
-
-#include "config.h"
-
-#include <ouroboros/atomics.h>
-#include <ouroboros/time.h>
-
-#include "cap.h"
-
-#include <string.h>
-
/*
- * Link-capacity estimation by watching the egress queue drain.
- *
- * A saturated link drains its queue at exactly its capacity, so we
- * estimate capacity by measuring the drain rate of the ring buffer
- * toward an n-1 flow (the flow to the layer below) while that ring
- * is backlogged.
- *
- * Sampling is lock-free and off the fast path: the ring depth is
- * read only at enqueue time, concurrently by many sender threads.
- * Each enqueue bumps relaxed counters (packets, bytes, empty-ring
- * hits). At most once per CAP_T_MIN, one thread wins a try-lock and
- * closes a measurement window.
- *
- * Over a window, packet conservation gives the slots that drained:
- * drained = queue at start (q0) + enqueued - queue now (q1)
- * A window stays open until CAP_N_MIN slots have drained, so its
- * length self-scales with the link rate (~1 ms at 1 Gbit, ~19 ms at
- * 10 Mbit). CAP_T_MAX discards a window that spanned a traffic gap.
- *
- * Only a backlogged link measures its own capacity, so a window
- * whose ring ran mostly idle is discarded (a few empty samples, as
- * from a token-bucket shaper, are tolerated). The drain rate feeds a
- * max filter that jumps up at once but decays slowly, converging on
- * the capacity from below. A window that touched an empty ring at
- * either edge may have drained into downstream buffers faster than
- * the wire, so it may only lower the estimate, never raise it.
- *
- * The estimate is published as a quarter-log2 code: capacity is only
- * ever needed to order-of-magnitude accuracy.
+ * Rate <-> 8-bit code (cap_enc / cap_dec): the high 6 bits hold a
+ * band e = floor(log2 rate), the low 2 a quarter k splitting the
+ * band at 256 * 2^(k/4) = {256, 304, 362, 431}; code = 4 * e + k.
+ * Capacity is only ever needed to order-of-magnitude accuracy.
*/
-#define CAP_T_MIN (BILLION / 1000) /* min fold spacing ~1 ms */
-#define CAP_T_MAX (1ULL << 27) /* stale window cap ~134 ms */
-#define CAP_N_MIN 16 /* drained slots to close */
-#define CAP_DEC_SHFT 4 /* max-filter decay 1/16 */
-#define CAP_IDL_SHFT 3 /* idle tolerance 1/8 */
-
-/* Try-lock on the busy flag: test-and-set acquire, store release. */
-#define CAP_TRY(p) (__atomic_exchange_n(p, 1, __ATOMIC_ACQUIRE) == 0)
-#define CAP_REL(p) (__atomic_store_n(p, 0, __ATOMIC_RELEASE))
-
-struct cap_est {
- uint64_t c_pkt; /* total packets enqueued (relaxed) */
- uint64_t c_byt; /* total bytes enqueued (relaxed) */
- uint64_t c_idl; /* times ring seen empty (relaxed) */
-
- uint64_t t_gate; /* last fold timestamp (ns) */
- uint8_t busy; /* fold in progress (try-lock) */
-
- uint64_t t0; /* window start (ns), 0 = no window */
- uint64_t q0; /* ring occupancy at window start */
- uint64_t pkt0; /* c_pkt snapshot at window start */
- uint64_t byt0; /* c_byt snapshot at window start */
- uint64_t idl0; /* c_idl snapshot at window start */
- uint64_t rate; /* filtered drain rate (bytes/s) */
-
- uint8_t cap; /* published capacity code (0=none) */
-};
-
-struct {
- struct cap_est est[PROC_MAX_FLOWS];
-} cap;
-
-int cap_init(void)
-{
- memset(&cap, 0, sizeof(cap));
-
- return 0;
-}
-
-void cap_fini(void)
-{
-}
-
-void cap_reset(int fd)
-{
- /* A racing update seeds one bogus window; the filter absorbs. */
- memset(&cap.est[fd], 0, sizeof(cap.est[fd]));
-}
+#include "cap.h"
-/*
- * Rate <-> 8-bit code (cap_enc / cap_dec). The code is a tiny float:
- * the high 6 bits are a band e = floor(log2 rate), the low 2 bits a
- * quarter k that splits each band [2^e, 2^(e+1)) into four, so
- * code = 4 * e + k. Each step is ~19% in rate; that coarseness is
- * deliberate, capacity only needs order-of-magnitude accuracy.
- *
- * The quarter cut points are 256 * 2^(k/4) rounded to an integer:
- * {256, 304, 362, 431} over the normalized range [256, 512).
- */
uint8_t cap_enc(uint64_t rate)
{
static const uint16_t thr[3] = {304, 362, 431};
@@ -143,9 +46,10 @@ uint8_t cap_enc(uint64_t rate)
e++;
}
- /* Top 9 bits: rate normalized to [256, 512). */
- top = e >= 8 ? (uint16_t) (rate >> (e - 8))
- : (uint16_t) (rate << (8 - e));
+ if (e >= 8)
+ top = (uint16_t) (rate >> (e - 8));
+ else
+ top = (uint16_t) (rate << (8 - e));
while (k < 3 && top >= thr[k])
k++;
@@ -193,99 +97,3 @@ void cap_stamp(uint8_t * pci,
if (*pci == 0 || own < *pci)
*pci = own;
}
-
-/* Fold flag held; q1 is the caller's pre-write ring sample. */
-static void cap_fold(struct cap_est * e,
- uint64_t q1,
- uint64_t now)
-{
- uint64_t pkt; /* current c_pkt snapshot */
- uint64_t byt; /* current c_byt snapshot */
- uint64_t idl; /* current c_idl snapshot */
- uint64_t dt; /* window duration (ns) */
- uint64_t enq; /* packets enqueued in window */
- uint64_t avg; /* mean packet size (bytes) */
- uint64_t r; /* window drain rate (bytes/s) */
- int64_t drained; /* slots drained over window */
-
- pkt = LOAD_RELAXED(&e->c_pkt);
- byt = LOAD_RELAXED(&e->c_byt);
- idl = LOAD_RELAXED(&e->c_idl);
-
- dt = now - e->t0;
- enq = pkt - e->pkt0;
-
- drained = (int64_t) (e->q0 + enq - q1);
-
- if (e->t0 == 0 || dt > CAP_T_MAX || enq == 0)
- goto reopen;
-
- if (drained < (int64_t) CAP_N_MIN)
- return; /* extend the window until enough drains */
-
- if ((idl - e->idl0) << CAP_IDL_SHFT > enq)
- goto reopen; /* mostly idle ring: not saturated */
-
- avg = (byt - e->byt0) / enq;
- r = (uint64_t) drained * avg * MILLION / (dt / 1000);
-
- if (r >= e->rate) {
- /* Empty-edged windows drain into buffers below. */
- if (e->q0 > 0 && q1 > 0)
- e->rate = r;
- } else {
- e->rate -= (e->rate - r) >> CAP_DEC_SHFT;
- }
-
- STORE_RELAXED(&e->cap, cap_enc(e->rate));
- reopen:
- e->t0 = now;
- e->q0 = q1;
- e->pkt0 = pkt;
- e->byt0 = byt;
- e->idl0 = idl;
-}
-
-/* Internal, timestamped entry point; tests drive this directly. */
-static void cap_update_at(int fd,
- size_t qlen,
- size_t len,
- uint64_t now)
-{
- struct cap_est * e = &cap.est[fd]; /* this flow's estimator */
-
- FETCH_ADD_RELAXED(&e->c_pkt, 1);
- FETCH_ADD_RELAXED(&e->c_byt, len);
-
- if (qlen == 0)
- FETCH_ADD_RELAXED(&e->c_idl, 1);
-
- if (now - LOAD_RELAXED(&e->t_gate) < CAP_T_MIN)
- return;
-
- if (!CAP_TRY(&e->busy))
- return;
-
- if (now - e->t_gate >= CAP_T_MIN) {
- cap_fold(e, qlen, now);
- STORE_RELAXED(&e->t_gate, now);
- }
-
- CAP_REL(&e->busy);
-}
-
-void cap_update(int fd,
- size_t qlen,
- size_t len)
-{
- struct timespec now;
-
- clock_gettime(PTHREAD_COND_CLOCK, &now);
-
- cap_update_at(fd, qlen, len, TS_TO_UINT64(now));
-}
-
-uint8_t cap_get(int fd)
-{
- return LOAD_RELAXED(&cap.est[fd].cap);
-}
diff --git a/src/ipcpd/unicast/cap.h b/src/ipcpd/unicast/cap.h
index df8c2ec1..ca6b6355 100644
--- a/src/ipcpd/unicast/cap.h
+++ b/src/ipcpd/unicast/cap.h
@@ -1,7 +1,7 @@
/*
* Ouroboros - Copyright (C) 2016 - 2026
*
- * Link capacity estimation
+ * Link capacity codes
*
* Dimitri Staessens <dimitri@ouroboros.rocks>
* Sander Vrijders <sander@ouroboros.rocks>
@@ -23,22 +23,8 @@
#ifndef OUROBOROS_IPCPD_UNICAST_CAP_H
#define OUROBOROS_IPCPD_UNICAST_CAP_H
-#include <stddef.h>
#include <stdint.h>
-int cap_init(void);
-
-void cap_fini(void);
-
-/* Account an egress packet; qlen is sampled before the write. */
-void cap_update(int fd,
- size_t qlen,
- size_t len);
-
-uint8_t cap_get(int fd);
-
-void cap_reset(int fd);
-
/* Quarter-log2 capacity code: ~2^(c / 4) bytes/s, 0 = unknown. */
uint8_t cap_enc(uint64_t rate);
diff --git a/src/ipcpd/unicast/dt.c b/src/ipcpd/unicast/dt.c
index bfc2ece7..84e62f05 100644
--- a/src/ipcpd/unicast/dt.c
+++ b/src/ipcpd/unicast/dt.c
@@ -424,7 +424,8 @@ static void handle_event(void * self,
#ifdef IPCP_FLOW_STATS
stat_used(fd, c->conn_info.addr);
#endif
- cap_reset(fd);
+ if (ipcp_flow_cap_arm(fd) < 0)
+ log_warn("Failed to arm capacity estimator.");
psched_add(dt.psched, fd);
log_dbg("Added fd %d to packet scheduler.", fd);
break;
@@ -451,6 +452,8 @@ static time_t packet_handler(int fd,
uint8_t * head;
size_t len;
size_t qlen;
+ size_t mlen;
+ uint8_t lcap;
bool marks;
len = ssm_pk_buff_len(spb);
@@ -485,11 +488,13 @@ static time_t packet_handler(int fd,
marks = ca_marks_ecn();
qlen = marks ? ipcp_flow_queued(ofd) : 0;
+ mlen = marks ? ipcp_flow_mean_len(ofd) : 0;
+ lcap = marks ? cap_enc(ipcp_flow_cap(ofd)) : 0;
- (void) ca_calc_ecn(qlen, head + dt_pci_info.ecn_o, qc, len);
+ (void) ca_calc_ecn(qlen, head + dt_pci_info.ecn_o, qc, mlen);
if (marks)
- cap_stamp(head + dt_pci_info.cap_o, cap_get(ofd));
+ cap_stamp(head + dt_pci_info.cap_o, lcap);
ret = ipcp_flow_write(ofd, spb);
if (ret < 0) {
@@ -504,7 +509,7 @@ static time_t packet_handler(int fd,
dt_stat_inc(ofd, snd, qc, len);
if (marks)
- cap_update(ofd, qlen, len);
+ ipcp_flow_cap_update(ofd, qlen, len);
} else {
dt_pci_shrink(spb);
if (dt_pci.eid >= PROC_RES_FDS) {
@@ -590,11 +595,6 @@ int dt_init(struct dt_config cfg)
dt_pci_info.eid_o = dt_pci_info.cap_o + CAP_LEN;
dt_pci_info.head_size = dt_pci_info.eid_o + dt_pci_info.eid_size;
- if (cap_init() < 0) {
- log_err("Failed to init capacity estimator.");
- goto fail_cap;
- }
-
if (connmgr_comp_init(COMPID_DT, &info) != 0) {
log_err("Failed to register with connmgr.");
goto fail_connmgr_comp_init;
@@ -663,8 +663,6 @@ int dt_init(struct dt_config cfg)
fail_routing:
connmgr_comp_fini(COMPID_DT);
fail_connmgr_comp_init:
- cap_fini();
- fail_cap:
return -1;
}
@@ -692,8 +690,6 @@ void dt_fini(void)
routing_fini();
connmgr_comp_fini(COMPID_DT);
-
- cap_fini();
}
int dt_start(void)
@@ -805,6 +801,8 @@ int dt_write_packet(uint64_t dst_addr,
uint8_t * head;
size_t len;
size_t qlen;
+ size_t mlen;
+ uint8_t lcap;
bool marks;
assert(spb);
@@ -843,11 +841,12 @@ int dt_write_packet(uint64_t dst_addr,
marks = ca_marks_ecn();
qlen = marks ? ipcp_flow_queued(fd) : 0;
+ mlen = marks ? ipcp_flow_mean_len(fd) : 0;
+ lcap = marks ? cap_enc(ipcp_flow_cap(fd)) : 0;
- (void) ca_calc_ecn(qlen, &dt_pci.ecn, qc, len);
+ (void) ca_calc_ecn(qlen, &dt_pci.ecn, qc, mlen);
- if (marks)
- dt_pci.cap = cap_get(fd);
+ dt_pci.cap = lcap;
if (ecn != NULL)
*ecn = dt_pci.ecn;
@@ -856,7 +855,7 @@ int dt_write_packet(uint64_t dst_addr,
ret = ipcp_flow_write(fd, spb);
if (ret < 0) {
- log_dbg("Failed to write packet to fd %d.", fd);
+ log_dbg("Failed to write packet to fd %d: %d.", fd, ret);
if (ret == -EFLOWDOWN)
notifier_event(NOTIFY_DT_FLOW_DOWN, &fd);
goto fail_write;
@@ -868,7 +867,7 @@ int dt_write_packet(uint64_t dst_addr,
dt_stat_inc(fd, snd, qc, len);
#endif
if (marks)
- cap_update(fd, qlen, len);
+ ipcp_flow_cap_update(fd, qlen, len);
return 0;
diff --git a/src/ipcpd/unicast/tests/cap_test.c b/src/ipcpd/unicast/tests/cap_test.c
index 7867b490..e3c3f8b3 100644
--- a/src/ipcpd/unicast/tests/cap_test.c
+++ b/src/ipcpd/unicast/tests/cap_test.c
@@ -1,7 +1,7 @@
/*
* Ouroboros - Copyright (C) 2016 - 2026
*
- * Unit tests for link capacity estimation
+ * Unit tests for link capacity codes
*
* Dimitri Staessens <dimitri@ouroboros.rocks>
* Sander Vrijders <sander@ouroboros.rocks>
@@ -24,41 +24,6 @@
#include <test/test.h>
-#define TICK (50 * 1000ULL) /* 50 us between packets */
-#define LEN 1000ULL /* default packet size (B) */
-#define QLEN 8 /* steady ring backlog */
-#define RATE (LEN * BILLION / TICK) /* LEN per TICK = 20 MB/s */
-
-#define SHP_LEN 1250ULL /* shaped-link packet (B) */
-#define SHP_STEP 20 /* packets per shaped window */
-#define SHP_RATE (SHP_LEN * BILLION / (SHP_STEP * TICK))
-
-static int test_cap_init_fini(void)
-{
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- if (cap_get(0) != 0 || cap_get(PROC_MAX_FLOWS - 1) != 0) {
- printf("Fresh estimator not unknown.\n");
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
/* Exact roundtrip holds for codes >= 32 (rates >= 256 B/s). */
static int test_cap_codec_roundtrip(void)
{
@@ -152,28 +117,36 @@ static int test_cap_stamp(void)
TEST_START();
pci = 42;
+
cap_stamp(&pci, 0);
+
if (pci != 42) {
printf("Unknown own code overwrote the byte.\n");
goto fail;
}
pci = 0;
+
cap_stamp(&pci, 97);
+
if (pci != 97) {
printf("Own code not written into unknown.\n");
goto fail;
}
pci = 97;
+
cap_stamp(&pci, 42);
+
if (pci != 42) {
printf("Lower own code did not lower the byte.\n");
goto fail;
}
pci = 42;
+
cap_stamp(&pci, 97);
+
if (pci != 42) {
printf("Higher own code raised the byte.\n");
goto fail;
@@ -187,384 +160,6 @@ static int test_cap_stamp(void)
return TEST_RC_FAIL;
}
-static int test_cap_est_busy_window(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- /* 1000 B every 50 us, ring steady at 8: drain = 20 MB/s. */
- for (i = 1; i <= 40; i++)
- cap_update_at(0, QLEN, LEN, i * TICK);
-
- if (cap_get(0) != cap_enc(RATE)) {
- printf("Estimated code: exp %u, got %u.\n",
- cap_enc(RATE), cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-static int test_cap_est_idle_tolerated(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- for (i = 1; i <= 40; i++)
- cap_update_at(0, i == 21 ? 0 : QLEN, LEN, i * TICK);
-
- if (cap_get(0) != cap_enc(RATE)) {
- printf("Grazed window: exp %u, got %u.\n",
- cap_enc(RATE), cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-static int test_cap_est_mostly_idle_rejects(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- for (i = 1; i <= 100; i++)
- cap_update_at(0, 0, LEN, i * TICK);
-
- if (cap_get(0) != 0) {
- printf("Idle ring estimated %u.\n", cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-static int test_cap_est_slow_link_extends(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- /* 1000 B every 100 us: 10 slots/ms closes on a 2 ms window. */
- for (i = 1; i <= 30; i++)
- cap_update_at(0, QLEN, LEN, i * 2 * TICK);
-
- if (cap_get(0) != cap_enc(RATE / 2)) {
- printf("Slow link: exp %u, got %u.\n",
- cap_enc(RATE / 2), cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-static int test_cap_est_shaped_link(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- /* 1250 B every ms; one empty observation per 20 packets. */
- for (i = 1; i <= 100; i++)
- cap_update_at(0, i % SHP_STEP == 0 ? 0 : 6, SHP_LEN,
- i * SHP_STEP * TICK);
-
- if (cap_get(0) != cap_enc(SHP_RATE)) {
- printf("Shaped link: exp %u, got %u.\n",
- cap_enc(SHP_RATE), cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-static int test_cap_est_stale_discard(void)
-{
- uint64_t t;
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- /* Open a window, trickle 4 slots, then ~200 ms of silence. */
- for (i = 1; i <= 5; i++)
- cap_update_at(0, QLEN, LEN, i * CAP_T_MIN);
-
- t = 205 * CAP_T_MIN;
-
- cap_update_at(0, QLEN, LEN, t);
-
- if (cap_get(0) != 0) {
- printf("Gap window estimated %u.\n", cap_get(0));
- goto fail_init;
- }
-
- for (i = 1; i <= 40; i++)
- cap_update_at(0, QLEN, LEN, t + i * TICK);
-
- if (cap_get(0) != cap_enc(RATE)) {
- printf("Post-gap: exp %u, got %u.\n",
- cap_enc(RATE), cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-static int test_cap_est_empty_start_no_raise(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- cap_update_at(0, 0, LEN, CAP_T_MIN);
-
- for (i = 1; i <= 40; i++)
- cap_update_at(0, QLEN, LEN, CAP_T_MIN + i * TICK);
-
- if (cap_get(0) != 0) {
- printf("Empty-start window raised to %u.\n",
- cap_get(0));
- goto fail_init;
- }
-
- for (i = 41; i <= 60; i++)
- cap_update_at(0, QLEN, LEN, CAP_T_MIN + i * TICK);
-
- if (cap_get(0) != cap_enc(RATE)) {
- printf("Backlogged window: exp %u, got %u.\n",
- cap_enc(RATE), cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-/* Max filter: fast attack on a high sample, slow release on lower. */
-static int test_cap_est_max_filter(void)
-{
- uint8_t high;
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- for (i = 1; i <= 40; i++)
- cap_update_at(0, QLEN, LEN, i * TICK);
-
- high = cap_get(0);
- if (high != cap_enc(RATE)) {
- printf("Attack missed: exp %u, got %u.\n", cap_enc(RATE),
- high);
- goto fail_init;
- }
-
- /* Halved packet size: valid samples at 10 MB/s. */
- for (i = 41; i <= 80; i++)
- cap_update_at(0, QLEN, LEN / 2, i * TICK);
-
- if (cap_get(0) >= high) {
- printf("Release did not decay: %u.\n", cap_get(0));
- goto fail_init;
- }
-
- if (cap_get(0) <= cap_enc(RATE / 2)) {
- printf("Release collapsed to %u.\n", cap_get(0));
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-/* No fold within CAP_T_MIN of the previous one. */
-static int test_cap_est_gate(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- cap_update_at(0, QLEN, LEN, CAP_T_MIN);
-
- for (i = 0; i < 5; i++)
- cap_update_at(0, QLEN, LEN, CAP_T_MIN + CAP_T_MIN / 2);
-
- if (cap.est[0].t_gate != CAP_T_MIN) {
- printf("Fold ran inside the gate.\n");
- goto fail_init;
- }
-
- if (LOAD_RELAXED(&cap.est[0].c_pkt) != 6) {
- printf("Gated packets not counted.\n");
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
-static int test_cap_reset(void)
-{
- size_t i;
-
- TEST_START();
-
- if (cap_init() < 0) {
- printf("Failed to init cap.\n");
- goto fail;
- }
-
- for (i = 1; i <= 40; i++)
- cap_update_at(0, QLEN, LEN, i * TICK);
-
- if (cap_get(0) == 0) {
- printf("No estimate to reset.\n");
- goto fail_init;
- }
-
- cap_reset(0);
-
- if (cap_get(0) != 0) {
- printf("Reset did not clear the estimate.\n");
- goto fail_init;
- }
-
- cap_fini();
-
- TEST_SUCCESS();
-
- return TEST_RC_SUCCESS;
- fail_init:
- cap_fini();
- fail:
- TEST_FAIL();
- return TEST_RC_FAIL;
-}
-
int cap_test(int argc,
char ** argv)
{
@@ -573,21 +168,10 @@ int cap_test(int argc,
(void) argc;
(void) argv;
- ret |= test_cap_init_fini();
ret |= test_cap_codec_roundtrip();
ret |= test_cap_codec_bounds();
ret |= test_cap_min();
ret |= test_cap_stamp();
- ret |= test_cap_est_busy_window();
- ret |= test_cap_est_idle_tolerated();
- ret |= test_cap_est_mostly_idle_rejects();
- ret |= test_cap_est_slow_link_extends();
- ret |= test_cap_est_shaped_link();
- ret |= test_cap_est_stale_discard();
- ret |= test_cap_est_empty_start_no_raise();
- ret |= test_cap_est_max_filter();
- ret |= test_cap_est_gate();
- ret |= test_cap_reset();
return ret;
}