diff options
Diffstat (limited to 'src/ipcpd/unicast/ca')
| -rw-r--r-- | src/ipcpd/unicast/ca/mb-ecn.c | 238 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/mb-ecn.h | 9 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/ops.h | 8 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/tests/ca_test.c | 14 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/tests/mb_ecn_test.c | 357 |
5 files changed, 566 insertions, 60 deletions
diff --git a/src/ipcpd/unicast/ca/mb-ecn.c b/src/ipcpd/unicast/ca/mb-ecn.c index e59aac88..a4c9f29e 100644 --- a/src/ipcpd/unicast/ca/mb-ecn.c +++ b/src/ipcpd/unicast/ca/mb-ecn.c @@ -40,14 +40,24 @@ #include <stdio.h> /* - * The sender paces with a token bucket at a rate driven by Δt-scaled - * AIMD: each step changes the rate proportional to elapsed wall-clock - * time, so the per-second dynamics do not depend on how often packets - * arrive. 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). There - * is no per-flow timer; the control runs on packet sends. + * 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. + * + * 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). + * + * 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. * * The floor and the AI slope scale with the path: forwarders stamp * their measured link capacity into the PCI (cap.c), the receiver @@ -77,12 +87,17 @@ #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 */ -/* Must stay >= 1 ms: the MD term scales by dtc / MILLION (truncates). */ -#define CA_DT_CTRL (BILLION / 1000) +#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_SS_TC (BILLION / 50) +#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_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 */ @@ -126,31 +141,29 @@ struct mb_ecn_ctx { uint16_t tx_ece; /* congestion reported from downstream */ uint16_t tx_ecp; /* previous tx_ece (rise detection) */ uint8_t tx_loc; /* local first-hop ecn mark (fallback) */ - uint8_t tx_cap; /* path capacity code fed back to us */ bool tx_cav; /* past slow start */ + bool ai_hold; /* freeze AI after loss until clear */ uint64_t rate; /* paced send rate (bytes/s) */ uint64_t rate_min; /* capacity-derived rate floor (B/s) */ 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) */ uint64_t lead; /* pacer lead of last send (bytes) */ uint64_t last_ts; /* last clock advance (ns) */ uint64_t last_ctrl; /* last rate update (ns) */ - uint64_t last_fb; /* last feedback applied (ns) */ + 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 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 n_ctrl; /* control steps taken */ - uint64_t t_ctrl; /* wall time covered by steps (ns) */ - uint64_t t_bank; /* increase time banked in steps (ns) */ - uint64_t n_fb; /* feedback updates received */ - uint64_t n_ttl; /* feedback aged out (TTL) */ - uint64_t n_cap; /* capacity updates applied */ - uint64_t ss_peak; /* peak rate in slow start (bytes/s) */ - uint64_t snd_byt; /* bytes offered this window (capped) */ uint64_t snd_win; /* utilisation window start (ns) */ uint64_t snd_r0; /* rate at window start */ @@ -158,14 +171,31 @@ struct mb_ecn_ctx { bool backlogged; /* offered load keeps the pacer busy */ bool src_limited; /* rate held at offered-load ceiling */ bool started; /* a real send has occurred */ + + /* Diagnostics only, read by mb_ecn_print_stats. */ + uint8_t tx_cap; /* path capacity code fed back to us */ + uint64_t n_ctrl; /* control steps taken */ + uint64_t t_ctrl; /* wall time covered by steps (ns) */ + uint64_t t_bank; /* increase time banked in steps (ns) */ + 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) */ }; +/* Layer slow-start time constant (ns), from the declared RTT. */ +static uint64_t mb_ecn_ss_tc = (uint64_t) CA_SS_RTT_MUL * + CA_SS_RTT_DEF * MILLION; + struct ca_ops mb_ecn_ca_ops = { .ctx_create = mb_ecn_ctx_create, .ctx_destroy = mb_ecn_ctx_destroy, .ctx_update_snd = mb_ecn_ctx_update_snd, .ctx_update_rcv = mb_ecn_ctx_update_rcv, .ctx_update_ece = mb_ecn_ctx_update_ece, + .ctx_hb_due = mb_ecn_ctx_hb_due, + .ctx_rtt = mb_ecn_ctx_rtt, .calc_ecn = mb_ecn_calc_ecn, .marks_ecn = true, .print_stats = mb_ecn_print_stats @@ -190,6 +220,21 @@ static uint64_t mb_ecn_ece_ttl(uint64_t rate) return ttl > (uint64_t) CA_ECE_TTL ? ttl : (uint64_t) CA_ECE_TTL; } +/* Derive the layer slow-start slope from the declared RTT (ms). */ +void mb_ecn_init(uint32_t rtt_ms) +{ + uint64_t tc; + + if (rtt_ms == 0) /* unspecified: safe default */ + rtt_ms = CA_SS_RTT_DEF; + + tc = (uint64_t) CA_SS_RTT_MUL * rtt_ms * MILLION; + if (tc < (uint64_t) CA_SS_TC_MIN) + tc = CA_SS_TC_MIN; + + mb_ecn_ss_tc = tc; +} + void * mb_ecn_ctx_create(void) { struct timespec now; @@ -209,6 +254,7 @@ void * mb_ecn_ctx_create(void) ctx->rate = CA_RATE_INIT; ctx->rate_min = CA_RATE_MIN; 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); @@ -218,6 +264,7 @@ void * mb_ecn_ctx_create(void) ctx->last_ts = t; ctx->last_ctrl = t; ctx->last_fb = t; + ctx->last_sig = t; ctx->last_loc = t; ctx->last_cap = t; @@ -252,7 +299,7 @@ static void mb_ecn_slow_start(struct mb_ecn_ctx * ctx, uint64_t dta) { if (ctx->backlogged) - ctx->rate += ctx->rate * dta / CA_SS_TC; + ctx->rate += ctx->rate * dta / ctx->ss_tc; ctx->r_bkt = ctx->rate; } @@ -264,11 +311,18 @@ static void mb_ecn_increase(struct mb_ecn_ctx * ctx, if (!ctx->backlogged) return; + /* After a loss, hold until a clean signal drains the queue. */ + if (ctx->ai_hold) + return; + ctx->rate += ctx->ai_rate * dta / BILLION; ctx->rate += ctx->rate * dta / CA_PROBE_TC; } -/* Multiplicative decrease: proportional cut plus a one-sided lead. */ +/* + * Multiplicative decrease: cut proportional to mark x elapsed time, + * plus a one-sided lead that cuts extra while the mark is rising. + */ static void mb_ecn_decrease(struct mb_ecn_ctx * ctx, uint64_t dtc) { @@ -281,19 +335,21 @@ static void mb_ecn_decrease(struct mb_ecn_ctx * ctx, m = ctx->tx_ece > 0 ? ctx->tx_ece : (uint16_t) (ctx->tx_loc << CA_SHFT); if (m == 0) { - ctx->tx_ecp = 0; + ctx->dec_acc = 0; /* unmarked time is not banked */ + ctx->tx_ecp = 0; return; } mark = MIN(m, CA_ECE_REF); - rise = 0; - if (m > ctx->tx_ecp) - rise = MIN(m - ctx->tx_ecp, CA_ECE_REF); - cut = ctx->rate * rise / (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); - /* Honest elapsed ms, so a starved sender still cuts. */ - dtm = dtc / MILLION; + /* Honest elapsed ms; the sub-ms remainder carries over. */ + ctx->dec_acc += dtc; + dtm = ctx->dec_acc / MILLION; + ctx->dec_acc -= dtm * MILLION; if (mark * dtm >= CA_ECE_REF * 500) cut += ctx->rate / 2; else @@ -428,11 +484,15 @@ static void mb_ecn_win(struct mb_ecn_ctx * ctx, { uint64_t elapsed = t - ctx->snd_win; uint64_t offered; + bool was = ctx->backlogged; offered = ctx->snd_byt * BILLION / elapsed; ctx->backlogged = offered * CA_USE_DEN >= ctx->snd_r0 * CA_USE_NUM; + if (!was && ctx->backlogged) /* resume: fresh liveness baseline */ + ctx->last_res = t; + mb_ecn_offered(ctx, offered, elapsed); if (ctx->backlogged) @@ -444,9 +504,50 @@ static void mb_ecn_win(struct mb_ecn_ctx * ctx, } /* Age out congestion, local-mark and capacity signals once stale. */ +/* Heartbeat interval: ~1 RTT, floored so fast links don't over-probe. */ +static uint64_t mb_ecn_t_hb(const struct mb_ecn_ctx * ctx) +{ + uint64_t t = ctx->ss_tc >> 1; + + return t > (uint64_t) CA_HB_MIN ? t : CA_HB_MIN; +} + +/* Feedback collapsed while backlogged: halve like an RTO, stay in AIMD. */ +static void mb_ecn_loss(struct mb_ecn_ctx * ctx, + uint64_t t) +{ + ctx->rate -= ctx->rate / 2; + 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; + ctx->ai_hold = true; + ctx->n_loss++; +} + static void mb_ecn_age(struct mb_ecn_ctx * ctx, uint64_t t) { + uint64_t ttl = ctx->ece_ttl; + uint64_t ref = ctx->last_sig > ctx->last_res + ? ctx->last_sig : ctx->last_res; + uint64_t gap = t - ref; + + /* + * Sustained silence while backlogged is feedback collapse: cut + * the rate in half and stay in AIMD, so a recovering flow climbs + * back additively instead of re-ramping. Repeated silence decays + * it geometrically toward the floor. + */ + if (ctx->backlogged && ctx->n_fb + ctx->n_rtt > 0 + && gap > (uint64_t) CA_HB_LOSS * ttl) { + mb_ecn_loss(ctx, t); + return; + } + if (t - ctx->last_fb > ctx->ece_ttl) { if (ctx->tx_ece > 0) ctx->n_ttl++; @@ -505,15 +606,20 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx, /* Lazy warm-up seed: packet #1 is never an idle resume. */ if (!ctx->started) { - ctx->started = true; - ctx->last_ts = t; - ctx->snd_win = t; - ctx->snd_r0 = ctx->rate; + ctx->started = true; + ctx->last_ts = t; + ctx->last_res = t; + ctx->snd_win = t; + ctx->snd_r0 = ctx->rate; } 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) + ctx->backlogged = false; + mb_ecn_age(ctx, t); /* Offered-load estimator: accumulate, gate growth, size ceiling. */ @@ -521,9 +627,6 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx, if (ctx->snd_byt > (uint64_t) CA_SND_BYT_MAX) ctx->snd_byt = CA_SND_BYT_MAX; - if (dt > (uint64_t) CA_DT_CAP) - ctx->backlogged = false; - if (t - ctx->snd_win >= (uint64_t) CA_SND_WIN) mb_ecn_win(ctx, t); @@ -683,8 +786,14 @@ static void mb_ecn_ece(struct mb_ecn_ctx * ctx, uint64_t tgt; ctx->tx_ece = ece; - ctx->tx_cav = true; - ctx->last_fb = t; + 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) + ctx->ai_hold = false; + + ctx->last_fb = t; + ctx->last_sig = t; ctx->n_fb++; /* Scale the floor and AI slope to the path bottleneck. */ @@ -730,6 +839,44 @@ void mb_ecn_ctx_update_ece(void * _ctx, mb_ecn_ece(ctx, ece, cap, TS_TO_UINT64(now)); } +/* Due when the path stayed quiet for a heartbeat interval; arms the gap. */ +bool mb_ecn_ctx_hb_due(void * _ctx, + uint64_t now) +{ + struct mb_ecn_ctx * ctx = _ctx; + uint64_t t_hb = mb_ecn_t_hb(ctx); + uint64_t last; + + last = ctx->last_sig > ctx->last_hb ? ctx->last_sig : ctx->last_hb; + if (now - last < t_hb) + return false; + + ctx->last_hb = now; + + return true; +} + +/* Fold a heartbeat RTT sample into the ramp clock; also counts as life. */ +void mb_ecn_ctx_rtt(void * _ctx, + uint64_t now, + uint64_t rtt) +{ + struct mb_ecn_ctx * ctx = _ctx; + uint64_t tgt; + + tgt = (uint64_t) CA_SS_RTT_MUL * rtt; + if (tgt < (uint64_t) CA_SS_TC_MIN) /* track the true RTT both */ + tgt = CA_SS_TC_MIN; /* ways: overshoot ~e^{1/2} */ + + if (tgt > (uint64_t) CA_SS_TC_MAX) /* at the real RTT, not the */ + tgt = CA_SS_TC_MAX; /* declared worst case */ + + ctx->ss_tc += (tgt >> CA_RTT_SHFT) - (ctx->ss_tc >> CA_RTT_SHFT); + + ctx->last_sig = now; /* liveness only: never ages the ece signal */ + ctx->n_rtt++; +} + int mb_ecn_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, @@ -762,7 +909,7 @@ ssize_t mb_ecn_print_stats(void * _ctx, int code; uint16_t m; - if (len < 1024) + if (len < CA_STATS_STRLEN) return 0; /* No signal seen: the rate is unconstrained drift, not a target. */ @@ -776,6 +923,9 @@ ssize_t mb_ecn_print_stats(void * _ctx, if (!ctx->tx_cav) { regime = "Slow start"; code = 0; + } else if (ctx->ai_hold) { + regime = "Loss recovery"; + code = 4; } else if (ctx->src_limited) { regime = "Source limited"; code = 3; @@ -803,7 +953,10 @@ ssize_t mb_ecn_print_stats(void * _ctx, "Path capacity (bytes/s): %20" PRIu64 "\n" "Capacity rate floor (bytes/s): %20" PRIu64 "\n" "Capacity updates (count): %20" PRIu64 "\n" - "Slow start peak rate (bytes/s): %20" PRIu64 "\n", + "Slow start peak rate (bytes/s): %20" PRIu64 "\n" + "Signal-loss cuts (count): %20" PRIu64 "\n" + "Heartbeat RTT samples (count): %20" PRIu64 "\n" + "Ramp time constant (ns): %20" PRIu64 "\n", "Multi-bit ECN", ctx->tx_ece, ctx->rx_ece, @@ -812,7 +965,8 @@ ssize_t mb_ecn_print_stats(void * _ctx, ctx->n_ctrl, ctx->t_ctrl, ctx->t_bank, ctx->n_fb, ctx->n_ttl, cap_dec(ctx->tx_cap), ctx->rate_min, ctx->n_cap, - peak); + peak, + ctx->n_loss, ctx->n_rtt, ctx->ss_tc); return strlen(buf); } diff --git a/src/ipcpd/unicast/ca/mb-ecn.h b/src/ipcpd/unicast/ca/mb-ecn.h index 7bf0b29f..781e25a8 100644 --- a/src/ipcpd/unicast/ca/mb-ecn.h +++ b/src/ipcpd/unicast/ca/mb-ecn.h @@ -25,6 +25,8 @@ #include "ops.h" +void mb_ecn_init(uint32_t rtt_ms); + void * mb_ecn_ctx_create(void); void mb_ecn_ctx_destroy(void * ctx); @@ -45,6 +47,13 @@ void mb_ecn_ctx_update_ece(void * ctx, uint16_t ece, uint8_t cap); +bool mb_ecn_ctx_hb_due(void * ctx, + uint64_t now); + +void mb_ecn_ctx_rtt(void * ctx, + uint64_t now, + uint64_t rtt); + int mb_ecn_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, diff --git a/src/ipcpd/unicast/ca/ops.h b/src/ipcpd/unicast/ca/ops.h index a86c1b3b..b01e8573 100644 --- a/src/ipcpd/unicast/ca/ops.h +++ b/src/ipcpd/unicast/ca/ops.h @@ -46,6 +46,14 @@ struct ca_ops { uint16_t ece, uint8_t cap); + /* Optional, can be NULL: heartbeat pacing + RTT feedback. */ + bool (* ctx_hb_due)(void * ctx, + uint64_t now); + + void (* ctx_rtt)(void * ctx, + uint64_t now, + uint64_t rtt); + int (* calc_ecn)(size_t queued, uint8_t * ecn, qoscube_t qc, diff --git a/src/ipcpd/unicast/ca/tests/ca_test.c b/src/ipcpd/unicast/ca/tests/ca_test.c index 4c44d29e..1b86eab8 100644 --- a/src/ipcpd/unicast/ca/tests/ca_test.c +++ b/src/ipcpd/unicast/ca/tests/ca_test.c @@ -44,7 +44,7 @@ static int test_ca_init_fini(enum pol_cong_avoid pol, { TEST_START("(%s)", name); - if (ca_init(pol) < 0) { + if (ca_init(pol, 100) < 0) { printf("Failed to init ca for %s.\n", name); goto fail; } @@ -74,7 +74,7 @@ static int test_ca_init_invalid(void) { TEST_START(); - if (ca_init(CA_INVALID) == 0) { + if (ca_init(CA_INVALID, 100) == 0) { printf("Init accepted an invalid policy.\n"); ca_fini(); goto fail; @@ -96,7 +96,7 @@ static int test_ca_ctx_share(enum pol_cong_avoid pol, TEST_START("(%s)", name); - if (ca_init(pol) < 0) { + if (ca_init(pol, 100) < 0) { printf("Failed to init ca for %s.\n", name); goto fail; } @@ -162,7 +162,7 @@ static int test_ca_ctx_distinct(void) TEST_START(); - if (ca_init(CA_NONE) < 0) { + if (ca_init(CA_NONE, 100) < 0) { printf("Failed to init ca.\n"); goto fail; } @@ -226,7 +226,7 @@ static int test_ca_ctx_refcount(void) TEST_START(); - if (ca_init(CA_NONE) < 0) { + if (ca_init(CA_NONE, 100) < 0) { printf("Failed to init ca.\n"); goto fail; } @@ -283,7 +283,7 @@ static int test_ca_ctx_recreate(void) TEST_START(); - if (ca_init(CA_NONE) < 0) { + if (ca_init(CA_NONE, 100) < 0) { printf("Failed to init ca.\n"); goto fail; } @@ -340,7 +340,7 @@ static int test_ca_fini_drains(void) TEST_START(); - if (ca_init(CA_NONE) < 0) { + if (ca_init(CA_NONE, 100) < 0) { printf("Failed to init ca.\n"); goto fail; } diff --git a/src/ipcpd/unicast/ca/tests/mb_ecn_test.c b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c index 8e3a73da..4bbc12aa 100644 --- a/src/ipcpd/unicast/ca/tests/mb_ecn_test.c +++ b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c @@ -21,7 +21,6 @@ */ #include "mb-ecn.c" - #include <test/test.h> #define MS (MILLION) /* one millisecond in ns */ @@ -41,6 +40,7 @@ static struct mb_ecn_ctx * mk_ctx(void) ctx->last_ts = 0; ctx->last_ctrl = 0; ctx->last_fb = 0; + ctx->last_sig = 0; ctx->last_loc = 0; ctx->last_cap = 0; @@ -51,6 +51,7 @@ static struct mb_ecn_ctx * mk_ctx(void) ctx->backlogged = true; ctx->src_limited = false; ctx->started = false; + ctx->ss_tc = 20 * MS; /* fixed slope for deterministic SS */ return ctx; } @@ -815,7 +816,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) and a repeated mark must not re-fire the one-sided lead. + * 2 * CA_TW_INIT); a repeated mark adds only the proportional cut. */ static int test_mb_ecn_ece_ttl_covers_cadence(void) { @@ -1048,10 +1049,13 @@ static int test_mb_ecn_rate_floor(void) goto fail; } - /* Cut larger than headroom must clamp; stay inside CA_ECE_TTL. */ + /* + * A starved 600 ms at full mark takes the rate/2 branch, far + * below the floor; stays inside the initial ~976 ms mark TTL. + */ ctx->rate = CA_RATE_MIN + 1000; mb_ecn_ece(ctx, CA_ECE_REF, 0, 0); - mb_ecn_snd(ctx, LEN, 30 * MS, &ftag); + mb_ecn_snd(ctx, LEN, 600 * MS, &ftag); if (ctx->rate != CA_RATE_MIN) { printf("rate floor breached: %" PRIu64 ".\n", ctx->rate); @@ -1116,8 +1120,9 @@ static int test_mb_ecn_fixed_point(void) } /* - * One-sided lead: a mark rise cuts rate * rise / (REF * DEN) once; - * a flat or falling mark leaves only the dt-scaled proportional cut. + * 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) { @@ -1137,17 +1142,17 @@ static int test_mb_ecn_lead_cut(void) ctx->rate = (uint64_t) 100 << 20; ctx->tx_cav = true; - /* Rise 0 -> 256: lead cuts ~rise/(REF*DEN) = 1/8 of the rate. */ + /* 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 / 10) { - printf("lead cut missing: dropped %" PRIu64 ".\n", drop); + if (drop < prev / 16) { + printf("rising mark under-cut: %" PRIu64 ".\n", drop); goto fail_ctx; } - /* Flat mark: rise 0, only the ~0.05%% proportional cut. */ + /* 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); @@ -1157,7 +1162,7 @@ static int test_mb_ecn_lead_cut(void) goto fail_ctx; } - /* Falling mark: one-sided lead must not fire. */ + /* 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); @@ -1421,6 +1426,69 @@ static int test_mb_ecn_decrease_dt_invariant(void) return TEST_RC_FAIL; } +/* Sub-ms control steps must not lose decrease time to truncation. */ +static int test_mb_ecn_decrease_subms_carry(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint64_t cut_a; + uint64_t cut_b; + uint64_t r0; + uint64_t fta = 0; + uint64_t ftb = 0; + size_t i; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + r0 = (uint64_t) 100 << 20; + a->rate = r0; + b->rate = r0; + + mb_ecn_ece(a, 256, 0, 0); + mb_ecn_ece(b, 256, 0, 0); + + /* Same 30 ms of marked time; b's steps have a 0.5 ms tail. */ + for (i = 1; i <= 10; i++) + mb_ecn_snd(a, LEN, i * 3 * MS, &fta); + + for (i = 1; i <= 20; i++) + mb_ecn_snd(b, LEN, i * 3 * MS / 2, &ftb); + + cut_a = r0 - a->rate; + cut_b = r0 - b->rate; + + if (cut_a == 0 || cut_b == 0) { + printf("no cut: %" PRIu64 " %" PRIu64 ".\n", cut_a, cut_b); + goto fail_ctx; + } + + /* Within 10%: the sub-ms remainder must carry, not vanish. */ + if (cut_a > cut_b + cut_a / 10 || cut_b > cut_a + cut_a / 10) { + printf("sub-ms decrease lost: %" PRIu64 " vs %" PRIu64 + ".\n", cut_a, cut_b); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + /* Resuming after a long idle gap: bounded AI, no cut from stale marks. */ static int test_mb_ecn_idle_resume_bounded(void) { @@ -2528,6 +2596,269 @@ static int test_mb_ecn_first_send_warmup(void) return TEST_RC_FAIL; } +/* + * Couple one or two backlogged flows through a shared bottleneck of + * capacity cap: each step marks the shared queue with the real + * calc_ecn, feeds each flow that mark delayed by its own lag (in + * steps), drives it backlogged for one step, then drains the queue. + * A faithful discrete run of the fluid model on the real control law, + * with the forwarder abstracted to a single shared price. b may be + * NULL for a single-flow run. + */ +#define TF_STEP (5 * MS) /* control step (ns) */ +#define TF_HIST 64 /* mark ring, bounds max lag */ +#define TF_QMAX (8192 * LEN) /* bottleneck buffer (bytes) */ +#define TF_MAXN 32 /* flows per shared-link run */ + +static void shared_link_run(struct mb_ecn_ctx * a, + struct mb_ecn_ctx * b, + uint64_t cap, + size_t lag_a, + size_t lag_b, + size_t steps) +{ + uint16_t hist[TF_HIST]; + uint64_t ta = 0; + uint64_t tb = 0; + uint64_t fta = 0; + uint64_t ftb = 0; + uint64_t q = 0; + uint64_t drain = cap * TF_STEP / BILLION; + uint64_t tgt; + uint64_t arr; + uint8_t cc = cap_enc(cap); + uint8_t ecn; + uint16_t ea; + uint16_t eb; + size_t k; + + memset(hist, 0, sizeof(hist)); + + for (k = 0; k < steps; k++) { + tgt = (k + 1) * TF_STEP; + ecn = 0; + + mb_ecn_calc_ecn(q / LEN, &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]; + if (ea > 0) /* no feedback until congestion */ + mb_ecn_ece(a, ea, cc, ta); + + /* Active flow: heartbeat/window liveness stays fresh. */ + a->last_sig = ta; + if (tgt > ta) + ta = drive_backlogged(a, &fta, ta, tgt - ta, LEN); + + arr = a->rate * TF_STEP / BILLION; + + if (b != NULL) { + eb = k < lag_b ? 0 : hist[(k - lag_b) % TF_HIST]; + if (eb > 0) + mb_ecn_ece(b, eb, cc, tb); + + b->last_sig = tb; + if (tgt > tb) + tb = drive_backlogged(b, &ftb, tb, + tgt - tb, LEN); + + arr += b->rate * TF_STEP / BILLION; + } + + q += arr; + q = q > drain ? q - drain : 0; + if (q > (uint64_t) TF_QMAX) + q = TF_QMAX; + } +} + +/* + * Two flows sharing one bottleneck, both fed every queue mark, must + * converge from a lopsided start toward an equal split (the fluid + * model's drho/dt -> 0) rather than latch winner-take-all. Isolates + * the rate law from the forwarder: neither flow is starved of marks. + */ +static int test_mb_ecn_two_flow_converge(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint64_t cap = 1ULL << 20; + uint64_t fair = (1ULL << 20) / 2; + uint64_t lo; + uint64_t hi; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + a->rate = cap; /* a hogs, b starts small */ + b->rate = CA_RATE_INIT; + + shared_link_run(a, b, cap, 0, 0, 6000); + + lo = a->rate < b->rate ? a->rate : b->rate; + hi = a->rate > b->rate ? a->rate : b->rate; + + if (lo < fair / 4) { + printf("flow starved: %" PRIu64 " / %" PRIu64 ".\n", + a->rate, b->rate); + goto fail_ctx; + } + + if (hi > 3 * lo) { + printf("did not converge: %" PRIu64 " / %" PRIu64 ".\n", + a->rate, b->rate); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * A remote bottleneck exits slow start only on fed-back congestion, + * so a longer feedback delay lets the exponential ramp overshoot + * further: the slow-start peak grows with RTT. Nothing caps the ramp + * at the path capacity, so the sender latches near the first-hop rate. + */ +static int test_mb_ecn_ramp_overshoot_grows_with_rtt(void) +{ + struct mb_ecn_ctx * lo_rtt; + struct mb_ecn_ctx * hi_rtt; + uint64_t cap = 1ULL << 20; + + TEST_START(); + + lo_rtt = mk_ctx(); + hi_rtt = mk_ctx(); + if (lo_rtt == NULL || hi_rtt == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + /* Same bottleneck; feedback lags 1 vs 8 steps (~5 vs 40 ms). */ + shared_link_run(lo_rtt, NULL, cap, 1, 0, 2000); + shared_link_run(hi_rtt, NULL, cap, 8, 0, 2000); + + if (hi_rtt->ss_peak <= lo_rtt->ss_peak) { + printf("overshoot did not grow with RTT: %" PRIu64 + " vs %" PRIu64 ".\n", + hi_rtt->ss_peak, lo_rtt->ss_peak); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(lo_rtt); + mb_ecn_ctx_destroy(hi_rtt); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(lo_rtt); + mb_ecn_ctx_destroy(hi_rtt); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * 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. + */ +static int test_mb_ecn_washout_bucket(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t r0 = (uint64_t) 100 << 20; + uint64_t gap; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + gap = r0 >> 4; + + /* 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); + 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); + goto fail_ctx; + } + + if (ctx->rate <= r0 - gap) { + printf("washout reversed a ramp: %" PRIu64 ".\n", ctx->rate); + goto fail_ctx; + } + + if (ctx->r_bkt != ctx->rate || ctx->wash_acc != 0) { + printf("washout did not reset the bucket.\n"); + 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); + 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"); + 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; +} + int mb_ecn_test(int argc, char ** argv) { @@ -2561,6 +2892,7 @@ int mb_ecn_test(int argc, ret |= test_mb_ecn_starved_decrease_escape(); ret |= test_mb_ecn_starved_local_fallback(); ret |= test_mb_ecn_decrease_dt_invariant(); + ret |= test_mb_ecn_decrease_subms_carry(); ret |= test_mb_ecn_idle_resume_bounded(); ret |= test_mb_ecn_rate_floor(); ret |= test_mb_ecn_ece_staleness(); @@ -2583,6 +2915,9 @@ int mb_ecn_test(int argc, ret |= test_mb_ecn_max_filter(); ret |= test_mb_ecn_idle_clears_backlogged(); 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(); return ret; } |