diff options
| author | Dimitri Staessens <dimitri@ouroboros.rocks> | 2026-08-16 19:31:09 +0000 |
|---|---|---|
| committer | Sander Vrijders <sander@ouroboros.rocks> | 2026-08-31 08:31:45 +0200 |
| commit | 016c3c438e9b066bb45d4934ad039a49bde7014d (patch) | |
| tree | 968c83282c3a7143f4fe5b1954309db38cfc732f | |
| parent | 5c239c128c04883dbed6d66f574edf8b48d11e11 (diff) | |
| download | ouroboros-016c3c438e9b066bb45d4934ad039a49bde7014d.tar.gz ouroboros-016c3c438e9b066bb45d4934ad039a49bde7014d.zip | |
ipcpd: Use capacity queue estimation for mb-ecn
The mb-ecn algorithm was using rbuff queue depths in packets to mark,
but sockets in the poa component report capacity in bytes. The tx
rings are now adaptive to block on queuing delay instead of when full
to prevent buffer bloat, controllable via fccntl (FLOWSTXQDLY and
FLOWGTXQDLY).
Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>
29 files changed, 3569 insertions, 1082 deletions
diff --git a/cmake/config/tests.cmake b/cmake/config/tests.cmake index 37c06ca9..c0d0ed84 100644 --- a/cmake/config/tests.cmake +++ b/cmake/config/tests.cmake @@ -15,3 +15,9 @@ if(DISABLE_TESTS_CORE_DUMPS) else() message(STATUS "Core dumps in tests enabled") endif() + +set(MB_ECN_LAB_FULL FALSE CACHE BOOL + "Run the full mb-ecn lab rate sweep, up to 10 Gb/s") +if(MB_ECN_LAB_FULL) + message(STATUS "Full mb-ecn lab rate sweep enabled") +endif() diff --git a/doc/man/fccntl.3 b/doc/man/fccntl.3 index 767342b9..61022837 100644 --- a/doc/man/fccntl.3 +++ b/doc/man/fccntl.3 @@ -76,6 +76,16 @@ buffer. Takes a \fBsize_t \fIqlen\fR * as third argument. \fBFLOWGTXQLEN\fR - get the current number of packets in the transmit buffer. Takes a \fBsize_t \fIqlen\fR * as third argument. +\fBFLOWSTXQDLY\fR - set the target queueing delay for the transmit +occupancy limiter. Takes a \fBstruct timespec * \fItimeo\fR as third +argument. A zero delay leaves the ring unlimited, bounded only by its +slot count. The build-time default is applied when the flow is +created. + +\fBFLOWGTXQDLY\fR - retrieve the current target queueing delay for +the transmit occupancy limiter. Takes a \fBstruct timespec * \fItimeo\fR +as third argument. + \fBFRCTSFLAGS\fR - set the current flow flags. Takes an \fBuint16_t \fIflags\fR as third argument. Supported flags are: @@ -106,7 +116,8 @@ An invalid argument was passed. .B -EPERM Operation not permitted. This is returned when requesting the value of a timeout (FLOWGSNDTIMEO or FLOWGRCVTIMEO) when no such timeout was -set. +set, or when FLOWSTXQDLY/FLOWGTXQDLY is used on a flow with no +transmit occupancy ring. .B -EBADF Invalid flow descriptor passed. diff --git a/include/ouroboros/fccntl.h b/include/ouroboros/fccntl.h index e91e91dd..9a048095 100644 --- a/include/ouroboros/fccntl.h +++ b/include/ouroboros/fccntl.h @@ -67,6 +67,8 @@ #define FLOWGRXQLEN 00000010 /* Get queue length on rx */ #define FLOWGTXQLEN 00000011 /* Get queue length on tx */ #define FLOWGMTU 00000012 /* Get per-packet MTU */ +#define FLOWSTXQDLY 00000013 /* Set tx queueing delay */ +#define FLOWGTXQDLY 00000014 /* Get tx queueing delay */ /* FRCT operations */ #define FRCTSFLAGS 00001000 /* Set flags for FRCT */ diff --git a/include/ouroboros/ipcp-dev.h b/include/ouroboros/ipcp-dev.h index 8677adf0..ab99d950 100644 --- a/include/ouroboros/ipcp-dev.h +++ b/include/ouroboros/ipcp-dev.h @@ -71,13 +71,25 @@ int ipcp_flow_fini(int fd); int ipcp_flow_get_qoscube(int fd, qoscube_t * cube); +/* Bytes queued in the transmit path of the flow. */ size_t ipcp_flow_queued(int fd); +/* Mean size of a packet written to the flow, bytes. */ +size_t ipcp_flow_mean_len(int fd); + /* - * Identifies the transmit queue behind the flow. Flows attached to - * one point of attachment share a queue and answer the same id. + * Capacity estimator of the transmit queue behind the flow. + * Flows attached to one point of attachment share a queue and + * share the estimator. Armed by the IPCP; unarmed flows read 0. */ -int ipcp_flow_queue_id(int fd); +int ipcp_flow_cap_arm(int fd); + +void ipcp_flow_cap_update(int fd, + size_t qlen, + size_t len); + +/* Estimated capacity of the transmit queue (bytes/s), 0 = unknown. */ +uint64_t ipcp_flow_cap(int fd); int ipcp_spb_reserve(struct ssm_pk_buff ** spb, size_t len); diff --git a/include/ouroboros/ssm_rbuff.h b/include/ouroboros/ssm_rbuff.h index e77eec09..9ff880e9 100644 --- a/include/ouroboros/ssm_rbuff.h +++ b/include/ouroboros/ssm_rbuff.h @@ -55,6 +55,14 @@ void ssm_rbuff_clr_bits(struct ssm_rbuff * rb, uint32_t ssm_rbuff_get_flags(struct ssm_rbuff * rb); +size_t ssm_rbuff_get_limit(struct ssm_rbuff * rb); + +void ssm_rbuff_set_txq_target(struct ssm_rbuff * rb, + const struct timespec * ts); + +void ssm_rbuff_get_txq_target(struct ssm_rbuff * rb, + struct timespec * ts); + void ssm_rbuff_fini(struct ssm_rbuff * rb); int ssm_rbuff_mlock(struct ssm_rbuff * rb); @@ -62,6 +70,9 @@ int ssm_rbuff_mlock(struct ssm_rbuff * rb); int ssm_rbuff_write(struct ssm_rbuff * rb, size_t off); +int ssm_rbuff_write_prio(struct ssm_rbuff * rb, + size_t off); + int ssm_rbuff_write_b(struct ssm_rbuff * rb, size_t off, const struct timespec * abstime); 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; +} |