diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/ipcpd/unicast/ca.c | 6 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca.h | 6 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/mb-ecn.c | 371 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/mb-ecn.h | 3 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/nop.c | 6 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/nop.h | 3 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/ops.h | 4 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/tests/CMakeLists.txt | 34 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/tests/mb_ecn_lab_test.c | 1294 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/tests/mb_ecn_test.c | 723 | ||||
| -rw-r--r-- | src/ipcpd/unicast/cap.c | 212 | ||||
| -rw-r--r-- | src/ipcpd/unicast/cap.h | 16 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dt.c | 35 | ||||
| -rw-r--r-- | src/ipcpd/unicast/tests/cap_test.c | 434 | ||||
| -rw-r--r-- | src/lib/cap.c | 187 | ||||
| -rw-r--r-- | src/lib/cap.h | 63 | ||||
| -rw-r--r-- | src/lib/config.h.in | 1 | ||||
| -rw-r--r-- | src/lib/dev.c | 145 | ||||
| -rw-r--r-- | src/lib/poa/poa.c | 33 | ||||
| -rw-r--r-- | src/lib/poa/poa.h | 10 | ||||
| -rw-r--r-- | src/lib/ssm/rbuff.c | 195 | ||||
| -rw-r--r-- | src/lib/ssm/tests/rbuff_test.c | 392 | ||||
| -rw-r--r-- | src/lib/tests/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | src/lib/tests/cap_test.c | 427 |
24 files changed, 3523 insertions, 1078 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; } diff --git a/src/lib/cap.c b/src/lib/cap.c new file mode 100644 index 00000000..f116bfb0 --- /dev/null +++ b/src/lib/cap.c @@ -0,0 +1,187 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Link capacity estimation + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library 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 Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +/* + * 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 transmit + * queue toward an n-1 flow (the flow to the layer below) while that + * queue is backlogged. + * + * Sampling is lock-free and off the fast path: the queue depth is + * read only at enqueue time, concurrently by many sender threads. + * Each enqueue bumps relaxed counters (packets, bytes, empty-queue + * hits). At most once per CAP_T_MIN, one thread wins a try-lock and + * closes a measurement window. + * + * Over a window, byte conservation gives the bytes that drained: + * drained = queue at start (q0) + enqueued - queue now (q1) + * A window stays open until CAP_N_MIN packets' worth has 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. + */ + +#if defined(__linux__) || defined(__CYGWIN__) +#ifndef _DEFAULT_SOURCE +#define _DEFAULT_SOURCE +#endif +#else +#ifndef _POSIX_C_SOURCE +#define _POSIX_C_SOURCE 200809L +#endif +#endif + +#include "config.h" + +#include <ouroboros/atomics.h> +#include <ouroboros/time.h> + +#include "cap.h" + +#include <string.h> + +#define CAP_T_MIN (BILLION / 1000) /* min close spacing ~1 ms */ +#define CAP_T_MAX (1ULL << 27) /* voiding traffic gap ~134 ms */ +#define CAP_N_MIN 16 /* drained packets to close */ +#define CAP_DEC_SHFT 4 /* max-filter decay 1/16 */ +#define CAP_IDL_SHFT 3 /* idle tolerance 1/8 */ + +/* Busy-flag try-lock: test-and-set acquire, store release. */ +#define CAP_TAS(p) __atomic_exchange_n(p, 1, __ATOMIC_ACQUIRE) +#define CAP_REL(p) (__atomic_store_n(p, 0, __ATOMIC_RELEASE)) + +void cap_clear(struct cap_est * e) +{ + memset(e, 0, sizeof(*e)); +} + +uint64_t cap_rate(const struct cap_est * e) +{ + return LOAD_RELAXED(&e->est); +} + +/* Busy flag held; q1 is the caller's pre-write ring sample. */ +static void cap_close(struct cap_est * e, + uint64_t q1, + uint64_t now, + uint64_t gap) +{ + 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; /* bytes 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 + (byt - e->byt0) - q1); + + if (e->t0 == 0 || enq == 0) + goto reopen; + + if (gap > CAP_T_MAX) + goto reopen; /* traffic stopped: window void */ + + avg = (byt - e->byt0) / enq; + if (drained < (int64_t) (CAP_N_MIN * avg)) + return; /* extend the window until enough drains */ + + if ((idl - e->idl0) << CAP_IDL_SHFT > enq) + goto reopen; /* mostly idle ring: not saturated */ + + r = (uint64_t) drained * MILLION / (dt / 1000); + if (r >= e->rate) { + if (e->q0 > 0 && q1 > 0) /* empty edge drains below */ + e->rate = r; + } else { + e->rate -= (e->rate - r) >> CAP_DEC_SHFT; + } + + STORE_RELAXED(&e->est, e->rate); + reopen: + e->t0 = now; + e->q0 = q1; + e->pkt0 = pkt; + e->byt0 = byt; + e->idl0 = idl; +} + +void cap_update_at(struct cap_est * e, + size_t qlen, + size_t len, + uint64_t now) +{ + uint64_t prev; + + FETCH_ADD_RELAXED(&e->c_pkt, 1); + FETCH_ADD_RELAXED(&e->c_byt, len); + + if (qlen == 0) + FETCH_ADD_RELAXED(&e->c_idl, 1); + + prev = LOAD_RELAXED(&e->t_last); + if (prev > now) + prev = now; /* a racing writer stamped ahead */ + + STORE_RELAXED(&e->t_last, now); + + if (now - LOAD_RELAXED(&e->t_gate) < CAP_T_MIN) + return; + + if (CAP_TAS(&e->busy) != 0) + return; + + if (now - e->t_gate >= CAP_T_MIN) { + cap_close(e, qlen, now, now - prev); + STORE_RELAXED(&e->t_gate, now); + } + + CAP_REL(&e->busy); +} + +void cap_update(struct cap_est * e, + size_t qlen, + size_t len) +{ + struct timespec now; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + cap_update_at(e, qlen, len, TS_TO_UINT64(now)); +} diff --git a/src/lib/cap.h b/src/lib/cap.h new file mode 100644 index 00000000..3d94d9a3 --- /dev/null +++ b/src/lib/cap.h @@ -0,0 +1,63 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Link capacity estimation + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library 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 Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#ifndef OUROBOROS_LIB_CAP_H +#define OUROBOROS_LIB_CAP_H + +#include <stddef.h> +#include <stdint.h> + +#define CAP_ALIGN 64 + +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 window close (ns) */ + uint64_t t_last; /* last update, to spot a gap (ns) */ + uint8_t busy; /* close 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) */ + + uint64_t est; /* published estimate (bytes/s) */ +} __attribute__((aligned(CAP_ALIGN))); + +void cap_clear(struct cap_est * e); + +void cap_update(struct cap_est * e, + size_t qlen, + size_t len); + +void cap_update_at(struct cap_est * e, + size_t qlen, + size_t len, + uint64_t now); + +uint64_t cap_rate(const struct cap_est * e); + +#endif /* OUROBOROS_LIB_CAP_H */ diff --git a/src/lib/config.h.in b/src/lib/config.h.in index 3c6985c3..62925a48 100644 --- a/src/lib/config.h.in +++ b/src/lib/config.h.in @@ -48,6 +48,7 @@ #cmakedefine HAVE_PMULL #define SHM_LOCKFILE_NAME "@SHM_LOCKFILE_NAME@" +#define SSM_RBUFF_TXQ_DELAY @SSM_RBUFF_TXQ_DELAY@ /* ms */ #define FLOW_ALLOC_TIMEOUT @FLOW_ALLOC_TIMEOUT@ #define TPM_DEBUG_REPORT_INTERVAL @TPM_DEBUG_REPORT_INTERVAL@ diff --git a/src/lib/dev.c b/src/lib/dev.c index eb706691..bce64c37 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -27,6 +27,7 @@ #endif #include "config.h" +#include "cap.h" #include "ssm.h" #include "poa/poa.h" @@ -60,6 +61,7 @@ #include <ouroboros/ssm_flow_set.h> #include <ouroboros/ssm_pool.h> #include <ouroboros/ssm_rbuff.h> +#include <ouroboros/time.h> #include <ouroboros/tw.h> #include <ouroboros/utils.h> @@ -85,6 +87,7 @@ #define DONE_PART -2 #define CRCLEN (sizeof(uint32_t)) +#define FLOW_AVG_SHIFT 3 #define SECMEMSZ 16384 #define MSGBUFSZ 2048 @@ -123,7 +126,13 @@ struct flow { struct frcti * frcti; + /* Mean written packet size (bytes), EWMA over the send path. */ + size_t mean_len; + struct poa_flow * poa; /* NULL for shared memory flows */ + + /* Egress capacity estimator; armed by the IPCP, else NULL. */ + struct cap_est * cap; }; struct flow_set { @@ -730,6 +739,8 @@ static void do_flow_fini(int fd) crypt_destroy_ctx(proc.flows[fd].crypt); + free(proc.flows[fd].cap); + flow_clear(fd); } @@ -768,6 +779,7 @@ static int flow_init(struct flow_info * info, struct poa_flow * pf) { struct timespec now; + struct timespec txq; struct flow * flow; int fd; int err = -ENOMEM; @@ -795,6 +807,11 @@ static int flow_init(struct flow_info * info, if (flow->tx_rb == NULL) goto fail_tx_rb; + txq.tv_sec = SSM_RBUFF_TXQ_DELAY / 1000; + txq.tv_nsec = (SSM_RBUFF_TXQ_DELAY % 1000) * MILLION; + + ssm_rbuff_set_txq_target(flow->tx_rb, &txq); + flow->set = ssm_flow_set_open(info->n_1_pid); if (flow->set == NULL) goto fail_set; @@ -1475,6 +1492,25 @@ int fccntl(int fd, goto einval; *maxp = flow_user_mtu(flow, flow->info.mtu); break; + case FLOWSTXQDLY: + timeo = va_arg(l, struct timespec *); + if (timeo == NULL) + goto einval; + + if (flow->tx_rb == NULL) + goto eperm; + + ssm_rbuff_set_txq_target(flow->tx_rb, timeo); + break; + case FLOWGTXQDLY: + timeo = va_arg(l, struct timespec *); + if (timeo == NULL) + goto einval; + + if (flow->tx_rb == NULL) + goto eperm; + ssm_rbuff_get_txq_target(flow->tx_rb, timeo); + break; case FLOWSFLAGS: old_acc = flow->oflags & FLOWFACCMODE; flow->oflags = va_arg(l, uint32_t); @@ -1609,6 +1645,25 @@ int fccntl(int fd, return -EPERM; } +/* + * The ring counts slots, so the queue is only bytes if we know what a + * packet weighs. Ordered so the unsigned arithmetic cannot wrap. + */ +static void flow_mean_len_update(struct flow * flow, + size_t len) +{ + size_t avg = LOAD_RELAXED(&flow->mean_len); + + if (avg == 0) { + STORE_RELAXED(&flow->mean_len, len); + return; + } + + avg = avg + (len >> FLOW_AVG_SHIFT) - (avg >> FLOW_AVG_SHIFT); + + STORE_RELAXED(&flow->mean_len, avg == 0 ? 1 : avg); +} + static int flow_tx_spb(struct flow * flow, struct ssm_pk_buff * spb, uint16_t flags, @@ -1643,6 +1698,8 @@ static int flow_tx_spb(struct flow * flow, if (flow->poa != NULL) return poa_flow_tx(flow->poa, spb, block, abstime); + flow_mean_len_update(flow, ssm_pk_buff_len(spb)); + if (!block) ret = ssm_rbuff_write(flow->tx_rb, idx); else @@ -2941,25 +2998,92 @@ size_t ipcp_flow_queued(int fd) if (proc.flows[fd].poa != NULL) return poa_flow_qlen(proc.flows[fd].poa); - return ssm_rbuff_queued(proc.flows[fd].tx_rb); + return ssm_rbuff_queued(proc.flows[fd].tx_rb) + * LOAD_RELAXED(&proc.flows[fd].mean_len); } -int ipcp_flow_queue_id(int fd) +size_t ipcp_flow_mean_len(int fd) { - int qid; + assert(fd >= 0 && fd < PROC_MAX_FLOWS); + assert(proc.flows[fd].info.id >= 0); + + if (proc.flows[fd].poa != NULL) + return poa_flow_mean_len(proc.flows[fd].poa); + + return LOAD_RELAXED(&proc.flows[fd].mean_len); +} + +/* An update racing the arm seeds one bogus window; the filter absorbs. */ +int ipcp_flow_cap_arm(int fd) +{ + struct flow * flow; + struct cap_est * e; + + assert(fd >= 0 && fd < PROC_MAX_FLOWS); + assert(proc.flows[fd].info.id >= 0); + + flow = &proc.flows[fd]; + if (flow->poa != NULL) { + cap_clear(poa_flow_cap_est(flow->poa)); + return 0; + } + + e = flow->cap; + if (e != NULL) { + cap_clear(e); + return 0; + } + + if (posix_memalign((void **) &e, CAP_ALIGN, sizeof(*e)) != 0) + return -ENOMEM; + + cap_clear(e); + + STORE_RELEASE(&flow->cap, e); + + return 0; +} + +void ipcp_flow_cap_update(int fd, + size_t qlen, + size_t len) +{ + struct flow * flow; + struct cap_est * e; assert(fd >= 0 && fd < PROC_MAX_FLOWS); assert(proc.flows[fd].info.id >= 0); - if (proc.flows[fd].poa == NULL) - return fd; + flow = &proc.flows[fd]; + if (flow->poa != NULL) { + cap_update(poa_flow_cap_est(flow->poa), qlen, len); + return; + } + + e = LOAD_ACQUIRE(&flow->cap); + if (e == NULL) + return; - /* An unidentified PoA answers for itself, never for an fd. */ - qid = poa_flow_qid(proc.flows[fd].poa); - if (qid < 0 || qid >= POA_MAX_POAS) - return fd; + cap_update(e, qlen, len); +} + +uint64_t ipcp_flow_cap(int fd) +{ + struct flow * flow; + struct cap_est * e; + + assert(fd >= 0 && fd < PROC_MAX_FLOWS); + assert(proc.flows[fd].info.id >= 0); + + flow = &proc.flows[fd]; + if (flow->poa != NULL) + return cap_rate(poa_flow_cap_est(flow->poa)); + + e = LOAD_ACQUIRE(&flow->cap); + if (e == NULL) + return 0; - return PROC_MAX_FLOWS + qid; + return cap_rate(e); } int local_flow_transfer(int src_fd, @@ -3028,4 +3152,5 @@ int local_flow_transfer(int src_fd, return ret; } +#include "cap.c" #include "poa/poa.c" diff --git a/src/lib/poa/poa.c b/src/lib/poa/poa.c index 3ad17c4f..b40d9fea 100644 --- a/src/lib/poa/poa.c +++ b/src/lib/poa/poa.c @@ -1678,9 +1678,9 @@ size_t poa_flow_mean_len(const struct poa_flow * pf) return LOAD_RELAXED(&pf->poa->avg_len); } -int poa_flow_qid(const struct poa_flow * pf) +struct cap_est * poa_flow_cap_est(struct poa_flow * pf) { - return pf->poa->qid; + return &pf->poa->cap; } void poa_flow_ready(struct poa_flow * pf) @@ -1830,7 +1830,6 @@ static struct poa * poa_create(enum poa_type type, poa->ops = ops; poa->mpl = ops->mpl; poa->n_eids = n_eids; - poa->qid = -1; return poa; @@ -2213,30 +2212,15 @@ void poa_fini(void) pthread_mutex_destroy(&poas.mtx); } -/* - * Lowest queue id no attached PoA holds; detaching frees it by - * leaving the list. Caller holds poas.lock. - */ -static int poa_qid_alloc(void) +static size_t poa_count(void) { struct list_head * p; - bool used[POA_MAX_POAS]; - int i; - - memset(used, 0, sizeof(used)); - - list_for_each(p, &poas.list) { - struct poa * poa = list_entry(p, struct poa, next); - - if (poa->qid >= 0 && poa->qid < POA_MAX_POAS) - used[poa->qid] = true; - } + size_t n = 0; - for (i = 0; i < POA_MAX_POAS; i++) - if (!used[i]) - return i; + list_for_each(p, &poas.list) + n++; - return -1; + return n; } static int poa_add(const struct poa_spec * spec, @@ -2275,8 +2259,7 @@ static int poa_add(const struct poa_spec * spec, pthread_rwlock_wrlock(&poas.lock); - poa->qid = poa_qid_alloc(); - if (poa->qid < 0) { + if (poa_count() >= POA_MAX_POAS) { pthread_rwlock_unlock(&poas.lock); goto fail_start; } diff --git a/src/lib/poa/poa.h b/src/lib/poa/poa.h index 014986a3..9edb0335 100644 --- a/src/lib/poa/poa.h +++ b/src/lib/poa/poa.h @@ -34,6 +34,8 @@ #include <ouroboros/time.h> #include <ouroboros/utils.h> +#include "../cap.h" + #include <errno.h> #include <limits.h> #include <poll.h> @@ -191,9 +193,6 @@ struct poa { time_t mpl; - /* Identifies the transmit queue the flows on this PoA share. */ - int qid; - /* Mean sent packet size (bytes), EWMA over the send path. */ size_t avg_len; /* Cost of one packet in the queue, in the transport's terms. */ @@ -204,6 +203,9 @@ struct poa { size_t q_cache; uint64_t q_time; + /* Capacity estimator of the queue the flows on this PoA share. */ + struct cap_est cap; + /* Queued management frames, capped; poas.mgmt_mtx guards. */ size_t n_mgmt; @@ -245,7 +247,7 @@ size_t poa_flow_qlen(const struct poa_flow * pf); size_t poa_flow_qpkts(const struct poa_flow * pf); -int poa_flow_qid(const struct poa_flow * pf); +struct cap_est * poa_flow_cap_est(struct poa_flow * pf); size_t poa_flow_mean_len(const struct poa_flow * pf); diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c index 04978d82..e35a27a9 100644 --- a/src/lib/ssm/rbuff.c +++ b/src/lib/ssm/rbuff.c @@ -57,6 +57,8 @@ #define LOAD_ACQUIRE(ptr) (__atomic_load_n(ptr, __ATOMIC_ACQUIRE)) #define STORE_RELEASE(ptr, val) \ (__atomic_store_n(ptr, val, __ATOMIC_RELEASE)) +#define STORE_RELAXED(ptr, val) \ + (__atomic_store_n(ptr, val, __ATOMIC_RELAXED)) #define HEAD(rb) (rb->shm_base[LOAD_RELAXED(rb->head)]) #define TAIL(rb) (rb->shm_base[LOAD_RELAXED(rb->tail)]) @@ -70,6 +72,20 @@ #define IS_FULL(rb) (QUEUED(rb) == (SSM_RBUFF_SIZE - 1)) #define IS_EMPTY(rb) (HEAD_IDX(rb) == TAIL_IDX(rb)) +/* + * Occupancy limiter: bound a tx ring by queueing delay instead of + * slot count, so a slow link does not accumulate seconds of backlog. + * A zero target is unlimited: the wait predicate then reduces to + * physical fullness. A ring is unlimited until a target is set. + */ +#define TXQ_MIN_SLOTS 4 /* floor: jitter margin */ +#define TXQ_PRIO_MUL 2 /* headroom kept for retx */ +#define TXQ_SHIFT 2 /* EWMA weight 1/4 */ +#define TXQ_SAMPLE_MASK 15 /* resample every 16 writes */ +#define TXQ_MIN_DT_NS 10000LL /* skip sub-10us samples */ + +#define TXQ_UNLIMITED (SSM_RBUFF_SIZE - 1) + struct ssm_rbuff { ssize_t * shm_base; /* start of shared memory */ size_t * head; /* start of ringbuffer */ @@ -81,8 +97,20 @@ struct ssm_rbuff { pid_t pid; /* pid of the owner */ int flow_id; /* flow_id of the flow */ size_t n_users; /* in-flight users */ + uint64_t txq_target; /* target queue delay, ns */ + size_t txq_limit; /* current occupancy limit */ + int64_t txq_rate; /* EWMA drain rate, slots/s */ + uint64_t txq_ns; /* last sample time, ns */ + size_t txq_wr; /* writes since last sample */ + size_t txq_q0; /* queued count at sample */ }; +#define TXQ_ON(rb) (LOAD_RELAXED(&(rb)->txq_target) != 0) +#define TXQ_LIMIT(rb) (TXQ_ON(rb) ? LOAD_RELAXED(&(rb)->txq_limit) \ + : TXQ_UNLIMITED) +#define OVER_LIMIT(rb) (QUEUED(rb) >= TXQ_LIMIT(rb)) + + #define MM_FLAGS (PROT_READ | PROT_WRITE) static struct ssm_rbuff * rbuff_create(pid_t pid, @@ -121,6 +149,12 @@ static struct ssm_rbuff * rbuff_create(pid_t pid, rb->pid = pid; rb->flow_id = flow_id; rb->n_users = 0; + rb->txq_target = 0; /* unlimited until set */ + rb->txq_limit = TXQ_UNLIMITED; + rb->txq_rate = 0; + rb->txq_ns = 0; + rb->txq_wr = 0; + rb->txq_q0 = 0; return rb; @@ -251,8 +285,106 @@ static void __cleanup_rbuff_reader(void * o) __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); } -int ssm_rbuff_write(struct ssm_rbuff * rb, - size_t off) +/* + * Refresh the drain-rate estimate and derived occupancy limit. + * Called with rb->mtx held, at most once per TXQ_SAMPLE_MASK writes. + */ +static void rbuff_txq_sample(struct ssm_rbuff * rb, + size_t queued) +{ + struct timespec now; + uint64_t now_ns; + uint64_t last_ns; + int64_t dt_ns; + int64_t written; + int64_t grown; + int64_t drained; + int64_t sample_rate; + int64_t rate; + int64_t target; + size_t limit; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + now_ns = TS_TO_UINT64(now); + + last_ns = LOAD_RELAXED(&rb->txq_ns); + if (last_ns == 0) { + /* No prior sample: seed and stay at the default limit. */ + STORE_RELAXED(&rb->txq_ns, now_ns); + STORE_RELAXED(&rb->txq_q0, queued); + STORE_RELAXED(&rb->txq_wr, 0); + return; + } + + dt_ns = (int64_t) (now_ns - last_ns); + if (dt_ns < TXQ_MIN_DT_NS) + return; + + written = (int64_t) LOAD_RELAXED(&rb->txq_wr); + grown = (int64_t) queued - (int64_t) LOAD_RELAXED(&rb->txq_q0); + + drained = written - grown; + if (drained < 0) + drained = 0; + + sample_rate = drained * BILLION / dt_ns; + + rate = LOAD_RELAXED(&rb->txq_rate); + rate += (sample_rate - rate) >> TXQ_SHIFT; + if (rate < 0) + rate = 0; + + target = (int64_t) LOAD_RELAXED(&rb->txq_target); + + limit = (size_t) (rate * target / BILLION); + if (limit < TXQ_MIN_SLOTS) + limit = TXQ_MIN_SLOTS; + + if (limit > TXQ_UNLIMITED) + limit = TXQ_UNLIMITED; + + STORE_RELAXED(&rb->txq_rate, rate); + STORE_RELAXED(&rb->txq_limit, limit); + STORE_RELAXED(&rb->txq_ns, now_ns); + STORE_RELAXED(&rb->txq_q0, queued); + STORE_RELAXED(&rb->txq_wr, 0); +} + +/* Bumps the write counter, resampling every TXQ_SAMPLE_MASK writes. */ +static void rbuff_txq_touch(struct ssm_rbuff * rb) +{ + size_t wr; + + wr = LOAD_RELAXED(&rb->txq_wr) + 1; + + STORE_RELAXED(&rb->txq_wr, wr); + + if ((wr & TXQ_SAMPLE_MASK) == 0) + rbuff_txq_sample(rb, QUEUED(rb)); +} + +/* + * A retransmission outranks new data but stays bounded: its ceiling is + * a multiple of the limit, so the headroom above it is reserved and the + * queueing delay stays within a known factor of the target. + */ +static size_t rbuff_txq_prio_limit(struct ssm_rbuff * rb) +{ + size_t lim; + + if (!TXQ_ON(rb)) + return TXQ_UNLIMITED; + + lim = LOAD_RELAXED(&rb->txq_limit) * TXQ_PRIO_MUL; + + return lim > TXQ_UNLIMITED ? TXQ_UNLIMITED : lim; +} + +/* prio outranks new data up to its own, higher, ceiling. */ +static int rbuff_write_nb(struct ssm_rbuff * rb, + size_t off, + bool prio) { size_t flags; bool was_empty; @@ -276,7 +408,8 @@ int ssm_rbuff_write(struct ssm_rbuff * rb, robust_mutex_lock(rb->mtx); - if (IS_FULL(rb)) { + if (QUEUED(rb) >= (prio ? rbuff_txq_prio_limit(rb) + : TXQ_LIMIT(rb))) { ret = -EAGAIN; goto fail_mutex; } @@ -289,6 +422,10 @@ int ssm_rbuff_write(struct ssm_rbuff * rb, if (was_empty) pthread_cond_broadcast(rb->add); + /* Only an enqueue feeds the estimator; a refusal wrote nothing. */ + if (TXQ_ON(rb)) + rbuff_txq_touch(rb); + pthread_mutex_unlock(rb->mtx); __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); @@ -301,6 +438,19 @@ int ssm_rbuff_write(struct ssm_rbuff * rb, return ret; } +int ssm_rbuff_write(struct ssm_rbuff * rb, + size_t off) +{ + return rbuff_write_nb(rb, off, false); +} + +/* For a packet the peer is already waiting on; skips the limit. */ +int ssm_rbuff_write_prio(struct ssm_rbuff * rb, + size_t off) +{ + return rbuff_write_nb(rb, off, true); +} + int ssm_rbuff_write_b(struct ssm_rbuff * rb, size_t off, const struct timespec * abstime) @@ -329,7 +479,7 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb, pthread_cleanup_push(__cleanup_rbuff_reader, rb); - while (IS_FULL(rb) && ret != -ETIMEDOUT) { + while (OVER_LIMIT(rb) && ret != -ETIMEDOUT) { flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); if (flags & RB_FLOWDOWN) { ret = -EFLOWDOWN; @@ -346,6 +496,9 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb, ADVANCE_HEAD(rb); if (was_empty) pthread_cond_broadcast(rb->add); + + if (TXQ_ON(rb)) + rbuff_txq_touch(rb); } pthread_mutex_unlock(rb->mtx); @@ -484,6 +637,40 @@ uint32_t ssm_rbuff_get_flags(struct ssm_rbuff * rb) return (uint32_t) __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); } +/* Current occupancy limit; SSM_RBUFF_SIZE - 1 when unlimited. */ +size_t ssm_rbuff_get_limit(struct ssm_rbuff * rb) +{ + assert(rb != NULL); + + return TXQ_LIMIT(rb); +} + +/* Target queueing delay; a zero target is unlimited. */ +void ssm_rbuff_set_txq_target(struct ssm_rbuff * rb, + const struct timespec * ts) +{ + assert(rb != NULL); + assert(ts != NULL); + + STORE_RELAXED(&rb->txq_limit, TXQ_UNLIMITED); + STORE_RELAXED(&rb->txq_rate, 0); + STORE_RELAXED(&rb->txq_ns, 0); + STORE_RELAXED(&rb->txq_wr, 0); + STORE_RELAXED(&rb->txq_q0, 0); + + STORE_RELAXED(&rb->txq_target, TS_TO_UINT64(*ts)); +} + +/* Current target queueing delay for the tx occupancy limiter. */ +void ssm_rbuff_get_txq_target(struct ssm_rbuff * rb, + struct timespec * ts) +{ + assert(rb != NULL); + assert(ts != NULL); + + UINT64_TO_TS(LOAD_RELAXED(&rb->txq_target), ts); +} + void ssm_rbuff_fini(struct ssm_rbuff * rb) { assert(rb != NULL); diff --git a/src/lib/ssm/tests/rbuff_test.c b/src/lib/ssm/tests/rbuff_test.c index 48e5a714..57e6198e 100644 --- a/src/lib/ssm/tests/rbuff_test.c +++ b/src/lib/ssm/tests/rbuff_test.c @@ -34,6 +34,9 @@ #include <ouroboros/errno.h> #include <ouroboros/time.h> +/* Mirrors TXQ_MIN_SLOTS in ssm/rbuff.c; keep in sync. */ +#define FLOOR_SLOTS 4 + #include <errno.h> #include <stdio.h> #include <unistd.h> @@ -652,6 +655,389 @@ static int test_ssm_rbuff_threaded(void) return TEST_RC_FAIL; } +static int test_ssm_rbuff_limit_off(void) +{ + struct ssm_rbuff * rb; + size_t i; + + TEST_START(); + + rb = ssm_rbuff_create(getpid(), 11); + if (rb == NULL) { + printf("Failed to create rbuff.\n"); + goto fail; + } + + if (ssm_rbuff_get_limit(rb) != SSM_RBUFF_SIZE - 1) { + printf("Expected default limit %d, got %zu.\n", + SSM_RBUFF_SIZE - 1, ssm_rbuff_get_limit(rb)); + goto fail_rb; + } + + for (i = 0; i < SSM_RBUFF_SIZE - 1; ++i) { + if (ssm_rbuff_write(rb, i) < 0) { + printf("Failed to write at index %zu.\n", i); + goto fail_rb; + } + } + + if (ssm_rbuff_write(rb, 999) != -EAGAIN) { + printf("Expected -EAGAIN on physically full buffer.\n"); + goto fail_rb; + } + + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + + fail_rb: + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_ssm_rbuff_limit_slow(void) +{ + struct ssm_rbuff * rb; + struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec delay = {0, 10 * MILLION}; + size_t limit; + size_t i; + + TEST_START(); + + rb = ssm_rbuff_create(getpid(), 12); + if (rb == NULL) { + printf("Failed to create rbuff.\n"); + goto fail; + } + + ssm_rbuff_set_txq_target(rb, &dfl); + + for (i = 0; i < 32; ++i) { + if (ssm_rbuff_write_b(rb, i, NULL) < 0) { + printf("Failed to write at index %zu.\n", i); + goto fail_rb; + } + nanosleep(&delay, NULL); + + if (ssm_rbuff_read(rb) < 0) { + printf("Failed to read at index %zu.\n", i); + goto fail_rb; + } + } + + limit = ssm_rbuff_get_limit(rb); + if (limit > FLOOR_SLOTS) { + printf("Expected limit near the floor, got %zu.\n", limit); + goto fail_rb; + } + + ssm_rbuff_destroy(rb); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + + fail_rb: + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_ssm_rbuff_limit_fast(void) +{ + struct ssm_rbuff * rb; + struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + size_t limit; + size_t i; + + TEST_START(); + + rb = ssm_rbuff_create(getpid(), 13); + if (rb == NULL) { + printf("Failed to create rbuff.\n"); + goto fail; + } + + ssm_rbuff_set_txq_target(rb, &dfl); + + for (i = 0; i < 200; ++i) { + if (ssm_rbuff_write_b(rb, i, NULL) < 0) { + printf("Failed to write at index %zu.\n", i); + goto fail_rb; + } + + if (ssm_rbuff_read(rb) < 0) { + printf("Failed to read at index %zu.\n", i); + goto fail_rb; + } + } + + limit = ssm_rbuff_get_limit(rb); + if (limit != SSM_RBUFF_SIZE - 1) { + printf("Expected limit %d, got %zu.\n", + SSM_RBUFF_SIZE - 1, limit); + goto fail_rb; + } + + ssm_rbuff_destroy(rb); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + + fail_rb: + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_ssm_rbuff_limit_floor(void) +{ + struct ssm_rbuff * rb; + struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec interval = {0, 50 * MILLION}; + struct timespec now; + struct timespec abs_timeout; + size_t limit; + int ret = 0; + size_t i; + + TEST_START(); + + rb = ssm_rbuff_create(getpid(), 14); + if (rb == NULL) { + printf("Failed to create rbuff.\n"); + goto fail; + } + + ssm_rbuff_set_txq_target(rb, &dfl); + + clock_gettime(PTHREAD_COND_CLOCK, &now); + ts_add(&now, &interval, &abs_timeout); + + for (i = 0; i < SSM_RBUFF_SIZE; ++i) { + ret = ssm_rbuff_write_b(rb, i, &abs_timeout); + if (ret == -ETIMEDOUT) + break; + + if (ret < 0) { + printf("Write failed at index %zu: %d.\n", i, ret); + goto fail_rb; + } + } + + if (ret != -ETIMEDOUT) { + printf("Expected the limiter to block the ring.\n"); + goto fail_rb; + } + + limit = ssm_rbuff_get_limit(rb); + if (limit > FLOOR_SLOTS) { + printf("Expected floor limit, got %zu.\n", limit); + goto fail_rb; + } + + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + + fail_rb: + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_ssm_rbuff_txq_target(void) +{ + struct ssm_rbuff * rb; + struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec delay = {0, 5 * MILLION}; + struct timespec small = {0, 2 * MILLION}; + struct timespec big = {0, 200 * MILLION}; + struct timespec def; + struct timespec got; + size_t limit_small; + size_t limit_big; + size_t i; + + TEST_START(); + + rb = ssm_rbuff_create(getpid(), 15); + if (rb == NULL) { + printf("Failed to create rbuff.\n"); + goto fail; + } + + /* A fresh ring is unlimited; rx rings must not inherit a bound. */ + ssm_rbuff_get_txq_target(rb, &got); + if (got.tv_sec != 0 || got.tv_nsec != 0) { + printf("A new ring is not unlimited.\n"); + goto fail_rb; + } + + ssm_rbuff_set_txq_target(rb, &dfl); + ssm_rbuff_get_txq_target(rb, &def); + + ssm_rbuff_set_txq_target(rb, &small); + + for (i = 0; i < 64; ++i) { + if (ssm_rbuff_write_b(rb, i, NULL) < 0) { + printf("Failed to write at index %zu.\n", i); + goto fail_rb; + } + nanosleep(&delay, NULL); + + if (ssm_rbuff_read(rb) < 0) { + printf("Failed to read at index %zu.\n", i); + goto fail_rb; + } + } + + limit_small = ssm_rbuff_get_limit(rb); + + ssm_rbuff_set_txq_target(rb, &big); + + for (i = 0; i < 64; ++i) { + if (ssm_rbuff_write_b(rb, i, NULL) < 0) { + printf("Failed to write at index %zu.\n", i); + goto fail_rb; + } + nanosleep(&delay, NULL); + + if (ssm_rbuff_read(rb) < 0) { + printf("Failed to read at index %zu.\n", i); + goto fail_rb; + } + } + + limit_big = ssm_rbuff_get_limit(rb); + if (limit_big <= limit_small) { + printf("Expected a larger target to grow the limit: " + "%zu -> %zu.\n", limit_small, limit_big); + goto fail_rb; + } + + ssm_rbuff_set_txq_target(rb, &dfl); + ssm_rbuff_get_txq_target(rb, &got); + + if (got.tv_sec != def.tv_sec || got.tv_nsec != def.tv_nsec) { + printf("NULL did not restore the default target.\n"); + goto fail_rb; + } + + ssm_rbuff_destroy(rb); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + + fail_rb: + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_ssm_rbuff_write_over_limit(void) +{ + struct ssm_rbuff * rb; + struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec age = {0, 20 * 1000}; + size_t count; + int ret = 0; + + TEST_START(); + + rb = ssm_rbuff_create(getpid(), 16); + if (rb == NULL) { + printf("Failed to create rbuff.\n"); + goto fail; + } + + ssm_rbuff_set_txq_target(rb, &dfl); + + for (count = 0; count < SSM_RBUFF_SIZE; ++count) { + ret = ssm_rbuff_write(rb, count); + if (ret == -EAGAIN) + break; + + if (ret < 0) { + printf("Write failed at index %zu: %d.\n", count, ret); + goto fail_rb; + } + + /* Age the seed sample past the estimator's dt floor. */ + if (count == 16) + nanosleep(&age, NULL); + } + + if (ret != -EAGAIN) { + printf("Expected the limiter to reject a write.\n"); + goto fail_rb; + } + + if (count >= SSM_RBUFF_SIZE / 2) { + printf("Expected -EAGAIN well before a full ring, " + "got %zu writes.\n", count); + goto fail_rb; + } + + if (ssm_rbuff_queued(rb) != count) { + printf("Queued %zu does not match write count %zu.\n", + ssm_rbuff_queued(rb), count); + goto fail_rb; + } + + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + + fail_rb: + while (ssm_rbuff_read(rb) >= 0) + ; + + ssm_rbuff_destroy(rb); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + int rbuff_test(int argc, char ** argv) { @@ -670,6 +1056,12 @@ int rbuff_test(int argc, ret |= test_ssm_rbuff_blocking(); ret |= test_ssm_rbuff_blocking_timeout(); ret |= test_ssm_rbuff_blocking_flowdown(); + ret |= test_ssm_rbuff_limit_off(); + ret |= test_ssm_rbuff_limit_slow(); + ret |= test_ssm_rbuff_limit_fast(); + ret |= test_ssm_rbuff_limit_floor(); + ret |= test_ssm_rbuff_txq_target(); + ret |= test_ssm_rbuff_write_over_limit(); return ret; } diff --git a/src/lib/tests/CMakeLists.txt b/src/lib/tests/CMakeLists.txt index 1f2e9ba2..d470d539 100644 --- a/src/lib/tests/CMakeLists.txt +++ b/src/lib/tests/CMakeLists.txt @@ -10,6 +10,7 @@ create_test_sourcelist(${PARENT_DIR}_tests test_suite.c auth_test_slh_dsa.c bitmap_test.c btree_test.c + cap_test.c crypt_test.c poa_test.c hash_test.c diff --git a/src/lib/tests/cap_test.c b/src/lib/tests/cap_test.c new file mode 100644 index 00000000..ea0e1fef --- /dev/null +++ b/src/lib/tests/cap_test.c @@ -0,0 +1,427 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Unit tests for link capacity estimation + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library 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 Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#include "../cap.c" + +#include <test/test.h> + +#include <inttypes.h> +#include <stdbool.h> + +#define TICK (50 * 1000ULL) /* 50 us between packets */ +#define LEN 1000ULL /* default packet size (B) */ +#define QLEN (8 * LEN) /* steady backlog (bytes) */ +#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)) + +/* Draining CAP_N_MIN of these outlasts CAP_T_MAX without a gap. */ +#define LOW_STEP (250 * TICK) /* 12.5 ms between packets */ +#define LOW_RATE (LEN * BILLION / LOW_STEP) + +/* Within the quarter-log2 band the wire code publishes. */ +static bool rate_is_near(uint64_t got, + uint64_t exp) +{ + return got >= exp - exp / 8 && got <= exp + exp / 8; +} + +static int test_cap_est_clear(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + if (cap_rate(&e) != 0) { + printf("Fresh estimator not unknown.\n"); + goto fail; + } + + for (i = 1; i <= 40; i++) + cap_update_at(&e, QLEN, LEN, i * TICK); + + if (cap_rate(&e) == 0) { + printf("No estimate to clear.\n"); + goto fail; + } + + cap_clear(&e); + + if (cap_rate(&e) != 0) { + printf("Clear did not drop the estimate.\n"); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* 1000 B every 50 us, ring steady at 8: drain = 20 MB/s. */ +static int test_cap_est_busy_window(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 40; i++) + cap_update_at(&e, QLEN, LEN, i * TICK); + + if (!rate_is_near(cap_rate(&e), RATE)) { + printf("Estimated rate: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) RATE, cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_cap_est_idle_tolerated(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 40; i++) + cap_update_at(&e, i == 21 ? 0 : QLEN, LEN, i * TICK); + + if (!rate_is_near(cap_rate(&e), RATE)) { + printf("Grazed window: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) RATE, cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_cap_est_mostly_idle_rejects(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 100; i++) + cap_update_at(&e, 0, LEN, i * TICK); + + if (cap_rate(&e) != 0) { + printf("Idle ring estimated %" PRIu64 ".\n", cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* 1000 B every 100 us: 10 slots/ms closes on a 2 ms window. */ +static int test_cap_est_slow_link_extends(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 30; i++) + cap_update_at(&e, QLEN, LEN, i * 2 * TICK); + + if (!rate_is_near(cap_rate(&e), RATE / 2)) { + printf("Slow link: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) (RATE / 2), cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* 1250 B every ms; one empty observation per 20 packets. */ +static int test_cap_est_shaped_link(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 100; i++) + cap_update_at(&e, i % SHP_STEP == 0 ? 0 : 6 * SHP_LEN, + SHP_LEN, i * SHP_STEP * TICK); + + if (!rate_is_near(cap_rate(&e), SHP_RATE)) { + printf("Shaped link: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) SHP_RATE, cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* Open a window, trickle 4 slots, then ~200 ms of silence. */ +static int test_cap_est_stale_discard(void) +{ + struct cap_est e; + uint64_t t; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 5; i++) + cap_update_at(&e, QLEN, LEN, i * CAP_T_MIN); + + t = 205 * CAP_T_MIN; + + cap_update_at(&e, QLEN, LEN, t); + + if (cap_rate(&e) != 0) { + printf("Gap window estimated %" PRIu64 ".\n", cap_rate(&e)); + goto fail; + } + + for (i = 1; i <= 40; i++) + cap_update_at(&e, QLEN, LEN, t + i * TICK); + + if (!rate_is_near(cap_rate(&e), RATE)) { + printf("Post-gap: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) RATE, cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_cap_est_empty_start_no_raise(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + cap_update_at(&e, 0, LEN, CAP_T_MIN); + + for (i = 1; i <= 40; i++) + cap_update_at(&e, QLEN, LEN, CAP_T_MIN + i * TICK); + + if (cap_rate(&e) != 0) { + printf("Empty-start window raised to %" PRIu64 ".\n", + cap_rate(&e)); + goto fail; + } + + for (i = 41; i <= 60; i++) + cap_update_at(&e, QLEN, LEN, CAP_T_MIN + i * TICK); + + if (!rate_is_near(cap_rate(&e), RATE)) { + printf("Backlogged window: exp %" PRIu64 ", got %" PRIu64 + ".\n", (uint64_t) RATE, cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * Max filter: fast attack on a high sample, slow release on the + * lower samples from a halved packet size (10 MB/s). + */ +static int test_cap_est_max_filter(void) +{ + struct cap_est e; + uint64_t high; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 40; i++) + cap_update_at(&e, QLEN, LEN, i * TICK); + + high = cap_rate(&e); + if (!rate_is_near(high, RATE)) { + printf("Attack missed: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) RATE, high); + goto fail; + } + + for (i = 41; i <= 80; i++) + cap_update_at(&e, QLEN, LEN / 2, i * TICK); + + if (cap_rate(&e) >= high) { + printf("Release did not decay: %" PRIu64 ".\n", cap_rate(&e)); + goto fail; + } + + if (cap_rate(&e) <= RATE / 2) { + printf("Release collapsed to %" PRIu64 ".\n", cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* No window close within CAP_T_MIN of the last one. */ +static int test_cap_est_gate(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + cap_update_at(&e, QLEN, LEN, CAP_T_MIN); + + for (i = 0; i < 5; i++) + cap_update_at(&e, QLEN, LEN, CAP_T_MIN + CAP_T_MIN / 2); + + if (e.t_gate != CAP_T_MIN) { + printf("Window closed inside the gate.\n"); + goto fail; + } + + if (LOAD_RELAXED(&e.c_pkt) != 6) { + printf("Gated packets not counted.\n"); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * A link slow enough that CAP_N_MIN packets take longer than + * CAP_T_MAX to drain still publishes, as long as the sender keeps + * offering: only silence voids a window. + */ +static int test_cap_est_low_rate_publishes(void) +{ + struct cap_est e; + size_t i; + + TEST_START(); + + cap_clear(&e); + + for (i = 1; i <= 20; i++) + cap_update_at(&e, QLEN, LEN, i * LOW_STEP); + + if (!rate_is_near(cap_rate(&e), LOW_RATE)) { + printf("Low rate: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) LOW_RATE, cap_rate(&e)); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +int cap_test(int argc, + char ** argv) +{ + int ret = 0; + + (void) argc; + (void) argv; + + ret |= test_cap_est_clear(); + 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_est_low_rate_publishes(); + + return ret; +} |