diff options
Diffstat (limited to 'src')
49 files changed, 6614 insertions, 480 deletions
diff --git a/src/ipcpd/common/connmgr.c b/src/ipcpd/common/connmgr.c index 48ad79ba..6384a45c 100644 --- a/src/ipcpd/common/connmgr.c +++ b/src/ipcpd/common/connmgr.c @@ -30,6 +30,7 @@ #include <ouroboros/logs.h> #include <ouroboros/notifier.h> #include <ouroboros/pthread.h> +#include <ouroboros/qos.h> #include "connmgr.h" #include "ipcp.h" @@ -123,7 +124,6 @@ static int add_comp_conn(enum comp_id id, static void * flow_acceptor(void * o) { int fd; - qosspec_t qs; struct conn_info rcv_info; struct conn_info fail_info; struct timespec timeo = TIMESPEC_INIT_MS(CONNMGR_RCV_TIMEOUT); @@ -134,7 +134,8 @@ static void * flow_acceptor(void * o) memset(&fail_info, 0, sizeof(fail_info)); while (true) { - int id; + qosspec_t qs = qos_raw; + int id; fd = flow_accept(&qs, NULL); if (fd < 0) { diff --git a/src/ipcpd/config.h.in b/src/ipcpd/config.h.in index 7edec526..517016cc 100644 --- a/src/ipcpd/config.h.in +++ b/src/ipcpd/config.h.in @@ -48,6 +48,8 @@ #define IPCP_UNICAST_MPL @IPCP_UNICAST_MPL@ #define IPCP_UNICAST_MTU @IPCP_UNICAST_MTU@ #define CONNMGR_RCV_TIMEOUT @CONNMGR_RCV_TIMEOUT@ +#define CA_BUCKETS @IPCP_CA_BUCKETS@ +#cmakedefine IPCP_CA_PER_FLOW #cmakedefine DISABLE_CORE_LOCK #cmakedefine BUILD_CONTAINER diff --git a/src/ipcpd/unicast/CMakeLists.txt b/src/ipcpd/unicast/CMakeLists.txt index d3388112..2373f877 100644 --- a/src/ipcpd/unicast/CMakeLists.txt +++ b/src/ipcpd/unicast/CMakeLists.txt @@ -6,6 +6,7 @@ protobuf_generate_c(DHT_PROTO_SRCS DHT_PROTO_HDRS set(UNICAST_SOURCES addr-auth.c ca.c + cap.c connmgr.c dir.c dt.c @@ -43,7 +44,9 @@ ouroboros_target_debug_definitions(${IPCP_UNICAST_TARGET}) install(TARGETS ${IPCP_UNICAST_TARGET} RUNTIME DESTINATION ${CMAKE_INSTALL_SBINDIR}) if(BUILD_TESTS) + add_subdirectory(ca/tests) add_subdirectory(dir/tests) add_subdirectory(pff/tests) add_subdirectory(routing/tests) + add_subdirectory(tests) endif() diff --git a/src/ipcpd/unicast/ca.c b/src/ipcpd/unicast/ca.c index a1751672..f2a65fff 100644 --- a/src/ipcpd/unicast/ca.c +++ b/src/ipcpd/unicast/ca.c @@ -22,17 +22,48 @@ #define OUROBOROS_PREFIX "ca" +#include "config.h" + +#include <ouroboros/list.h> #include <ouroboros/logs.h> #include "ca.h" #include "ca/pol.h" +#include <pthread.h> +#include <stdlib.h> + +/* + * A ca_ctx holds congestion state for a (peer address, qos cube) PATH, + * not for a flow. In the default build the façade interns one ctx per + * (addr, qc) and shares it across every flow on that path; the policy + * runs on the shared ctx and cannot tell one flow from many. Per-flow + * ctx (IPCP_CA_PER_FLOW) is a testing build only: it skips interning so + * every flow gets its own ctx. + */ + +struct ca_ctx { + uint64_t addr; + qoscube_t qc; + size_t refs; + void * pol; /* policy ctx (ops->ctx_create result) */ + struct list_head next; +}; + struct { - struct ca_ops * ops; + struct ca_ops * ops; +#ifndef IPCP_CA_PER_FLOW + struct list_head buckets[CA_BUCKETS]; + pthread_mutex_t mtx; +#endif } ca; -int ca_init(enum pol_cong_avoid pol) +int ca_init(enum pol_cong_avoid pol, + uint32_t rtt_ms) { +#ifndef IPCP_CA_PER_FLOW + size_t i; +#endif switch(pol) { case CA_NONE: log_dbg("Disabling congestion control."); @@ -41,68 +72,196 @@ int ca_init(enum pol_cong_avoid pol) case CA_MB_ECN: log_dbg("Using multi-bit ECN."); ca.ops = &mb_ecn_ca_ops; + mb_ecn_init(rtt_ms); break; default: return -1; } +#ifndef IPCP_CA_PER_FLOW + for (i = 0; i < CA_BUCKETS; i++) + list_head_init(&ca.buckets[i]); + + if (pthread_mutex_init(&ca.mtx, NULL) != 0) + return -1; +#endif return 0; } void ca_fini(void) { +#ifndef IPCP_CA_PER_FLOW + size_t i; + + /* Data path is stopped; drain any ctx a flow left interned. */ + for (i = 0; i < CA_BUCKETS; i++) { + struct list_head * p; + struct list_head * h; + + list_for_each_safe(p, h, &ca.buckets[i]) { + struct ca_ctx * ctx; + ctx = list_entry(p, struct ca_ctx, next); + list_del(&ctx->next); + ca.ops->ctx_destroy(ctx->pol); + free(ctx); + } + } + + pthread_mutex_destroy(&ca.mtx); +#endif ca.ops = NULL; } -void * ca_ctx_create(void) +#ifndef IPCP_CA_PER_FLOW +static size_t ca_bucket(uint64_t addr, + qoscube_t qc) { - return ca.ops->ctx_create(); + return (addr ^ (addr >> 32) ^ (uint64_t) qc) & (CA_BUCKETS - 1); +} +#endif + +void * ca_ctx_get(uint64_t addr, + qoscube_t qc) +{ + struct ca_ctx * ctx; +#ifndef IPCP_CA_PER_FLOW + struct list_head * p; + size_t b = ca_bucket(addr, qc); + + pthread_mutex_lock(&ca.mtx); + + list_for_each(p, &ca.buckets[b]) { + ctx = list_entry(p, struct ca_ctx, next); + if (ctx->addr == addr && ctx->qc == qc) { + ctx->refs++; + pthread_mutex_unlock(&ca.mtx); + return ctx; + } + } +#endif + ctx = malloc(sizeof(*ctx)); + if (ctx == NULL) + goto fail_ctx; + + ctx->pol = ca.ops->ctx_create(); + if (ctx->pol == NULL) + goto fail_pol; + + ctx->addr = addr; + ctx->qc = qc; + ctx->refs = 1; + +#ifndef IPCP_CA_PER_FLOW + list_add(&ctx->next, &ca.buckets[b]); + + pthread_mutex_unlock(&ca.mtx); +#endif + return ctx; + fail_pol: + free(ctx); + fail_ctx: +#ifndef IPCP_CA_PER_FLOW + pthread_mutex_unlock(&ca.mtx); +#endif + return NULL; } -void ca_ctx_destroy(void * ctx) +void ca_ctx_put(void * _ctx) { - return ca.ops->ctx_destroy(ctx); + struct ca_ctx * ctx = _ctx; + +#ifndef IPCP_CA_PER_FLOW + pthread_mutex_lock(&ca.mtx); + + if (--ctx->refs > 0) { + pthread_mutex_unlock(&ca.mtx); + return; + } + + list_del(&ctx->next); + + pthread_mutex_unlock(&ca.mtx); +#endif + ca.ops->ctx_destroy(ctx->pol); + + free(ctx); } -ca_wnd_t ca_ctx_update_snd(void * ctx, - size_t len) +time_t ca_ctx_update_snd(void * _ctx, + size_t len, + uint8_t lecn, + uint64_t * ftag) { - return ca.ops->ctx_update_snd(ctx, len); + struct ca_ctx * ctx = _ctx; + + return ca.ops->ctx_update_snd(ctx->pol, len, lecn, ftag); } -bool ca_ctx_update_rcv(void * ctx, +bool ca_ctx_update_rcv(void * _ctx, size_t len, uint8_t ecn, - uint16_t * ece) + uint8_t cap, + uint16_t * ece, + uint8_t * fcap) { - return ca.ops->ctx_update_rcv(ctx, len, ecn, ece); + struct ca_ctx * ctx = _ctx; + + return ca.ops->ctx_update_rcv(ctx->pol, len, ecn, cap, ece, fcap); } -void ca_ctx_update_ece(void * ctx, - uint16_t ece) +void ca_ctx_update_ece(void * _ctx, + uint16_t ece, + uint8_t cap) { - return ca.ops->ctx_update_ece(ctx, ece); + struct ca_ctx * ctx = _ctx; + + return ca.ops->ctx_update_ece(ctx->pol, ece, cap); } -void ca_wnd_wait(ca_wnd_t wnd) +bool ca_ctx_hb_due(void * _ctx, + uint64_t now) { - return ca.ops->wnd_wait(wnd); + struct ca_ctx * ctx = _ctx; + + if (ca.ops->ctx_hb_due == NULL) + return false; + + return ca.ops->ctx_hb_due(ctx->pol, now); +} + +void ca_ctx_rtt(void * _ctx, + uint64_t now, + uint64_t rtt) +{ + struct ca_ctx * ctx = _ctx; + + if (ca.ops->ctx_rtt == NULL) + return; + + ca.ops->ctx_rtt(ctx->pol, now, rtt); } -int ca_calc_ecn(int fd, +int ca_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, size_t len) { - return ca.ops->calc_ecn(fd, ecn, qc, len); + return ca.ops->calc_ecn(queued, ecn, qc, len); } -ssize_t ca_print_stats(void * ctx, +bool ca_marks_ecn(void) +{ + return ca.ops->marks_ecn; +} + +ssize_t ca_print_stats(void * _ctx, char * buf, size_t len) { + struct ca_ctx * ctx = _ctx; + if (ca.ops->print_stats == NULL) return 0; - return ca.ops->print_stats(ctx, buf, len); + return ca.ops->print_stats(ctx->pol, buf, len); } diff --git a/src/ipcpd/unicast/ca.h b/src/ipcpd/unicast/ca.h index 47ea15a0..fc6de445 100644 --- a/src/ipcpd/unicast/ca.h +++ b/src/ipcpd/unicast/ca.h @@ -29,38 +29,51 @@ #include <stdbool.h> #include <sys/types.h> -typedef union { - time_t wait; -} ca_wnd_t; +/* Buffer a policy's ca_print_stats output must fit in. */ +#define CA_STATS_STRLEN 2048 -int ca_init(enum pol_cong_avoid ca); +int ca_init(enum pol_cong_avoid ca, + uint32_t rtt_ms); void ca_fini(void); /* OPS */ -void * ca_ctx_create(void); +void * ca_ctx_get(uint64_t addr, + qoscube_t qc); -void ca_ctx_destroy(void * ctx); +void ca_ctx_put(void * ctx); -ca_wnd_t ca_ctx_update_snd(void * ctx, - size_t len); +time_t ca_ctx_update_snd(void * ctx, + size_t len, + uint8_t lecn, + uint64_t * ftag); bool ca_ctx_update_rcv(void * ctx, size_t len, uint8_t ecn, - uint16_t * ece); + uint8_t cap, + uint16_t * ece, + uint8_t * fcap); void ca_ctx_update_ece(void * ctx, - uint16_t ece); + uint16_t ece, + uint8_t cap); + +bool ca_ctx_hb_due(void * ctx, + uint64_t now); -void ca_wnd_wait(ca_wnd_t wnd); +void ca_ctx_rtt(void * ctx, + uint64_t now, + uint64_t rtt); -int ca_calc_ecn(int fd, +int ca_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, size_t len); +bool ca_marks_ecn(void); + ssize_t ca_print_stats(void * ctx, char * buf, size_t len); diff --git a/src/ipcpd/unicast/ca/mb-ecn.c b/src/ipcpd/unicast/ca/mb-ecn.c index b310c4fc..a4c9f29e 100644 --- a/src/ipcpd/unicast/ca/mb-ecn.c +++ b/src/ipcpd/unicast/ca/mb-ecn.c @@ -28,9 +28,10 @@ #include "config.h" -#include <ouroboros/ipcp-dev.h> #include <ouroboros/time.h> +#include <ouroboros/utils.h> +#include "cap.h" #include "mb-ecn.h" #include <inttypes.h> @@ -38,47 +39,206 @@ #include <string.h> #include <stdio.h> -/* congestion avoidance constants */ -#define CA_SHFT 5 /* Average over 32 pkts */ -#define CA_WND (1 << CA_SHFT) /* 32 pkts receiver wnd */ -#define CA_UPD (1 << (CA_SHFT - 2)) /* Update snd every 8 pkt */ -#define CA_SLOT 24 /* Initial slot = 16 ms */ -#define CA_INC 1UL << 16 /* ~4MiB/s^2 additive inc */ -#define CA_IWL 1UL << 16 /* Initial limit ~4MiB/s */ -#define CA_MINPS 8 /* Mimimum pkts / slot */ -#define CA_MAXPS 64 /* Maximum pkts / slot */ -#define ECN_Q_SHFT 4 -#define ts_to_ns(ts) ((size_t) ts.tv_sec * BILLION + ts.tv_nsec) +/* + * Multi-bit ECN congestion avoidance: a rate-based controller. The + * sender paces a token bucket at a rate steered by graded ECN + * feedback, so the backoff is proportional to the congestion. A + * backlogged flow ramps in slow start to find the path capacity, then + * settles into AIMD around its fair share. There is no sliding window + * and no per-flow timer; the control runs on sends. + * + * Every rate step is scaled by elapsed wall-clock time (Δt), not by + * packet count, so the per-second dynamics are RTT-independent. The + * receiver's averaging window and the sender's feedback staleness both + * stretch with the flow's byte rate, so a slow flow is measured and + * controlled like a fast one; CA_RATE_MIN only bounds those horizons + * (window <= CA_TW_ABSMAX, TTL ~8 s). + * + * The ramp clock ss_tc = 2 * RTT holds slow-start overshoot near 1.65x + * (e^1/2): it seeds from the declared max_rtt and then tracks the + * heartbeat's measured RTT. Feedback silence past the staleness horizon + * leaves slow start; a sustained run of it restarts at the floor. + * + * The floor and the AI slope scale with the path: forwarders stamp + * their measured link capacity into the PCI (cap.c), the receiver + * 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. + */ + +#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 */ +#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_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) */ + +#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_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 */ +#define CA_DT_CTRL (BILLION / 1000) /* min rate-update spacing */ +#define CA_DT_CAP (BILLION / 20) /* idle-resume Δt clamp 50ms */ +/* Floor of the rate-relative feedback staleness (ctx->ece_ttl). */ +#define CA_ECE_TTL ((1 << CA_TW_GAP_SHFT) * CA_TW_INIT) +#define CA_SS_RTT_MUL 2 /* ss_tc = 2 * layer RTT */ +#define CA_SS_TC_MIN (BILLION / 1000) /* ramp floor 1 ms */ +#define CA_SS_TC_MAX (4ULL * BILLION) /* ramp ceiling 4 s */ +#define CA_HB_MIN (40 * MILLION) /* heartbeat interval floor */ +#define CA_HB_LOSS 4 /* stale horizons -> restart */ +#define CA_RTT_SHFT 2 /* ss_tc EWMA weight 1/4 */ +#define CA_SS_RTT_DEF 200 /* default layer RTT (ms) */ +#define CA_WASH_BKT (BILLION / 32) /* washout bucket ~31 ms */ +#define CA_WASH_SHFT 2 /* damp 1/4 of bucket change */ + +#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 */ +#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 */ + +/* + * Retuning invariants (pinned by the unit tests): + * - (1 << CA_TW_GAP_SHFT) * CA_TW_INIT > 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_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. + */ struct mb_ecn_ctx { - uint16_t rx_ece; /* Level of congestion (upstream) */ - size_t rx_ctr; /* Receiver side packet counter */ - - uint16_t tx_ece; /* Level of congestion (downstream) */ - size_t tx_ctr; /* Sender side packet counter */ - size_t tx_wbc; /* Window byte count */ - size_t tx_wpc; /* Window packet count */ - size_t tx_wbl; /* Window byte limit */ - bool tx_cav; /* Congestion avoidance */ - size_t tx_mul; /* Slot size multiplier */ - size_t tx_inc; /* Additive increase */ - size_t tx_slot; + uint16_t rx_ece; /* smoothed congestion echo (32 * ecn) */ + uint64_t rx_acc; /* window integral of ecn * dt */ + uint64_t rx_byt; /* bytes arrived in current window */ + uint64_t rx_ts; /* last packet arrival (ns) */ + uint64_t rx_win; /* window start (ns) */ + uint64_t rx_tw; /* adaptive averaging window (ns) */ + uint8_t rx_cap; /* window bottleneck capacity code */ + + uint16_t tx_ece; /* congestion reported from downstream */ + uint16_t tx_ecp; /* previous tx_ece (rise detection) */ + uint8_t tx_loc; /* local first-hop ecn mark (fallback) */ + bool tx_cav; /* past slow start */ + bool ai_hold; /* freeze AI after loss until clear */ + uint64_t rate; /* paced send rate (bytes/s) */ + uint64_t rate_min; /* capacity-derived rate floor (B/s) */ + uint64_t ai_rate; /* additive-increase slope (B/s^2) */ + uint64_t ece_ttl; /* how long feedback stays valid (ns) */ + uint64_t ss_tc; /* slow-start time constant (ns) */ + uint64_t r_bkt; /* rate snapshot at last washout bucket */ + uint64_t wash_acc; /* washout bucket time accumulator (ns) */ + uint64_t dec_acc; /* sub-ms decrease time carried (ns) */ + uint64_t inv_rate; /* fixed-point 1/rate for pacing */ + uint64_t vt; /* virtual service clock (bytes) */ + uint64_t lead; /* pacer lead of last send (bytes) */ + uint64_t last_ts; /* last clock advance (ns) */ + uint64_t last_ctrl; /* last rate update (ns) */ + uint64_t last_fb; /* last congestion feedback (ns) */ + uint64_t last_sig; /* last liveness signal, incl. hb (ns) */ + uint64_t n_fb; /* feedback updates received */ + uint64_t last_hb; /* last heartbeat emitted (ns) */ + uint64_t last_res; /* last resume from idle (ns) */ + uint64_t last_loc; /* last local mark seen (ns) */ + uint64_t last_cap; /* last capacity applied (ns) */ + + uint64_t snd_byt; /* bytes offered this window (capped) */ + 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) */ + bool backlogged; /* offered load keeps the pacer busy */ + bool src_limited; /* rate held at offered-load ceiling */ + bool started; /* a real send has occurred */ + + /* Diagnostics only, read by mb_ecn_print_stats. */ + uint8_t tx_cap; /* path capacity code fed back to us */ + uint64_t n_ctrl; /* control steps taken */ + uint64_t t_ctrl; /* wall time covered by steps (ns) */ + uint64_t t_bank; /* increase time banked in steps (ns) */ + uint64_t n_ttl; /* feedback aged out (TTL) */ + uint64_t n_cap; /* capacity updates applied */ + uint64_t n_loss; /* signal-loss cuts (collapse) */ + uint64_t n_rtt; /* heartbeat RTT samples folded */ + uint64_t ss_peak; /* peak rate in slow start (bytes/s) */ }; +/* Layer slow-start time constant (ns), from the declared RTT. */ +static uint64_t mb_ecn_ss_tc = (uint64_t) CA_SS_RTT_MUL * + CA_SS_RTT_DEF * MILLION; + struct ca_ops mb_ecn_ca_ops = { .ctx_create = mb_ecn_ctx_create, .ctx_destroy = mb_ecn_ctx_destroy, .ctx_update_snd = mb_ecn_ctx_update_snd, .ctx_update_rcv = mb_ecn_ctx_update_rcv, .ctx_update_ece = mb_ecn_ctx_update_ece, - .wnd_wait = mb_ecn_wnd_wait, + .ctx_hb_due = mb_ecn_ctx_hb_due, + .ctx_rtt = mb_ecn_ctx_rtt, .calc_ecn = mb_ecn_calc_ecn, + .marks_ecn = true, .print_stats = mb_ecn_print_stats }; +static uint64_t mb_ecn_rate_inv(uint64_t rate) +{ + return ((uint64_t) BILLION << CA_INV_SHFT) / rate; +} + +/* + * Feedback arrives once per receiver window, and the window tracks + * the flow's byte rate. Mirror it: age the signal out only past the + * quiet horizon at the current rate, floored for fast flows. + */ +static uint64_t mb_ecn_ece_ttl(uint64_t rate) +{ + uint64_t ttl; + + ttl = (1 << CA_TW_GAP_SHFT) * CA_RX_WBYTES * BILLION / rate; + + return ttl > (uint64_t) CA_ECE_TTL ? ttl : (uint64_t) CA_ECE_TTL; +} + +/* Derive the layer slow-start slope from the declared RTT (ms). */ +void mb_ecn_init(uint32_t rtt_ms) +{ + uint64_t tc; + + if (rtt_ms == 0) /* unspecified: safe default */ + rtt_ms = CA_SS_RTT_DEF; + + tc = (uint64_t) CA_SS_RTT_MUL * rtt_ms * MILLION; + if (tc < (uint64_t) CA_SS_TC_MIN) + tc = CA_SS_TC_MIN; + + mb_ecn_ss_tc = tc; +} + void * mb_ecn_ctx_create(void) { struct timespec now; + uint64_t t; struct mb_ecn_ctx * ctx; ctx = malloc(sizeof(*ctx)); @@ -89,10 +249,29 @@ void * mb_ecn_ctx_create(void) memset(ctx, 0, sizeof(*ctx)); - ctx->tx_mul = CA_SLOT; - ctx->tx_wbl = CA_IWL; - ctx->tx_inc = CA_INC; - ctx->tx_slot = ts_to_ns(now) >> ctx->tx_mul; + t = TS_TO_UINT64(now); + + ctx->rate = CA_RATE_INIT; + ctx->rate_min = CA_RATE_MIN; + ctx->ai_rate = CA_AI_RATE; + ctx->ss_tc = mb_ecn_ss_tc; + ctx->ece_ttl = mb_ecn_ece_ttl(CA_RATE_INIT); + ctx->r_bkt = CA_RATE_INIT; + ctx->inv_rate = mb_ecn_rate_inv(CA_RATE_INIT); + ctx->rx_ts = t; + ctx->rx_win = t; + ctx->rx_tw = CA_TW_INIT; + ctx->last_ts = t; + ctx->last_ctrl = t; + ctx->last_fb = t; + ctx->last_sig = t; + ctx->last_loc = t; + ctx->last_cap = t; + + /* 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->backlogged = true; return (void *) ctx; } @@ -102,158 +281,619 @@ void mb_ecn_ctx_destroy(void * ctx) free(ctx); } -#define _slot_after(new, old) ((int64_t) (old - new) < 0) +/* Local first-hop mark exits slow start and covers dead feedback. */ +static void mb_ecn_loc(struct mb_ecn_ctx * ctx, + uint8_t lecn, + uint64_t t) +{ + if (lecn == 0) + return; + + ctx->tx_loc = lecn; + ctx->tx_cav = true; + ctx->last_loc = t; +} + +/* Slow start: ramp only while backlogged. */ +static void mb_ecn_slow_start(struct mb_ecn_ctx * ctx, + uint64_t dta) +{ + if (ctx->backlogged) + ctx->rate += ctx->rate * dta / ctx->ss_tc; + + ctx->r_bkt = ctx->rate; +} + +/* Additive increase plus a rate-independent proportional probe. */ +static void mb_ecn_increase(struct mb_ecn_ctx * ctx, + uint64_t dta) +{ + if (!ctx->backlogged) + return; + + /* After a loss, hold until a clean signal drains the queue. */ + if (ctx->ai_hold) + return; + + ctx->rate += ctx->ai_rate * dta / BILLION; + ctx->rate += ctx->rate * dta / CA_PROBE_TC; +} + +/* + * Multiplicative decrease: cut proportional to mark x elapsed time, + * plus a one-sided lead that cuts extra while the mark is rising. + */ +static void mb_ecn_decrease(struct mb_ecn_ctx * ctx, + uint64_t dtc) +{ + uint64_t dtm; + uint64_t mark; + uint64_t rise; + uint64_t cut; + uint16_t m; + + m = ctx->tx_ece > 0 ? ctx->tx_ece + : (uint16_t) (ctx->tx_loc << CA_SHFT); + if (m == 0) { + ctx->dec_acc = 0; /* unmarked time is not banked */ + ctx->tx_ecp = 0; + return; + } + + mark = MIN(m, CA_ECE_REF); + + /* One-sided lead: cut extra while the mark is still rising. */ + rise = m > ctx->tx_ecp ? MIN(m - ctx->tx_ecp, CA_ECE_REF) : 0; + cut = ctx->rate * rise / (CA_ECE_REF * CA_MD_KD_DIV); + + /* Honest elapsed ms; the sub-ms remainder carries over. */ + ctx->dec_acc += dtc; + dtm = ctx->dec_acc / MILLION; + ctx->dec_acc -= dtm * MILLION; + if (mark * dtm >= CA_ECE_REF * 500) + cut += ctx->rate / 2; + else + cut += ctx->rate * mark * dtm / (CA_ECE_REF * 1000); + + if (cut > ctx->rate / 2) + 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; + + ctx->r_bkt = ctx->rate; + ctx->wash_acc = 0; +} + +/* 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) { + ctx->src_limited = false; + 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; + + 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; + } +} + +static void mb_ecn_ctrl(struct mb_ecn_ctx * ctx, + uint64_t dtc) +{ + uint64_t dta; + uint64_t lo; + + /* AI and slow start bank at most CA_DT_CAP of idle time. */ + dta = MIN(dtc, (uint64_t) CA_DT_CAP); + + ctx->n_ctrl++; + ctx->t_ctrl += dtc; + ctx->t_bank += dta; + + 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); + } + + mb_ecn_ceiling(ctx); + + /* Capacity floor only while backlogged; else the absolute floor. */ + lo = ctx->backlogged ? ctx->rate_min : (uint64_t) CA_RATE_MIN; + if (ctx->rate < lo) + ctx->rate = lo; + + if (ctx->rate > CA_RATE_MAX) + ctx->rate = CA_RATE_MAX; + + ctx->inv_rate = mb_ecn_rate_inv(ctx->rate); + ctx->ece_ttl = mb_ecn_ece_ttl(ctx->rate); + + if (!ctx->tx_cav && ctx->rate > ctx->ss_peak) + ctx->ss_peak = ctx->rate; +} + +/* Fold offered into the max filter: rise at once, decay 1/16 per window. */ +static void mb_ecn_offered(struct mb_ecn_ctx * ctx, + uint64_t offered, + uint64_t elapsed) +{ + uint64_t n; + + if (offered >= ctx->snd_rate) { + ctx->snd_rate = offered; + return; + } + + n = MIN(elapsed / CA_SND_WIN, CA_SND_DEC_CAP); + while (n-- > 0 && ctx->snd_rate > offered) + ctx->snd_rate -= (ctx->snd_rate - offered) >> CA_SND_DEC_SHFT; +} + +/* + * Close the utilisation window: set backlogged from the level test, + * fold offered into the max filter, then reset the window. + */ +static void mb_ecn_win(struct mb_ecn_ctx * ctx, + uint64_t t) +{ + uint64_t elapsed = t - ctx->snd_win; + uint64_t offered; + bool was = ctx->backlogged; + + offered = ctx->snd_byt * BILLION / elapsed; + + ctx->backlogged = offered * CA_USE_DEN >= ctx->snd_r0 * CA_USE_NUM; + + if (!was && ctx->backlogged) /* resume: fresh liveness baseline */ + ctx->last_res = t; + + mb_ecn_offered(ctx, offered, elapsed); + + if (ctx->backlogged) + ctx->src_limited = false; + + ctx->snd_win = t; + ctx->snd_byt = 0; + ctx->snd_r0 = ctx->rate; +} + +/* Age out congestion, local-mark and capacity signals once stale. */ +/* Heartbeat interval: ~1 RTT, floored so fast links don't over-probe. */ +static uint64_t mb_ecn_t_hb(const struct mb_ecn_ctx * ctx) +{ + uint64_t t = ctx->ss_tc >> 1; + + return t > (uint64_t) CA_HB_MIN ? t : CA_HB_MIN; +} + +/* Feedback collapsed while backlogged: halve like an RTO, stay in AIMD. */ +static void mb_ecn_loss(struct mb_ecn_ctx * ctx, + uint64_t t) +{ + ctx->rate -= ctx->rate / 2; + if (ctx->rate < (uint64_t) CA_RATE_MIN) + ctx->rate = CA_RATE_MIN; + + ctx->r_bkt = ctx->rate; + ctx->inv_rate = mb_ecn_rate_inv(ctx->rate); + ctx->ece_ttl = mb_ecn_ece_ttl(ctx->rate); + ctx->last_sig = t; + ctx->ai_hold = true; + ctx->n_loss++; +} + +static void mb_ecn_age(struct mb_ecn_ctx * ctx, + uint64_t t) +{ + uint64_t ttl = ctx->ece_ttl; + uint64_t ref = ctx->last_sig > ctx->last_res + ? ctx->last_sig : ctx->last_res; + uint64_t gap = t - ref; + + /* + * Sustained silence while backlogged is feedback collapse: cut + * the rate in half and stay in AIMD, so a recovering flow climbs + * back additively instead of re-ramping. Repeated silence decays + * it geometrically toward the floor. + */ + if (ctx->backlogged && ctx->n_fb + ctx->n_rtt > 0 + && gap > (uint64_t) CA_HB_LOSS * ttl) { + mb_ecn_loss(ctx, t); + return; + } + + if (t - ctx->last_fb > ctx->ece_ttl) { + if (ctx->tx_ece > 0) + ctx->n_ttl++; + ctx->tx_ece = 0; + } + + if (t - ctx->last_loc > ctx->ece_ttl) + ctx->tx_loc = 0; + + /* Stale capacity: fall back to the compile-time defaults. */ + if (t - ctx->last_cap > ctx->ece_ttl << CA_CAP_TTL_SHFT) { + ctx->rate_min = CA_RATE_MIN; + ctx->ai_rate = CA_AI_RATE; + ctx->tx_cap = 0; + } +} + +/* Advance the virtual clock; a gap past CA_DT_CAP credits a burst. */ +static void mb_ecn_advance(struct mb_ecn_ctx * ctx, + uint64_t dt, + size_t len, + uint64_t ftag) +{ + uint64_t burst; + uint64_t owed; + + if (dt <= (uint64_t) CA_DT_CAP) { + ctx->vt += ctx->rate * dt / BILLION; + return; + } + + burst = ctx->rate * CA_DT_CAP / BILLION; + if (burst < (uint64_t) len) + burst = len; + + owed = ftag > ctx->vt ? ftag - ctx->vt + burst : burst; + + /* Clamp so owed * BILLION cannot wrap (2^33 B backlog). */ + if (owed > (1ULL << 33)) + owed = 1ULL << 33; + + if (dt >= owed * BILLION / ctx->rate) + ctx->vt += owed; + else + ctx->vt += ctx->rate * dt / BILLION; +} + +static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx, + size_t len, + uint64_t t, + uint64_t * ftag) +{ + uint64_t dt; + uint64_t dtc; + uint64_t s; + + /* Lazy warm-up seed: packet #1 is never an idle resume. */ + if (!ctx->started) { + ctx->started = true; + ctx->last_ts = t; + ctx->last_res = t; + ctx->snd_win = t; + ctx->snd_r0 = ctx->rate; + } + + dt = t - ctx->last_ts; + ctx->last_ts = t; + + /* Idle gap clears backlog before aging: no false loss on resume. */ + if (dt > (uint64_t) CA_DT_CAP) + ctx->backlogged = false; + + mb_ecn_age(ctx, t); + + /* Offered-load estimator: accumulate, gate growth, size ceiling. */ + ctx->snd_byt += len; + if (ctx->snd_byt > (uint64_t) CA_SND_BYT_MAX) + ctx->snd_byt = CA_SND_BYT_MAX; -ca_wnd_t mb_ecn_ctx_update_snd(void * _ctx, - size_t len) + if (t - ctx->snd_win >= (uint64_t) CA_SND_WIN) + mb_ecn_win(ctx, t); + + /* Rate update before the vt advance: burst uses the clamped rate. */ + dtc = t - ctx->last_ctrl; + if (dtc >= (uint64_t) CA_DT_CTRL) { + ctx->last_ctrl = t; + mb_ecn_ctrl(ctx, dtc); + } + + mb_ecn_advance(ctx, dt, len, *ftag); + + /* SFQ start tag: behind the clock starts now, ahead waits. */ + s = *ftag > ctx->vt ? *ftag : ctx->vt; + *ftag = s + len; + + ctx->lead = s - ctx->vt; + + /* Reciprocal pacing; folded so any lead * rate stays in range. */ + if (s > ctx->vt) + return (time_t) ((ctx->lead * (ctx->inv_rate >> 16)) + >> (CA_INV_SHFT - 16)); + + return 0; +} + +time_t mb_ecn_ctx_update_snd(void * _ctx, + size_t len, + uint8_t lecn, + uint64_t * ftag) { struct timespec now; - size_t slot; - ca_wnd_t wnd; + uint64_t t; struct mb_ecn_ctx * ctx = _ctx; clock_gettime(PTHREAD_COND_CLOCK, &now); - slot = ts_to_ns(now) >> ctx->tx_mul; + t = TS_TO_UINT64(now); - ctx->tx_ctr++; - ctx->tx_wpc++; - ctx->tx_wbc += len; + mb_ecn_loc(ctx, lecn, t); - if (ctx->tx_ctr > CA_WND) - ctx->tx_ece = 0; + return mb_ecn_snd(ctx, len, t, ftag); +} - if (_slot_after(slot, ctx->tx_slot)) { - bool carry = false; /* may carry over if window increases */ - - ctx->tx_slot = slot; - - if (!ctx->tx_cav) { /* Slow start */ - if (ctx->tx_wbc > ctx->tx_wbl) - ctx->tx_wbl <<= 1; - } else { - if (ctx->tx_ece) /* Mult. Decrease */ - ctx->tx_wbl -= (ctx->tx_wbl * ctx->tx_ece) - >> (CA_SHFT + 8); - else /* Add. Increase */ - ctx->tx_wbl = ctx->tx_wbc + ctx->tx_inc; - } - - /* Window scaling */ - if (ctx->tx_wpc < CA_MINPS) { - size_t fact = 0; /* factor to scale the window up */ - size_t pkts = ctx->tx_wpc; - while (pkts < CA_MINPS) { - pkts <<= 1; - fact++; - } - ctx->tx_mul += fact; - ctx->tx_slot >>= fact; - if ((ctx->tx_slot & ((1 << fact) - 1)) == 0) { - carry = true; - ctx->tx_slot += 1; - } - ctx->tx_wbl <<= fact; - ctx->tx_inc <<= fact; - } else if (ctx->tx_wpc > CA_MAXPS) { - size_t fact = 0; /* factor to scale the window down */ - size_t pkts = ctx->tx_wpc; - while (pkts > CA_MAXPS) { - pkts >>= 1; - fact++; - } - ctx->tx_mul -= fact; - ctx->tx_slot <<= fact; - ctx->tx_wbl >>= fact; - ctx->tx_inc >>= fact; - } else { - ctx->tx_slot = slot; - } - - if (!carry) { - ctx->tx_wbc = 0; - ctx->tx_wpc = 0; - } - } +/* Estimator idle, or a gap past ~4 current windows: restart fresh. */ +static bool mb_ecn_rcv_fresh(const struct mb_ecn_ctx * ctx, + uint64_t dt) +{ + if (ctx->rx_ece == 0 && ctx->rx_acc == 0) + return true; - if (ctx->tx_wbc > ctx->tx_wbl) - wnd.wait = ((ctx->tx_slot + 1) << ctx->tx_mul) - ts_to_ns(now); + return dt > ctx->rx_tw << CA_TW_GAP_SHFT; +} + +/* Size the next averaging window to ~CA_N_TARGET packets at this rate. */ +static void mb_ecn_resize(struct mb_ecn_ctx * ctx, + uint64_t win) +{ + uint64_t tw = CA_RX_WBYTES * win / ctx->rx_byt; + + if (tw > ctx->rx_tw) + ctx->rx_tw += (tw - ctx->rx_tw) >> CA_TW_SM_SHFT; else - wnd.wait = 0; + ctx->rx_tw -= (ctx->rx_tw - tw) >> CA_TW_SM_SHFT; - return wnd; + if (ctx->rx_tw < CA_TW_MIN) + ctx->rx_tw = CA_TW_MIN; + + if (ctx->rx_tw > CA_TW_ABSMAX) + ctx->rx_tw = CA_TW_ABSMAX; } -void mb_ecn_wnd_wait(ca_wnd_t wnd) +static bool mb_ecn_rcv(struct mb_ecn_ctx * ctx, + size_t len, + uint8_t ecn, + uint8_t cap, + uint16_t * ece, + uint8_t * fcap, + uint64_t t) { - if (wnd.wait > 0) { - struct timespec s = TIMESPEC_INIT_S(0); - if (wnd.wait > BILLION) /* Don't care throttling < 1s */ - s.tv_sec = 1; - else - s.tv_nsec = wnd.wait; + uint64_t dt; + uint64_t win; + + dt = t - ctx->rx_ts; + ctx->rx_ts = t; - nanosleep(&s, NULL); + if (ctx->rx_ece == 0 && ctx->rx_acc == 0 && ecn == 0) + return false; + + /* Onset, or ~4 windows of silence: emit fresh, undiluted. */ + if (mb_ecn_rcv_fresh(ctx, dt)) { + ctx->rx_win = t; + ctx->rx_acc = 0; + ctx->rx_byt = len; + ctx->rx_cap = cap; /* fresh, seeds the new window */ + ctx->rx_ece = (uint16_t) (ecn << CA_SHFT); + *ece = ctx->rx_ece; + *fcap = ctx->rx_cap; + return true; } + + /* 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; + + 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) + return false; + } + + /* Time-integral mean over the actual window elapsed (never rx_tw). */ + ctx->rx_ece = (uint16_t) ((ctx->rx_acc << CA_SHFT) / win); + + if (ctx->rx_byt > 0) + mb_ecn_resize(ctx, win); + + *fcap = ctx->rx_cap; + + ctx->rx_win = t; + ctx->rx_acc = 0; + ctx->rx_byt = 0; + ctx->rx_cap = 0; /* the next window starts unknown */ + + *ece = ctx->rx_ece; + + return true; } bool mb_ecn_ctx_update_rcv(void * _ctx, size_t len, uint8_t ecn, - uint16_t * ece) + uint8_t cap, + uint16_t * ece, + uint8_t * fcap) { - struct mb_ecn_ctx* ctx = _ctx; - bool update; + struct timespec now; + struct mb_ecn_ctx * ctx = _ctx; - (void) len; + clock_gettime(PTHREAD_COND_CLOCK, &now); - if ((ctx->rx_ece | ecn) == 0) - return false; + return mb_ecn_rcv(ctx, len, ecn, cap, ece, fcap, TS_TO_UINT64(now)); +} - if (ecn == 0) { /* End of congestion */ - ctx->rx_ece >>= 2; - update = ctx->rx_ece == 0; - } else { - if (ctx->rx_ece == 0) { /* Start of congestion */ - ctx->rx_ece = ecn; - ctx->rx_ctr = 0; - update = true; - } else { /* Congestion update */ - ctx->rx_ece -= ctx->rx_ece >> CA_SHFT; - ctx->rx_ece += ecn; - update = (ctx->rx_ctr++ & (CA_UPD - 1)) == true; - } +static void mb_ecn_ece(struct mb_ecn_ctx * ctx, + uint16_t ece, + uint8_t cap, + uint64_t t) +{ + uint64_t tgt; + + 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) + ctx->ai_hold = false; + + ctx->last_fb = t; + ctx->last_sig = t; + ctx->n_fb++; + + /* Scale the floor and AI slope to the path bottleneck. */ + if (cap != 0) { + tgt = cap_dec(cap) >> CA_CAP_SHFT; + if (tgt < CA_RATE_MIN) + tgt = CA_RATE_MIN; + + if (tgt > CA_RMIN_MAX) + tgt = CA_RMIN_MAX; + + if (tgt > ctx->rate_min) + ctx->rate_min += (tgt - ctx->rate_min) + >> CA_CAP_SM_SHFT; + else + ctx->rate_min -= (ctx->rate_min - tgt) + >> CA_CAP_SM_SHFT; + + ctx->ai_rate = ctx->rate_min; + ctx->tx_cap = cap; + ctx->last_cap = t; + ctx->n_cap++; } - *ece = ctx->rx_ece; + /* Control from the feedback path: a starved sender recovers. */ + if (t - ctx->last_ctrl < (uint64_t) CA_DT_CTRL) + return; - return update; -} + mb_ecn_ctrl(ctx, t - ctx->last_ctrl); + ctx->last_ctrl = t; +} void mb_ecn_ctx_update_ece(void * _ctx, - uint16_t ece) + uint16_t ece, + uint8_t cap) +{ + struct timespec now; + struct mb_ecn_ctx * ctx = _ctx; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + mb_ecn_ece(ctx, ece, cap, TS_TO_UINT64(now)); +} + +/* Due when the path stayed quiet for a heartbeat interval; arms the gap. */ +bool mb_ecn_ctx_hb_due(void * _ctx, + uint64_t now) +{ + struct mb_ecn_ctx * ctx = _ctx; + uint64_t t_hb = mb_ecn_t_hb(ctx); + uint64_t last; + + last = ctx->last_sig > ctx->last_hb ? ctx->last_sig : ctx->last_hb; + if (now - last < t_hb) + return false; + + ctx->last_hb = now; + + return true; +} + +/* Fold a heartbeat RTT sample into the ramp clock; also counts as life. */ +void mb_ecn_ctx_rtt(void * _ctx, + uint64_t now, + uint64_t rtt) { - struct mb_ecn_ctx* ctx = _ctx; + struct mb_ecn_ctx * ctx = _ctx; + uint64_t tgt; + + tgt = (uint64_t) CA_SS_RTT_MUL * rtt; + if (tgt < (uint64_t) CA_SS_TC_MIN) /* track the true RTT both */ + tgt = CA_SS_TC_MIN; /* ways: overshoot ~e^{1/2} */ + + if (tgt > (uint64_t) CA_SS_TC_MAX) /* at the real RTT, not the */ + tgt = CA_SS_TC_MAX; /* declared worst case */ + + ctx->ss_tc += (tgt >> CA_RTT_SHFT) - (ctx->ss_tc >> CA_RTT_SHFT); - ctx->tx_ece = ece; - ctx->tx_ctr = 0; - ctx->tx_cav = true; + ctx->last_sig = now; /* liveness only: never ages the ece signal */ + ctx->n_rtt++; } -int mb_ecn_calc_ecn(int fd, +int mb_ecn_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, size_t len) { - size_t q; + size_t q; + uint8_t mark; (void) len; (void) qc; - q = ipcp_flow_queued(fd); + /* Saturate: a queue past 255 quanta must not wrap to a low mark. */ + q = queued / CA_MARK_Q; + mark = q > 255 ? (uint8_t) 255 : (uint8_t) q; - *ecn |= (uint8_t) (q >> ECN_Q_SHFT); + if (mark > *ecn) + *ecn = mark; return 0; } @@ -262,35 +902,71 @@ ssize_t mb_ecn_print_stats(void * _ctx, char * buf, size_t len) { - struct mb_ecn_ctx* ctx = _ctx; - char * regime; + struct mb_ecn_ctx * ctx = _ctx; + char * regime; + uint64_t rate; + uint64_t peak; + int code; + uint16_t m; - if (len < 1024) + if (len < CA_STATS_STRLEN) return 0; - if (!ctx->tx_cav) + /* No signal seen: the rate is unconstrained drift, not a target. */ + rate = ctx->tx_cav ? ctx->rate : 0; + peak = ctx->tx_cav ? ctx->ss_peak : 0; + + /* Match the controller: MD fires on m, incl. the local fallback. */ + m = ctx->tx_ece > 0 ? ctx->tx_ece + : (uint16_t) (ctx->tx_loc << CA_SHFT); + + if (!ctx->tx_cav) { regime = "Slow start"; - else if (ctx->tx_ece) - regime = "Multiplicative dec"; - else + code = 0; + } else if (ctx->ai_hold) { + regime = "Loss recovery"; + code = 4; + } else if (ctx->src_limited) { + regime = "Source limited"; + code = 3; + } else if (m > 0) { + regime = "Proportional dec"; + code = 2; + } else { regime = "Additive inc"; + code = 1; + } sprintf(buf, "Congestion avoidance algorithm: %20s\n" "Upstream congestion level: %20u\n" - "Upstream packet counter: %20zu\n" "Downstream congestion level: %20u\n" - "Downstream packet counter: %20zu\n" - "Congestion window size (ns): %20" PRIu64 "\n" - "Packets in this window: %20zu\n" - "Bytes in this window: %20zu\n" - "Max bytes in this window: %20zu\n" - "Current congestion regime: %20s\n", + "Paced rate (bytes/s): %20" PRIu64 "\n" + "Pacer lead (bytes): %20" PRIu64 "\n" + "Congestion regime (code): %20d\n" + "Current congestion regime: %20s\n" + "Control steps (count): %20" PRIu64 "\n" + "Control time elapsed (ns): %20" PRIu64 "\n" + "Control time banked (ns): %20" PRIu64 "\n" + "Feedback updates (count): %20" PRIu64 "\n" + "Feedback timeouts (count): %20" PRIu64 "\n" + "Path capacity (bytes/s): %20" PRIu64 "\n" + "Capacity rate floor (bytes/s): %20" PRIu64 "\n" + "Capacity updates (count): %20" PRIu64 "\n" + "Slow start peak rate (bytes/s): %20" PRIu64 "\n" + "Signal-loss cuts (count): %20" PRIu64 "\n" + "Heartbeat RTT samples (count): %20" PRIu64 "\n" + "Ramp time constant (ns): %20" PRIu64 "\n", "Multi-bit ECN", - ctx->tx_ece, ctx->tx_ctr, - ctx->rx_ece, ctx->rx_ctr, (uint64_t) (1ULL << ctx->tx_mul), - ctx->tx_wpc, ctx->tx_wbc, ctx->tx_wbl, - regime); + ctx->tx_ece, + ctx->rx_ece, + rate, ctx->lead, code, + regime, + ctx->n_ctrl, ctx->t_ctrl, ctx->t_bank, + ctx->n_fb, ctx->n_ttl, + cap_dec(ctx->tx_cap), ctx->rate_min, ctx->n_cap, + peak, + ctx->n_loss, ctx->n_rtt, ctx->ss_tc); return strlen(buf); } diff --git a/src/ipcpd/unicast/ca/mb-ecn.h b/src/ipcpd/unicast/ca/mb-ecn.h index 1be27764..781e25a8 100644 --- a/src/ipcpd/unicast/ca/mb-ecn.h +++ b/src/ipcpd/unicast/ca/mb-ecn.h @@ -25,24 +25,36 @@ #include "ops.h" +void mb_ecn_init(uint32_t rtt_ms); + void * mb_ecn_ctx_create(void); void mb_ecn_ctx_destroy(void * ctx); -ca_wnd_t mb_ecn_ctx_update_snd(void * ctx, - size_t len); +time_t mb_ecn_ctx_update_snd(void * ctx, + size_t len, + uint8_t lecn, + uint64_t * ftag); bool mb_ecn_ctx_update_rcv(void * ctx, size_t len, uint8_t ecn, - uint16_t * ece); + uint8_t cap, + uint16_t * ece, + uint8_t * fcap); void mb_ecn_ctx_update_ece(void * ctx, - uint16_t ece); + uint16_t ece, + uint8_t cap); + +bool mb_ecn_ctx_hb_due(void * ctx, + uint64_t now); -void mb_ecn_wnd_wait(ca_wnd_t wnd); +void mb_ecn_ctx_rtt(void * ctx, + uint64_t now, + uint64_t rtt); -int mb_ecn_calc_ecn(int fd, +int mb_ecn_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, size_t len); diff --git a/src/ipcpd/unicast/ca/nop.c b/src/ipcpd/unicast/ca/nop.c index e5cacf66..e6965297 100644 --- a/src/ipcpd/unicast/ca/nop.c +++ b/src/ipcpd/unicast/ca/nop.c @@ -30,8 +30,8 @@ struct ca_ops nop_ca_ops = { .ctx_update_snd = nop_ctx_update_snd, .ctx_update_rcv = nop_ctx_update_rcv, .ctx_update_ece = nop_ctx_update_ece, - .wnd_wait = nop_wnd_wait, .calc_ecn = nop_calc_ecn, + .marks_ecn = false, .print_stats = NULL }; @@ -45,51 +45,52 @@ void nop_ctx_destroy(void * ctx) (void) ctx; } -ca_wnd_t nop_ctx_update_snd(void * ctx, - size_t len) +time_t nop_ctx_update_snd(void * ctx, + size_t len, + uint8_t lecn, + uint64_t * ftag) { - ca_wnd_t wnd; - (void) ctx; (void) len; + (void) lecn; + (void) ftag; - memset(&wnd, 0, sizeof(wnd)); - - return wnd; -} - -void nop_wnd_wait(ca_wnd_t wnd) -{ - (void) wnd; + return 0; } bool nop_ctx_update_rcv(void * ctx, size_t len, uint8_t ecn, - uint16_t * ece) + uint8_t cap, + uint16_t * ece, + uint8_t * fcap) { (void) ctx; (void) len; (void) ecn; + (void) cap; (void) ece; + (void) fcap; return false; } void nop_ctx_update_ece(void * ctx, - uint16_t ece) + uint16_t ece, + uint8_t cap) { (void) ctx; (void) ece; + (void) cap; } -int nop_calc_ecn(int fd, +int nop_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, size_t len) { - (void) fd; + (void) queued; (void) len; (void) ecn; (void) qc; diff --git a/src/ipcpd/unicast/ca/nop.h b/src/ipcpd/unicast/ca/nop.h index 8b892e61..6ca206df 100644 --- a/src/ipcpd/unicast/ca/nop.h +++ b/src/ipcpd/unicast/ca/nop.h @@ -29,20 +29,23 @@ void * nop_ctx_create(void); void nop_ctx_destroy(void * ctx); -ca_wnd_t nop_ctx_update_snd(void * ctx, - size_t len); +time_t nop_ctx_update_snd(void * ctx, + size_t len, + uint8_t lecn, + uint64_t * ftag); bool nop_ctx_update_rcv(void * ctx, size_t len, uint8_t ecn, - uint16_t * ece); + uint8_t cap, + uint16_t * ece, + uint8_t * fcap); void nop_ctx_update_ece(void * ctx, - uint16_t ece); - -void nop_wnd_wait(ca_wnd_t wnd); + uint16_t ece, + uint8_t cap); -int nop_calc_ecn(int fd, +int nop_calc_ecn(size_t queued, uint8_t * ecn, qoscube_t qc, size_t len); diff --git a/src/ipcpd/unicast/ca/ops.h b/src/ipcpd/unicast/ca/ops.h index 6d2ddf1d..b01e8573 100644 --- a/src/ipcpd/unicast/ca/ops.h +++ b/src/ipcpd/unicast/ca/ops.h @@ -30,24 +30,38 @@ struct ca_ops { void (* ctx_destroy)(void * ctx); - ca_wnd_t (* ctx_update_snd)(void * ctx, - size_t len); + time_t (* ctx_update_snd)(void * ctx, + size_t len, + uint8_t lecn, + uint64_t * ftag); bool (* ctx_update_rcv)(void * ctx, size_t len, uint8_t ecn, - uint16_t * ece); + uint8_t cap, + uint16_t * ece, + uint8_t * fcap); void (* ctx_update_ece)(void * ctx, - uint16_t ece); + uint16_t ece, + uint8_t cap); + + /* Optional, can be NULL: heartbeat pacing + RTT feedback. */ + bool (* ctx_hb_due)(void * ctx, + uint64_t now); - void (* wnd_wait)(ca_wnd_t wnd); + void (* ctx_rtt)(void * ctx, + uint64_t now, + uint64_t rtt); - int (* calc_ecn)(int fd, + int (* calc_ecn)(size_t queued, uint8_t * ecn, qoscube_t qc, size_t len); + /* True if calc_ecn inspects the queue; gates the lookup. */ + bool marks_ecn; + /* Optional, can be NULL */ ssize_t (* print_stats)(void * ctx, char * buf, diff --git a/src/ipcpd/unicast/ca/tests/CMakeLists.txt b/src/ipcpd/unicast/ca/tests/CMakeLists.txt new file mode 100644 index 00000000..6e42163d --- /dev/null +++ b/src/ipcpd/unicast/ca/tests/CMakeLists.txt @@ -0,0 +1,44 @@ +get_filename_component(CURRENT_SOURCE_PARENT_DIR + ${CMAKE_CURRENT_SOURCE_DIR} DIRECTORY) +get_filename_component(CURRENT_BINARY_PARENT_DIR + ${CMAKE_CURRENT_BINARY_DIR} DIRECTORY) + +get_filename_component(UNICAST_SOURCE_DIR ${CURRENT_SOURCE_PARENT_DIR} DIRECTORY) +get_filename_component(UNICAST_BINARY_DIR ${CURRENT_BINARY_PARENT_DIR} DIRECTORY) + +get_filename_component(PARENT_PATH ${CMAKE_CURRENT_SOURCE_DIR} DIRECTORY) +get_filename_component(PARENT_DIR ${PARENT_PATH} NAME) + +compute_test_prefix() + +create_test_sourcelist(${PARENT_DIR}_tests test_suite.c + # Add new tests here + mb_ecn_test.c + ca_test.c + ) + +add_executable(${PARENT_DIR}_test ${${PARENT_DIR}_tests} + ${UNICAST_SOURCE_DIR}/ca.c + ${UNICAST_SOURCE_DIR}/cap.c + ${CURRENT_SOURCE_PARENT_DIR}/nop.c + ) + +target_include_directories(${PARENT_DIR}_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}_test) +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}) diff --git a/src/ipcpd/unicast/ca/tests/ca_test.c b/src/ipcpd/unicast/ca/tests/ca_test.c new file mode 100644 index 00000000..1b86eab8 --- /dev/null +++ b/src/ipcpd/unicast/ca/tests/ca_test.c @@ -0,0 +1,392 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Unit tests for the congestion-avoidance interface + * + * 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 "config.h" + +#include "ca.h" + +#include <test/test.h> + +#define ADDR_A 0x1111ULL +#define ADDR_B 0x2222ULL + +static const struct { + enum pol_cong_avoid pol; + const char * name; +} ca_pols[] = { + { CA_NONE, "none" }, + { CA_MB_ECN, "mb-ecn" } +}; + +#define CA_POLS (sizeof(ca_pols) / sizeof(ca_pols[0])) + +static int test_ca_init_fini(enum pol_cong_avoid pol, + const char * name) +{ + TEST_START("(%s)", name); + + if (ca_init(pol, 100) < 0) { + printf("Failed to init ca for %s.\n", name); + goto fail; + } + + ca_fini(); + + TEST_SUCCESS("(%s)", name); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL("(%s)", name); + return TEST_RC_FAIL; +} + +static int test_ca_init_fini_all(void) +{ + int ret = 0; + size_t i; + + for (i = 0; i < CA_POLS; i++) + ret |= test_ca_init_fini(ca_pols[i].pol, ca_pols[i].name); + + return ret; +} + +static int test_ca_init_invalid(void) +{ + TEST_START(); + + if (ca_init(CA_INVALID, 100) == 0) { + printf("Init accepted an invalid policy.\n"); + ca_fini(); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_ca_ctx_share(enum pol_cong_avoid pol, + const char * name) +{ + void * c1; + void * c2; + + TEST_START("(%s)", name); + + if (ca_init(pol, 100) < 0) { + printf("Failed to init ca for %s.\n", name); + goto fail; + } + + c1 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); + if (c1 == NULL) { + printf("Failed to get ctx.\n"); + goto fail_init; + } + + c2 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); + if (c2 == NULL) { + printf("Failed to get second ctx.\n"); + goto fail_c1; + } + +#ifdef IPCP_CA_PER_FLOW + if (c1 == c2) { + printf("Per-flow build shared a ctx across flows.\n"); + goto fail_c2; + } +#else + if (c1 != c2) { + printf("Aggregate build did not share ctx per (addr, qc).\n"); + goto fail_c2; + } +#endif + ca_ctx_put(c2); + ca_ctx_put(c1); + + ca_fini(); + + TEST_SUCCESS("(%s)", name); + + return TEST_RC_SUCCESS; + fail_c2: + ca_ctx_put(c2); + fail_c1: + ca_ctx_put(c1); + fail_init: + ca_fini(); + fail: + TEST_FAIL("(%s)", name); + return TEST_RC_FAIL; +} + +static int test_ca_ctx_share_all(void) +{ + int ret = 0; + size_t i; + + for (i = 0; i < CA_POLS; i++) + ret |= test_ca_ctx_share(ca_pols[i].pol, ca_pols[i].name); + + return ret; +} + +static int test_ca_ctx_distinct(void) +{ + void * a_be; + void * b_be; + void * a_video; + + TEST_START(); + + if (ca_init(CA_NONE, 100) < 0) { + printf("Failed to init ca.\n"); + goto fail; + } + + a_be = ca_ctx_get(ADDR_A, QOS_CUBE_BE); + if (a_be == NULL) { + printf("Failed to get ctx.\n"); + goto fail_init; + } + + b_be = ca_ctx_get(ADDR_B, QOS_CUBE_BE); + if (b_be == NULL) { + printf("Failed to get ctx.\n"); + goto fail_a_be; + } + + a_video = ca_ctx_get(ADDR_A, QOS_CUBE_VIDEO); + if (a_video == NULL) { + printf("Failed to get ctx.\n"); + goto fail_b_be; + } + + if (a_be == b_be) { + printf("Distinct addresses shared a ctx.\n"); + goto fail_a_video; + } + + if (a_be == a_video) { + printf("Distinct qos cubes shared a ctx.\n"); + goto fail_a_video; + } + + ca_ctx_put(a_video); + ca_ctx_put(b_be); + ca_ctx_put(a_be); + + ca_fini(); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_a_video: + ca_ctx_put(a_video); + fail_b_be: + ca_ctx_put(b_be); + fail_a_be: + ca_ctx_put(a_be); + fail_init: + ca_fini(); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* Refcount survival is an aggregate-only property. */ +#ifndef IPCP_CA_PER_FLOW +static int test_ca_ctx_refcount(void) +{ + void * c1; + void * c3; + + TEST_START(); + + if (ca_init(CA_NONE, 100) < 0) { + printf("Failed to init ca.\n"); + goto fail; + } + + c1 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); /* refs = 1 */ + if (c1 == NULL) { + printf("Failed to get ctx.\n"); + goto fail_init; + } + + if (ca_ctx_get(ADDR_A, QOS_CUBE_BE) == NULL) { /* refs = 2 */ + printf("Failed to get second ref.\n"); + goto fail_c1; + } + + ca_ctx_put(c1); /* refs = 1 */ + + c3 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); /* refs = 2 */ + if (c3 == NULL) { + printf("Failed to get third ref.\n"); + goto fail_c1; + } + + if (c3 != c1) { + printf("Refcounted ctx freed while still referenced.\n"); + goto fail_c3; + } + + ca_ctx_put(c3); + ca_ctx_put(c1); + + ca_fini(); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_c3: + ca_ctx_put(c3); + fail_c1: + ca_ctx_put(c1); + fail_init: + ca_fini(); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* The last put frees the interned ctx; a fresh get recreates it. */ +static int test_ca_ctx_recreate(void) +{ + void * c1; + void * c2; + void * c3; + + TEST_START(); + + if (ca_init(CA_NONE, 100) < 0) { + printf("Failed to init ca.\n"); + goto fail; + } + + c1 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); /* refs = 1 */ + if (c1 == NULL) { + printf("Failed to get ctx.\n"); + goto fail_init; + } + + ca_ctx_put(c1); /* refs = 0: freed and de-interned */ + + c2 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); /* fresh entry */ + if (c2 == NULL) { + printf("Get after release did not recreate.\n"); + goto fail_init; + } + + c3 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); /* refs = 2: shares */ + if (c3 == NULL) { + printf("Failed to share recreated ctx.\n"); + goto fail_c2; + } + + if (c3 != c2) { + printf("Recreated ctx did not intern.\n"); + goto fail_c3; + } + + ca_ctx_put(c3); + ca_ctx_put(c2); + + ca_fini(); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_c3: + ca_ctx_put(c3); + fail_c2: + ca_ctx_put(c2); + fail_init: + ca_fini(); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* ca_fini drains a ctx a flow left interned, with no leak. */ +static int test_ca_fini_drains(void) +{ + void * c1; + void * c2; + + TEST_START(); + + if (ca_init(CA_NONE, 100) < 0) { + printf("Failed to init ca.\n"); + goto fail; + } + + c1 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); /* refs = 1 */ + if (c1 == NULL) { + printf("Failed to get ctx.\n"); + goto fail_init; + } + + c2 = ca_ctx_get(ADDR_A, QOS_CUBE_BE); /* refs = 2 */ + if (c2 == NULL) { + printf("Failed to get second ref.\n"); + goto fail_init; + } + + /* Leave both refs live: ca_fini must drain and free the ctx. */ + ca_fini(); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_init: + ca_fini(); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} +#endif /* !IPCP_CA_PER_FLOW */ + +int ca_test(int argc, + char ** argv) +{ + int ret = 0; + + (void) argc; + (void) argv; + + ret |= test_ca_init_fini_all(); + ret |= test_ca_init_invalid(); + ret |= test_ca_ctx_share_all(); + ret |= test_ca_ctx_distinct(); +#ifndef IPCP_CA_PER_FLOW + ret |= test_ca_ctx_refcount(); + ret |= test_ca_ctx_recreate(); + ret |= test_ca_fini_drains(); +#endif + return ret; +} diff --git a/src/ipcpd/unicast/ca/tests/mb_ecn_test.c b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c new file mode 100644 index 00000000..4bbc12aa --- /dev/null +++ b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c @@ -0,0 +1,2923 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Unit tests 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; +} + +/* + * Drive ctx as a fully backlogged flow: offer a packet every paced + * wait, so the offered load tracks the paced rate. Returns end time. + */ +static uint64_t drive_backlogged(struct mb_ecn_ctx * ctx, + uint64_t * ftag, + uint64_t t, + uint64_t dur, + size_t len) +{ + uint64_t end = t + dur; + time_t w; + + while (t < end) { + w = mb_ecn_snd(ctx, len, t, ftag); + t += w > 0 ? (uint64_t) w : 1; + } + + return t; +} + +static int test_mb_ecn_ctx_create_destroy(void) +{ + struct mb_ecn_ctx * ctx; + + TEST_START(); + + ctx = mb_ecn_ctx_create(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + if (ctx->rate != CA_RATE_INIT) { + printf("Bad initial rate %" PRIu64 ".\n", ctx->rate); + goto fail_ctx; + } + + if (ctx->rate_min != CA_RATE_MIN) { + printf("Bad initial floor %" PRIu64 ".\n", ctx->rate_min); + goto fail_ctx; + } + + if (ctx->vt != 0) { + printf("Bad initial virtual clock %" PRIu64 ".\n", ctx->vt); + goto fail_ctx; + } + + if (ctx->tx_cav) { + printf("Context did not start in slow start.\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; +} + +static int test_mb_ecn_calc_ecn(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; + } + + /* 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); + 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; + } + + 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); + 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); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + 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) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + if (!mb_ecn_rcv(ctx, LEN, 4, 0, &ece, &fcap, MS)) { + printf("Onset did not update.\n"); + goto fail_ctx; + } + + if (ece != 4 << CA_SHFT) { + printf("Onset ece: exp %u, got %u.\n", 4 << CA_SHFT, ece); + goto fail_ctx; + } + + if (mb_ecn_rcv(ctx, LEN, 4, 0, &ece, &fcap, 2 * MS)) { + printf("Mid-window packet updated.\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 50% duty mark square wave emits the time mean, not the last peak. */ +static int test_mb_ecn_rcv_window_mean(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + size_t upd; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + 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 (upd != 2) { + printf("%zu updates in two windows.\n", upd); + goto fail_ctx; + } + + if (ece < 120 || ece > 136) { + printf("window mean: exp ~128, got %u.\n", ece); + 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; +} + +/* + * Two flows on the same wall-clock mark timeline, 15x apart in byte + * rate: the same congestion estimate, but the faster flow's window is + * shorter, so it feeds back more often (cadence tracks byte rate). + */ +static int test_mb_ecn_rcv_rate_independent(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint16_t ea; + uint16_t eb; + uint8_t fcap; + size_t ua; + size_t ub; + size_t i; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + ea = 0; + eb = 0; + ua = 0; + ub = 0; + + /* 300 ms of sustained mark 8; a at 1 kpps, b at ~66 pps. */ + for (i = 1; i <= 300; i++) { + ua += mb_ecn_rcv(a, LEN, 8, 0, &ea, &fcap, i * MS) ? 1 : 0; + if (i % 15 != 0) + continue; + + ub += mb_ecn_rcv(b, LEN, 8, 0, &eb, &fcap, i * MS) ? 1 : 0; + } + + if (ea > eb + 32 || eb > ea + 32) { + printf("estimates diverge: %u vs %u.\n", ea, eb); + goto fail_ctx; + } + + if (ua < ub + 2) { + printf("cadence not rate-scaled: %zu vs %zu.\n", ua, ub); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * Two flows on one bottleneck, equal byte rate but 7.5x apart in + * packet size: the same congestion estimate and the same feedback + * cadence. Framing does not skew the control signal (fair share). + */ +static int test_mb_ecn_rcv_size_fair(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint16_t ea; + uint16_t eb; + uint8_t fcap; + size_t ua; + size_t ub; + uint64_t ta; + uint64_t tb; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + ea = 0; + eb = 0; + ua = 0; + ub = 0; + ta = 0; + tb = 0; + + /* 1 MB/s each: a at 200 B / 200 us, b at 1500 B / 1.5 ms. */ + while (ta < 500 * MS) { + ta += 200 * 1000; + ua += mb_ecn_rcv(a, 200, 8, 0, &ea, &fcap, ta) ? 1 : 0; + } + + while (tb < 500 * MS) { + tb += 1500 * 1000; + ub += mb_ecn_rcv(b, 1500, 8, 0, &eb, &fcap, tb) ? 1 : 0; + } + + if (ea != 8 << CA_SHFT || eb != 8 << CA_SHFT) { + printf("size-skewed estimate: %u vs %u.\n", ea, eb); + goto fail_ctx; + } + + if (ua > ub + 3 || ub > ua + 3) { + printf("cadence skewed by size: %zu vs %zu.\n", ua, ub); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* Release emits exactly one 0 and leaves the estimator fully idle. */ +static int test_mb_ecn_rcv_release_exact_zero(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + size_t ends; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + for (i = 1; i <= 140; i++) + mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, i * MS); + + ends = 0; + for (i = 141; i <= 350; i++) { + if (!mb_ecn_rcv(ctx, LEN, 0, 0, &ece, &fcap, i * MS)) + continue; + + if (ece == 0) + ends++; + } + + if (ends != 1) { + printf("end of congestion fired %zu times.\n", ends); + goto fail_ctx; + } + + if (ctx->rx_ece != 0 || ctx->rx_acc != 0) { + printf("estimator not idle: ece %u acc %" PRIu64 ".\n", + ctx->rx_ece, ctx->rx_acc); + 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 gap past the window restarts fresh: no stale, diluted estimate. */ +static int test_mb_ecn_rcv_gap_restart(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + uint64_t t; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + 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; + if (!mb_ecn_rcv(ctx, LEN, 5, 0, &ece, &fcap, t)) { + printf("gap restart did not update.\n"); + goto fail_ctx; + } + + if (ece != 5 << CA_SHFT) { + printf("gap restart: exp %u, got %u.\n", 5 << CA_SHFT, ece); + goto fail_ctx; + } + + t += 10 * CA_TW_INIT; + 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; + } + + if (mb_ecn_rcv(ctx, LEN, 0, 0, &ece, &fcap, t + MS)) { + printf("idle packet updated.\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; +} + +/* 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; + uint64_t t; + + TEST_START(); + + 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"); + 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"); + goto fail_ctx; + } + + 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; + fail_ctx: + mb_ecn_ctx_destroy(ctx); + fail: + 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) +{ + 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; + } + + /* 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; + } + } + + if (count < 8 || count > 32) { + printf("rate %" PRIu64 ": %zu pkts/window.\n", + rates[ri], count); + goto fail_ctx; + } + + if (last_ece < 224 || last_ece > 288) { + printf("rate %" PRIu64 ": ece %u ~256.\n", + rates[ri], last_ece); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(ctx); + ctx = NULL; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(ctx); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * The window floors at CA_TW_MIN, tracks the rate below the old knee, + * and only a pathological fold hits the CA_TW_ABSMAX ceiling. + */ +static int test_mb_ecn_rcv_window_clip_bounds(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + uint64_t ia; + uint64_t t; + size_t closes; + + TEST_START(); + + /* 1 GbE is above the high knee: the window floors at CA_TW_MIN. */ + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ia = 8000ULL * BILLION / 1000000000ULL; + t = 0; + closes = 0; + while (closes < 40) { + t += ia; + if (mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t)) + closes++; + } + + if (ctx->rx_tw != CA_TW_MIN) { + printf("high-rate window: exp %" PRIu64 ", got %" PRIu64 + ".\n", (uint64_t) CA_TW_MIN, ctx->rx_tw); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(ctx); + + /* 1 Mbps: past the old knee, ~16 pkts = 16 * 8 ms = 131 ms. */ + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ia = 8000ULL * BILLION / 1000000ULL; + t = 0; + closes = 0; + while (closes < 40) { + t += ia; + if (mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t)) + closes++; + } + + if (ctx->rx_tw < 120 * MS || ctx->rx_tw > 140 * MS) { + printf("low-rate window: exp ~131 ms, got %" PRIu64 ".\n", + ctx->rx_tw); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(ctx); + + /* A near-empty window folds a huge target: ceiling holds. */ + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + mb_ecn_rcv(ctx, 10, 8, 0, &ece, &fcap, MS); + mb_ecn_rcv(ctx, 10, 8, 0, &ece, &fcap, MS + CA_TW_INIT); + + if (ctx->rx_tw != CA_TW_ABSMAX) { + printf("window ceiling breached: %" PRIu64 ".\n", + ctx->rx_tw); + 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 CA-limited slow flow grows its window to hold ~16 packets. */ +static int test_mb_ecn_rcv_slow_window(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* 1400 B every 171 ms (~8 KB/s), sustained mark 8. */ + for (i = 1; i <= 100; i++) + mb_ecn_rcv(ctx, 1400, 8, 0, &ece, &fcap, i * 171 * MS); + + /* Target window 16 * 1000 B at 8187 B/s ~= 2.0 s. */ + if (ctx->rx_tw < 1400 * MS || ctx->rx_tw > 2800 * MS) { + printf("slow window: exp ~2 s, got %" PRIu64 ".\n", + ctx->rx_tw); + goto fail_ctx; + } + + /* Steady mark 8 emits exactly 256 once the window settles. */ + if (ece != 8 << CA_SHFT) { + printf("slow-flow estimate: exp %u, got %u.\n", + 8 << CA_SHFT, ece); + 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 physically maximal window must fold without overflow or wrap. */ +static int test_mb_ecn_rcv_no_overflow_highrate(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + bool ok; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* 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; + + ok = mb_ecn_rcv(ctx, LEN, 15, 0, &ece, &fcap, 2 * CA_TW_INIT - 1); + + if (!ok) { + printf("max-window close did not fire.\n"); + goto fail_ctx; + } + + if (ece > 8160) { + printf("estimate %u wrapped.\n", ece); + goto fail_ctx; + } + + /* A wrapped numerator drives rx_tw to MAX; it must descend. */ + if (ctx->rx_tw >= CA_TW_INIT || ctx->rx_tw < CA_TW_MIN) { + printf("window %" PRIu64 " did not descend.\n", ctx->rx_tw); + 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; +} + +/* + * The sender holds a mark across the full inter-feedback gap (TTL > + * 2 * CA_TW_INIT); a repeated mark adds only the proportional cut. + */ +static int test_mb_ecn_ece_ttl_covers_cadence(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t prev; + uint64_t t; + uint64_t ftag = 0; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = (uint64_t) 100 << 20; + ctx->tx_cav = true; + + 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. */ + t = MS; + for (i = 0; i < 4; i++) { + t += (2 * CA_TW_INIT + 5 * MS) / 4; + mb_ecn_snd(ctx, LEN, t, &ftag); + if (ctx->tx_ece == 0) { + printf("mark cleared inside the feedback gap.\n"); + goto fail_ctx; + } + } + + /* Same mark again: rise 0, so only the proportional cut. */ + prev = ctx->rate; + t += MS; + mb_ecn_ece(ctx, 100, 0, t); + mb_ecn_snd(ctx, LEN, t, &ftag); + + if (prev - ctx->rate > prev / 100) { + printf("phantom lead cut: %" PRIu64 " -> %" PRIu64 ".\n", + prev, 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_slow_start(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t prev; + uint64_t t; + uint64_t ftag = 0; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + prev = ctx->rate; + t = 0; + + /* No feedback: the flow stays in slow start and grows each step. */ + for (i = 0; i < 16; i++) { + t += MS; + mb_ecn_snd(ctx, LEN, t, &ftag); + if (ctx->rate <= prev) { + printf("rate did not grow: %" PRIu64 ".\n", ctx->rate); + goto fail_ctx; + } + + prev = ctx->rate; + } + + /* Exponential ramp doubles in ~ln2 * CA_SS_TC ~= 14 ms. */ + if (ctx->rate < 2 * CA_RATE_INIT) { + printf("slow start too slow: %" 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_dt_scaling_invariant(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint64_t inc_a; + uint64_t inc_b; + uint64_t t; + uint64_t fta = 0; + uint64_t ftb = 0; + size_t i; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + /* Leave slow start; seed a realistic rate (truncation-free). */ + mb_ecn_ece(a, 0, 0, 0); + 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). */ + mb_ecn_snd(a, LEN, 30 * MS, &fta); + + /* b: thirty 1 ms steps over the same 30 ms. */ + t = 0; + for (i = 0; i < 30; i++) { + t += MS; + mb_ecn_snd(b, LEN, t, &ftb); + } + + inc_a = a->rate - ((uint64_t) 10 << 20); + inc_b = b->rate - ((uint64_t) 10 << 20); + + /* Equal within 1 %; the small gap is per-step integer truncation. */ + if (inc_a == 0 || inc_b == 0) { + printf("no additive increase: %" PRIu64 " %" PRIu64 ".\n", + inc_a, inc_b); + goto fail_ctx; + } + + if (inc_a > inc_b + inc_a / 100 || inc_b > inc_a + inc_a / 100) { + printf("cadence-dependent AI: %" PRIu64 " vs %" PRIu64 ".\n", + inc_a, inc_b); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_mb_ecn_multiplicative_decrease(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t prev; + uint64_t t; + uint64_t ftag = 0; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = (uint64_t) 100 << 20; + prev = ctx->rate; + t = 0; + + for (i = 0; i < 10; i++) { + t += MS; + mb_ecn_ece(ctx, CA_ECE_REF, 0, t); + mb_ecn_snd(ctx, LEN, t, &ftag); + if (ctx->rate >= prev) { + printf("rate did not shrink: %" PRIu64 ".\n", + ctx->rate); + goto fail_ctx; + } + + prev = 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; +} + +static int test_mb_ecn_rate_floor(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* + * A starved 600 ms at full mark takes the rate/2 branch, far + * below the floor; stays inside the initial ~976 ms mark TTL. + */ + ctx->rate = CA_RATE_MIN + 1000; + mb_ecn_ece(ctx, CA_ECE_REF, 0, 0); + mb_ecn_snd(ctx, LEN, 600 * MS, &ftag); + + if (ctx->rate != CA_RATE_MIN) { + printf("rate floor breached: %" 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_fixed_point(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t exp; + uint64_t t; + uint64_t ftag = 0; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + exp = CA_AI_RATE * CA_ECE_REF / + (128 - CA_ECE_REF * BILLION / CA_PROBE_TC); + t = 0; + + /* ~20 s: the probe raises the loop time constant to CA_PROBE_TC. */ + for (i = 0; i < 20000; i++) { + t += MS; + mb_ecn_ece(ctx, 128, 0, t); + mb_ecn_snd(ctx, LEN, t, &ftag); + } + + if (ctx->rate < exp - exp / 4 || ctx->rate > exp + exp / 4) { + printf("no fixed point: exp ~%" PRIu64 ", got %" PRIu64 ".\n", + exp, 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; +} + +/* + * 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) +{ + struct mb_ecn_ctx * ctx; + uint64_t prev; + uint64_t drop; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + 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); + 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); + 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); + 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 local first-hop mark exits slow start with no feedback needed. */ +static int test_mb_ecn_slow_start_local_brake(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t prev; + uint64_t ftag = 0; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + for (i = 1; i <= 50; i++) + mb_ecn_snd(ctx, LEN, i * MS, &ftag); + + if (ctx->tx_cav) { + printf("Left slow start without any signal.\n"); + goto fail_ctx; + } + + prev = ctx->rate; + mb_ecn_loc(ctx, 1, 50 * MS); + if (!ctx->tx_cav) { + printf("Local mark did not exit slow start.\n"); + goto fail_ctx; + } + + mb_ecn_snd(ctx, LEN, 51 * MS, &ftag); + if (ctx->rate > prev + prev / 20) { + printf("SS ramp survived the brake: %" 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; +} + +/* A clean path still ramps to line rate in well under a second. */ +static int test_mb_ecn_slow_start_clean_ramp(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* Backlogged, no marks: slow start sprints in a couple windows. */ + drive_backlogged(ctx, &ftag, MS, 2 * CA_SND_WIN, LEN); + + if (ctx->rate < (1ULL << 24)) { + printf("backlogged slow start too slow: %" 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; +} + +/* + * A sender starved of send-path control steps recovers through the + * feedback path: honest elapsed time, at most a 50% cut per step. + */ +static int test_mb_ecn_starved_decrease_escape(void) +{ + struct mb_ecn_ctx * ctx; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = (uint64_t) 100 << 20; + ctx->tx_cav = true; + + /* Feedback arrives once per second; no sends at all. */ + for (i = 1; i <= 6; i++) + mb_ecn_ece(ctx, 480, 0, i * BILLION); + + if (ctx->rate > (5ULL << 19)) { + printf("still wedged at %" PRIu64 " B/s.\n", ctx->rate); + goto fail_ctx; + } + + if (ctx->rate < CA_RATE_MIN) { + printf("rate floor breached: %" 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; +} + +/* With feedback fully dead, the local mark alone recovers the rate. */ +static int test_mb_ecn_starved_local_fallback(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t t; + uint64_t ftag = 0; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = (uint64_t) 100 << 20; + + for (i = 1; i <= 7; i++) { + t = i * BILLION; + mb_ecn_loc(ctx, 15, t); + mb_ecn_snd(ctx, LEN, t, &ftag); + } + + if (!ctx->tx_cav) { + printf("Local mark did not exit slow start.\n"); + goto fail_ctx; + } + + if (ctx->rate > (5ULL << 19)) { + printf("still wedged at %" PRIu64 " B/s.\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; +} + +/* + * MD Δt-invariance: the same elapsed time under the same mark cuts + * the same, in one big step or five small ones. + */ +static int test_mb_ecn_decrease_dt_invariant(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint64_t cut_a; + uint64_t cut_b; + uint64_t r0; + uint64_t fta = 0; + uint64_t ftb = 0; + size_t i; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + r0 = (uint64_t) 100 << 20; + a->rate = r0; + b->rate = r0; + + mb_ecn_ece(a, 256, 0, 0); + mb_ecn_ece(b, 256, 0, 0); + + /* a: one 50 ms step; b: five 10 ms steps (both within DT_CAP). */ + mb_ecn_snd(a, LEN, 50 * MS, &fta); + + for (i = 1; i <= 5; i++) + mb_ecn_snd(b, LEN, i * 10 * MS, &ftb); + + cut_a = r0 - a->rate; + cut_b = r0 - b->rate; + + if (cut_a == 0 || cut_b == 0) { + printf("no cut: %" PRIu64 " %" PRIu64 ".\n", cut_a, cut_b); + goto fail_ctx; + } + + /* Within 10%: residual is Euler compounding of MD and the probe. */ + if (cut_a > cut_b + cut_a / 10 || cut_b > cut_a + cut_a / 10) { + printf("cadence-dependent MD: %" PRIu64 " vs %" PRIu64 + ".\n", cut_a, cut_b); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* Sub-ms control steps must not lose decrease time to truncation. */ +static int test_mb_ecn_decrease_subms_carry(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint64_t cut_a; + uint64_t cut_b; + uint64_t r0; + uint64_t fta = 0; + uint64_t ftb = 0; + size_t i; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + r0 = (uint64_t) 100 << 20; + a->rate = r0; + b->rate = r0; + + mb_ecn_ece(a, 256, 0, 0); + mb_ecn_ece(b, 256, 0, 0); + + /* Same 30 ms of marked time; b's steps have a 0.5 ms tail. */ + for (i = 1; i <= 10; i++) + mb_ecn_snd(a, LEN, i * 3 * MS, &fta); + + for (i = 1; i <= 20; i++) + mb_ecn_snd(b, LEN, i * 3 * MS / 2, &ftb); + + cut_a = r0 - a->rate; + cut_b = r0 - b->rate; + + if (cut_a == 0 || cut_b == 0) { + printf("no cut: %" PRIu64 " %" PRIu64 ".\n", cut_a, cut_b); + goto fail_ctx; + } + + /* Within 10%: the sub-ms remainder must carry, not vanish. */ + if (cut_a > cut_b + cut_a / 10 || cut_b > cut_a + cut_a / 10) { + printf("sub-ms decrease lost: %" PRIu64 " vs %" PRIu64 + ".\n", cut_a, cut_b); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* Resuming after a long idle gap: bounded AI, no cut from stale marks. */ +static int test_mb_ecn_idle_resume_bounded(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t prev; + uint64_t bump; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = (uint64_t) 10 << 20; + + mb_ecn_loc(ctx, 15, MS); + mb_ecn_ece(ctx, 480, 0, MS); + + prev = ctx->rate; + + /* 600 s later: both signals stale; one capped AI + probe step. */ + mb_ecn_snd(ctx, LEN, 600 * BILLION, &ftag); + + if (ctx->rate < prev) { + printf("stale mark cut the rate: %" PRIu64 ".\n", + ctx->rate); + goto fail_ctx; + } + + bump = CA_AI_RATE * CA_DT_CAP / BILLION; + bump += (prev + bump) * CA_DT_CAP / CA_PROBE_TC; + + if (ctx->rate > prev + bump + 2) { + printf("idle resume cap breached: %" 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; +} + +/* The congestion signal ages out on wall-clock time, not packet count. */ +static int test_mb_ecn_ece_staleness(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* A fast flow's control step caches the floor TTL. */ + ctx->rate = (uint64_t) 1 << 20; + ctx->tx_cav = true; + mb_ecn_snd(ctx, LEN, MS, &ftag); + + if (ctx->ece_ttl != CA_ECE_TTL) { + printf("Fast-flow TTL: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) CA_ECE_TTL, ctx->ece_ttl); + goto fail_ctx; + } + + mb_ecn_ece(ctx, CA_ECE_REF, 0, 2 * MS); + + /* Just inside the TTL: the signal is still held. */ + mb_ecn_snd(ctx, LEN, 2 * MS + ctx->ece_ttl, &ftag); + if (ctx->tx_ece == 0) { + printf("signal aged out too early.\n"); + goto fail_ctx; + } + + /* Past the TTL without feedback: the signal is cleared. */ + mb_ecn_snd(ctx, LEN, 2 * MS + ctx->ece_ttl + 1, &ftag); + if (ctx->tx_ece != 0) { + printf("stale signal not cleared: %u.\n", ctx->tx_ece); + 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; +} + +/* The staleness horizon stretches with a slow flow's window. */ +static int test_mb_ecn_ece_ttl_tracks_rate(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t want; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = 8192; + ctx->rate_min = 8192; + ctx->ai_rate = 0; + ctx->tx_cav = true; + mb_ecn_snd(ctx, 1400, MS, &ftag); + + want = (1 << CA_TW_GAP_SHFT) * CA_RX_WBYTES * BILLION / ctx->rate; + if (ctx->ece_ttl != want) { + printf("Slow-flow TTL: exp %" PRIu64 ", got %" PRIu64 ".\n", + want, ctx->ece_ttl); + goto fail_ctx; + } + + if (ctx->ece_ttl < 7 * (uint64_t) BILLION) { + printf("TTL did not stretch: %" PRIu64 ".\n", + ctx->ece_ttl); + 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; +} + +/* + * First packet of a flow starts at the clock; a same-instant second + * packet leads by its length and is paced by lead / rate. + */ +static int test_mb_ecn_sfq_pace(void) +{ + struct mb_ecn_ctx * ctx; + time_t wait; + uint64_t ftag = 0; + time_t want; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = 1U << 20; + ctx->inv_rate = mb_ecn_rate_inv(ctx->rate); + + /* First send: start tag equals the clock, so no wait. */ + wait = mb_ecn_snd(ctx, 1500, 0, &ftag); + if (wait != 0) { + printf("first packet waited %ld, expected 0.\n", (long) wait); + goto fail_ctx; + } + + /* Same instant (dt = 0): the flow now leads by 1500 B. */ + wait = mb_ecn_snd(ctx, 1500, 0, &ftag); + want = (time_t) ((uint64_t) 1500 * BILLION / ctx->rate); + + if (wait != want) { + printf("paced wait %ld, expected %ld.\n", + (long) wait, (long) want); + 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; +} + +/* + * The proportional probe grows the rate by the same fraction per unit + * time regardless of the absolute rate: two clean flows 100x apart in + * rate grow by the same ratio. Deleting the probe leaves only the tiny + * additive increase, failing the growth floor. + */ +static int test_mb_ecn_probe_scale_invariant(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint64_t ra0; + uint64_t rb0; + double ga; + double gb; + uint64_t t; + size_t i; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + /* Clean path (mark 0), out of slow start, backlogged, 100x apart. */ + mb_ecn_ece(a, 0, 0, 0); + 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; + rb0 = b->rate; + + /* Drive control via the feedback path so backlogged stays set. */ + t = 0; + for (i = 0; i < 500; i++) { + t += MS; + mb_ecn_ece(a, 0, 0, t); + mb_ecn_ece(b, 0, 0, t); + } + + ga = (double) a->rate / ra0; + gb = (double) b->rate / rb0; + + if (ga < gb - gb / 50 || gb < ga - ga / 50) { + printf("probe not scale-invariant: %.4f vs %.4f.\n", ga, gb); + goto fail_ctx; + } + + /* And it must actually grow: the probe is present, not deleted. */ + if (ga < 1.05) { + printf("probe did not grow the rate: %.4f.\n", ga); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * The proportional probe e-folds the rate over CA_PROBE_TC: a clean flow + * grows by ~e in 8 s. Pinned to a literal e-band so a mistuned + * CA_PROBE_TC (e.g. 4 s gives e^2) fails. + */ +static int test_mb_ecn_probe_time_constant(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t r0; + double ratio; + uint64_t t; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* 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; + + /* 8000 x 1 ms of clean growth, driven via the feedback path. */ + t = 0; + for (i = 0; i < 8000; i++) { + t += MS; + mb_ecn_ece(ctx, 0, 0, t); + } + + /* Washout damps the probe to ~4/3 TC, so 8 s -> ~2.12x. */ + 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); + 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; +} + +/* The fed-back capacity is the MIN of the nonzero caps in the window. */ +static int test_mb_ecn_rcv_cap_window_min(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + uint64_t t; + bool upd; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* Onset packet carries no capacity: feed back unknown. */ + if (!mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, MS)) { + printf("Onset did not update.\n"); + goto fail_ctx; + } + + if (fcap != 0) { + printf("Onset fed back cap: exp 0, got %u.\n", fcap); + goto fail_ctx; + } + + 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; + if (!mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t)) { + printf("Window did not close.\n"); + goto fail_ctx; + } + + if (fcap != 36) { + printf("Window min cap: exp 36, got %u.\n", fcap); + goto fail_ctx; + } + + /* The next window starts unknown; follow the adapted rx_tw. */ + upd = false; + for (i = 0; i < 128 && !upd; i++) { + t += CA_TW_INIT; + upd = mb_ecn_rcv(ctx, LEN, 8, 0, &ece, &fcap, t); + } + + if (!upd) { + printf("Second window did not close.\n"); + goto fail_ctx; + } + + if (fcap != 0) { + printf("Stale cap %u leaked into the next window.\n", fcap); + 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; +} + +/* Onset and gap restarts emit the triggering packet's cap, fresh. */ +static int test_mb_ecn_rcv_cap_onset_fresh(void) +{ + struct mb_ecn_ctx * ctx; + uint16_t ece; + uint8_t fcap; + uint64_t t; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + if (!mb_ecn_rcv(ctx, LEN, 4, 77, &ece, &fcap, MS)) { + printf("Onset did not update.\n"); + goto fail_ctx; + } + + if (fcap != 77) { + printf("Onset cap: exp 77, got %u.\n", fcap); + goto fail_ctx; + } + + 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; + if (!mb_ecn_rcv(ctx, LEN, 4, 90, &ece, &fcap, t)) { + printf("Gap restart did not update.\n"); + goto fail_ctx; + } + + if (fcap != 90) { + printf("Gap restart cap: exp 90, got %u.\n", fcap); + 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; +} + +/* Fed-back capacity derives the floor and slope: EWMA toward C/32. */ +static int test_mb_ecn_ece_cap_derives_rates(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t tgt; + uint64_t want; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* Code 120 = 2^30 B/s; target floor = 2^25 B/s. */ + tgt = cap_dec(120) >> CA_CAP_SHFT; + + mb_ecn_ece(ctx, 100, 120, MS); + + want = CA_RATE_MIN + ((tgt - CA_RATE_MIN) >> CA_CAP_SM_SHFT); + if (ctx->rate_min != want) { + printf("Floor: exp %" PRIu64 ", got %" PRIu64 ".\n", + want, ctx->rate_min); + goto fail_ctx; + } + + if (ctx->ai_rate != ctx->rate_min) { + printf("AI slope did not track the floor.\n"); + goto fail_ctx; + } + + mb_ecn_ece(ctx, 100, 120, 2 * MS); + + want += (tgt - want) >> CA_CAP_SM_SHFT; + if (ctx->rate_min != want) { + printf("Floor EWMA: exp %" PRIu64 ", got %" PRIu64 ".\n", + want, ctx->rate_min); + goto fail_ctx; + } + + if (ctx->n_cap != 2) { + printf("Capacity updates: exp 2, got %" PRIu64 ".\n", + ctx->n_cap); + 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; +} + +/* Feedback without a capacity leaves the derived rates untouched. */ +static int test_mb_ecn_ece_cap_zero_keeps_rates(void) +{ + struct mb_ecn_ctx * ctx; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + mb_ecn_ece(ctx, 100, 0, MS); + + if (ctx->rate_min != CA_RATE_MIN || ctx->ai_rate != CA_AI_RATE) { + printf("Unknown cap moved the derived rates.\n"); + goto fail_ctx; + } + + if (ctx->n_cap != 0) { + printf("Unknown cap counted as an update.\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; +} + +/* The derived floor clamps to [CA_RATE_MIN, CA_RMIN_MAX]. */ +static int test_mb_ecn_ece_cap_clamps(void) +{ + struct mb_ecn_ctx * ctx; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* A path slower than the default floor cannot lower it. */ + mb_ecn_ece(ctx, 100, 1, MS); + + if (ctx->rate_min != CA_RATE_MIN) { + printf("Slow path lowered the floor: %" PRIu64 ".\n", + ctx->rate_min); + goto fail_ctx; + } + + /* A absurdly fast path saturates at the ceiling. */ + for (i = 1; i <= 40; i++) + mb_ecn_ece(ctx, 100, 255, (1 + i) * MS); + + if (ctx->rate_min > CA_RMIN_MAX) { + printf("Floor above the ceiling: %" PRIu64 ".\n", + ctx->rate_min); + goto fail_ctx; + } + + if (ctx->rate_min < CA_RMIN_MAX - 4) { + printf("Floor did not reach the ceiling: %" PRIu64 ".\n", + ctx->rate_min); + 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; +} + +/* Stale capacity reverts the derived rates to the defaults. */ +static int test_mb_ecn_cap_ttl_reverts(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + mb_ecn_ece(ctx, 100, 120, MS); + + if (ctx->rate_min == CA_RATE_MIN) { + printf("Capacity did not derive a floor.\n"); + goto fail_ctx; + } + + /* Just inside the TTL: the derived rates hold. */ + mb_ecn_snd(ctx, LEN, MS + (ctx->ece_ttl << CA_CAP_TTL_SHFT), &ftag); + + if (ctx->rate_min == CA_RATE_MIN) { + printf("Derived rates reverted too early.\n"); + goto fail_ctx; + } + + /* Past the TTL: back to the defaults. */ + mb_ecn_snd(ctx, LEN, MS + (ctx->ece_ttl << CA_CAP_TTL_SHFT) + 1, + &ftag); + + if (ctx->rate_min != CA_RATE_MIN || ctx->ai_rate != CA_AI_RATE) { + printf("Stale capacity kept the derived rates.\n"); + goto fail_ctx; + } + + if (ctx->tx_cap != 0) { + printf("Stale capacity code not cleared.\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; +} + +/* The control law uses the per-ctx AI slope. */ +static int test_mb_ecn_ctrl_per_ctx_ai(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t want; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* 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; + want += want * (30 * MS) / CA_PROBE_TC; + + mb_ecn_snd(ctx, LEN, 30 * MS, &ftag); + + if (ctx->rate != want) { + printf("AI not per-ctx: exp %" PRIu64 ", got %" PRIu64 ".\n", + want, 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; +} + +/* The rate clamp honours the per-ctx derived floor. */ +static int test_mb_ecn_ctrl_per_ctx_floor(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate_min = (uint64_t) 1 << 20; + ctx->rate = ((uint64_t) 1 << 20) + 1000; + + mb_ecn_ece(ctx, CA_ECE_REF, 0, 0); + mb_ecn_snd(ctx, LEN, 30 * MS, &ftag); + + if (ctx->rate != ctx->rate_min) { + printf("Floor not per-ctx: %" 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; +} + +/* + * A paced flow slower than one packet per CA_DT_CAP must not decay: + * the gap credit law grants its true elapsed service, so the lead + * stays pinned at ~one packet instead of growing without bound. + */ +static int test_mb_ecn_snd_slow_rate_paced(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + uint64_t t = 0; + time_t wait; + size_t i; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* 8 KB/s, 1400 B packets: inter-send gap ~171 ms > CA_DT_CAP. */ + ctx->rate = 8192; + ctx->rate_min = 8192; + ctx->ai_rate = 0; + ctx->inv_rate = mb_ecn_rate_inv(8192); + ctx->tx_cav = true; + + for (i = 0; i < 50; i++) { + wait = mb_ecn_snd(ctx, 1400, t, &ftag); + t += wait > 0 ? (uint64_t) wait : 1; + } + + if (ctx->lead > 2 * 1400) { + printf("Pacer starves a slow flow: lead %" PRIu64 ".\n", + ctx->lead); + 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 long idle makes the aggregate source-limited, so the offered-load + * ceiling bounds the resume rate (hence the burst) well below the + * pre-idle rate. + */ +static int test_mb_ecn_snd_idle_burst_bound(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = (uint64_t) 1 << 20; + ctx->inv_rate = mb_ecn_rate_inv(ctx->rate); + ctx->tx_cav = true; + + mb_ecn_snd(ctx, 1400, 0, &ftag); /* warm-up: sets started */ + + /* 600 s idle. */ + mb_ecn_snd(ctx, 1400, 600 * BILLION, &ftag); + + if (ctx->backlogged) { + printf("long idle did not clear backlogged.\n"); + goto fail_ctx; + } + + if (ctx->rate >= ((uint64_t) 1 << 20)) { + printf("idle resume rate not ceiling-bounded: %" 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; +} + +/* A deep backlog at a low rate must not wrap the wait computation. */ +static int test_mb_ecn_snd_wait_no_overflow(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t want; + uint64_t ftag; + time_t wait; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->rate = 8192; + ctx->inv_rate = mb_ecn_rate_inv(8192); + + /* 128 flows x 1400 B of SFQ lead at the floor rate. */ + ftag = 128 * 1400; + + wait = mb_ecn_snd(ctx, 1400, 0, &ftag); + want = (uint64_t) 128 * 1400 * BILLION / 8192; + + if ((uint64_t) wait < want - want / 100 || + (uint64_t) wait > want + want / 100) { + printf("Wait wrapped: exp ~%" PRIu64 ", got %" PRIu64 ".\n", + want, (uint64_t) wait); + 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 fully paced-backlogged flow reads backlogged after a window. */ +static int test_mb_ecn_backlogged_paced(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->backlogged = false; /* prove a window close re-earns it */ + + drive_backlogged(ctx, &ftag, MS, 4 * CA_SND_WIN, LEN); + + if (!ctx->backlogged) { + printf("paced-backlogged flow read source-limited.\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 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. + */ +static int test_mb_ecn_source_limited_ceiling(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + uint64_t t = 10 * MS; + + TEST_START(); + + 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) 1 << 20; + 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); + + if (ctx->rate != ((uint64_t) 2 << 20)) { + printf("ceiling: exp %" PRIu64 ", got %" PRIu64 ".\n", + (uint64_t) 2 << 20, ctx->rate); + goto fail_ctx; + } + + if (!ctx->src_limited) { + printf("ceiling bound but src_limited not set.\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; +} + +/* One quiet window must not collapse the max-filter; it decays ~1/16. */ +static int test_mb_ecn_max_filter(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + uint64_t hi = (uint64_t) 10 << 20; + uint64_t t = 10 * MS; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->started = true; + ctx->snd_rate = hi; + ctx->snd_r0 = hi; + ctx->snd_byt = 0; + ctx->snd_win = t; + ctx->last_ts = t; + ctx->last_ctrl = t; + + /* Close one window with almost no bytes offered. */ + mb_ecn_snd(ctx, LEN, t + CA_SND_WIN + 1, &ftag); + + if (ctx->snd_rate >= hi || ctx->snd_rate < hi - hi / 8) { + printf("max-filter: exp ~15/16 of %" PRIu64 ", got %" + PRIu64 " after one quiet window.\n", + hi, ctx->snd_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; +} + +/* A >CA_DT_CAP gap clears backlogged without touching the estimate. */ +static int test_mb_ecn_idle_clears_backlogged(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + uint64_t snd_rate = (uint64_t) 5 << 20; + uint64_t t = 10 * MS; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + ctx->started = true; + ctx->backlogged = true; + ctx->snd_rate = snd_rate; + ctx->snd_win = t; + ctx->last_ts = t; + ctx->last_ctrl = t; + + /* 55 ms gap: past CA_DT_CAP, under CA_SND_WIN (no window close). */ + mb_ecn_snd(ctx, LEN, t + 55 * MS, &ftag); + + if (ctx->backlogged) { + printf("idle gap did not clear backlogged.\n"); + goto fail_ctx; + } + + if (ctx->snd_rate != snd_rate) { + printf("idle step altered snd_rate %" PRIu64 ".\n", + ctx->snd_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; +} + +/* The first send is never misread as idle, whatever the wall clock. */ +static int test_mb_ecn_first_send_warmup(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t ftag = 0; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + /* started == false; a large first timestamp must not look idle. */ + mb_ecn_snd(ctx, LEN, 500 * MS, &ftag); + + if (!ctx->started) { + printf("first send did not set the warm-up sentinel.\n"); + goto fail_ctx; + } + + if (!ctx->backlogged) { + printf("first send misclassified as idle.\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; +} + +/* + * Couple one or two backlogged flows through a shared bottleneck of + * capacity cap: each step marks the shared queue with the real + * calc_ecn, feeds each flow that mark delayed by its own lag (in + * steps), drives it backlogged for one step, then drains the queue. + * A faithful discrete run of the fluid model on the real control law, + * with the forwarder abstracted to a single shared price. b may be + * NULL for a single-flow run. + */ +#define TF_STEP (5 * MS) /* control step (ns) */ +#define TF_HIST 64 /* mark ring, bounds max lag */ +#define TF_QMAX (8192 * LEN) /* bottleneck buffer (bytes) */ +#define TF_MAXN 32 /* flows per shared-link run */ + +static void shared_link_run(struct mb_ecn_ctx * a, + struct mb_ecn_ctx * b, + uint64_t cap, + size_t lag_a, + size_t lag_b, + size_t steps) +{ + uint16_t hist[TF_HIST]; + uint64_t ta = 0; + uint64_t tb = 0; + uint64_t fta = 0; + uint64_t ftb = 0; + uint64_t q = 0; + uint64_t drain = cap * TF_STEP / BILLION; + uint64_t tgt; + uint64_t arr; + uint8_t cc = cap_enc(cap); + uint8_t ecn; + uint16_t ea; + uint16_t eb; + size_t k; + + memset(hist, 0, sizeof(hist)); + + for (k = 0; k < steps; k++) { + tgt = (k + 1) * TF_STEP; + ecn = 0; + + mb_ecn_calc_ecn(q / LEN, &ecn, QOS_CUBE_BE, LEN); + hist[k % TF_HIST] = (uint16_t) (ecn << CA_SHFT); + + ea = k < lag_a ? 0 : hist[(k - lag_a) % TF_HIST]; + if (ea > 0) /* no feedback until congestion */ + mb_ecn_ece(a, ea, cc, ta); + + /* Active flow: heartbeat/window liveness stays fresh. */ + a->last_sig = ta; + if (tgt > ta) + ta = drive_backlogged(a, &fta, ta, tgt - ta, LEN); + + arr = a->rate * TF_STEP / BILLION; + + if (b != NULL) { + eb = k < lag_b ? 0 : hist[(k - lag_b) % TF_HIST]; + if (eb > 0) + mb_ecn_ece(b, eb, cc, tb); + + b->last_sig = tb; + if (tgt > tb) + tb = drive_backlogged(b, &ftb, tb, + tgt - tb, LEN); + + arr += b->rate * TF_STEP / BILLION; + } + + q += arr; + q = q > drain ? q - drain : 0; + if (q > (uint64_t) TF_QMAX) + q = TF_QMAX; + } +} + +/* + * Two flows sharing one bottleneck, both fed every queue mark, must + * converge from a lopsided start toward an equal split (the fluid + * model's drho/dt -> 0) rather than latch winner-take-all. Isolates + * the rate law from the forwarder: neither flow is starved of marks. + */ +static int test_mb_ecn_two_flow_converge(void) +{ + struct mb_ecn_ctx * a; + struct mb_ecn_ctx * b; + uint64_t cap = 1ULL << 20; + uint64_t fair = (1ULL << 20) / 2; + uint64_t lo; + uint64_t hi; + + TEST_START(); + + a = mk_ctx(); + b = mk_ctx(); + if (a == NULL || b == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + a->rate = cap; /* a hogs, b starts small */ + b->rate = CA_RATE_INIT; + + shared_link_run(a, b, cap, 0, 0, 6000); + + lo = a->rate < b->rate ? a->rate : b->rate; + hi = a->rate > b->rate ? a->rate : b->rate; + + if (lo < fair / 4) { + printf("flow starved: %" PRIu64 " / %" PRIu64 ".\n", + a->rate, b->rate); + goto fail_ctx; + } + + if (hi > 3 * lo) { + printf("did not converge: %" PRIu64 " / %" PRIu64 ".\n", + a->rate, b->rate); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(a); + mb_ecn_ctx_destroy(b); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * A remote bottleneck exits slow start only on fed-back congestion, + * so a longer feedback delay lets the exponential ramp overshoot + * further: the slow-start peak grows with RTT. Nothing caps the ramp + * at the path capacity, so the sender latches near the first-hop rate. + */ +static int test_mb_ecn_ramp_overshoot_grows_with_rtt(void) +{ + struct mb_ecn_ctx * lo_rtt; + struct mb_ecn_ctx * hi_rtt; + uint64_t cap = 1ULL << 20; + + TEST_START(); + + lo_rtt = mk_ctx(); + hi_rtt = mk_ctx(); + if (lo_rtt == NULL || hi_rtt == NULL) { + printf("Failed to create contexts.\n"); + goto fail_ctx; + } + + /* Same bottleneck; feedback lags 1 vs 8 steps (~5 vs 40 ms). */ + shared_link_run(lo_rtt, NULL, cap, 1, 0, 2000); + shared_link_run(hi_rtt, NULL, cap, 8, 0, 2000); + + if (hi_rtt->ss_peak <= lo_rtt->ss_peak) { + printf("overshoot did not grow with RTT: %" PRIu64 + " vs %" PRIu64 ".\n", + hi_rtt->ss_peak, lo_rtt->ss_peak); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(lo_rtt); + mb_ecn_ctx_destroy(hi_rtt); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(lo_rtt); + mb_ecn_ctx_destroy(hi_rtt); + TEST_FAIL(); + return TEST_RC_FAIL; +} + +/* + * Washout damps a fixed 1/4 of the rate change once per wall-clock + * bucket: a sub-bucket step banks time only, a full bucket removes a + * quarter of the gap either way without crossing the snapshot, and a + * sparse step (> CA_DT_CAP) resets it so a starved sender keeps its cut. + */ +static int test_mb_ecn_washout_bucket(void) +{ + struct mb_ecn_ctx * ctx; + uint64_t r0 = (uint64_t) 100 << 20; + uint64_t gap; + + TEST_START(); + + ctx = mk_ctx(); + if (ctx == NULL) { + printf("Failed to create context.\n"); + goto fail; + } + + gap = r0 >> 4; + + /* Sub-bucket: time banks, the rate does not move. */ + ctx->rate = r0; + ctx->r_bkt = r0 - gap; + ctx->wash_acc = 0; + mb_ecn_washout(ctx, MS, MS); + if (ctx->rate != r0 || ctx->wash_acc != MS) { + printf("sub-bucket washout moved the rate: %" PRIu64 ".\n", + ctx->rate); + goto fail_ctx; + } + + /* Full bucket, rate leads: cut a quarter of the gap, no crossing. */ + ctx->rate = r0; + ctx->r_bkt = r0 - gap; + ctx->wash_acc = 0; + mb_ecn_washout(ctx, MS, CA_WASH_BKT); + if (ctx->rate != r0 - (gap >> CA_WASH_SHFT)) { + printf("bucket down: got %" PRIu64 ".\n", ctx->rate); + goto fail_ctx; + } + + if (ctx->rate <= r0 - gap) { + printf("washout reversed a ramp: %" PRIu64 ".\n", ctx->rate); + goto fail_ctx; + } + + if (ctx->r_bkt != ctx->rate || ctx->wash_acc != 0) { + printf("washout did not reset the bucket.\n"); + goto fail_ctx; + } + + /* Full bucket, rate trails: add a quarter of the gap (symmetric). */ + ctx->rate = r0; + ctx->r_bkt = r0 + gap; + ctx->wash_acc = 0; + mb_ecn_washout(ctx, MS, CA_WASH_BKT); + if (ctx->rate != r0 + (gap >> CA_WASH_SHFT)) { + printf("bucket up: got %" PRIu64 ".\n", ctx->rate); + goto fail_ctx; + } + + /* Sparse step: reset to the current rate, keep the cut. */ + ctx->rate = r0; + ctx->r_bkt = r0 - gap; + ctx->wash_acc = CA_WASH_BKT / 2; + mb_ecn_washout(ctx, CA_DT_CAP + 1, MS); + if (ctx->rate != r0 || ctx->r_bkt != r0 || ctx->wash_acc != 0) { + printf("sparse step did not reset the bucket.\n"); + goto fail_ctx; + } + + mb_ecn_ctx_destroy(ctx); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_ctx: + mb_ecn_ctx_destroy(ctx); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +int mb_ecn_test(int argc, + char ** argv) +{ + int ret = 0; + + (void) argc; + (void) argv; + + ret |= test_mb_ecn_ctx_create_destroy(); + 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_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(); + ret |= test_mb_ecn_ece_ttl_covers_cadence(); + ret |= test_mb_ecn_slow_start(); + ret |= test_mb_ecn_dt_scaling_invariant(); + ret |= test_mb_ecn_probe_scale_invariant(); + ret |= test_mb_ecn_multiplicative_decrease(); + ret |= test_mb_ecn_fixed_point(); + ret |= test_mb_ecn_lead_cut(); + ret |= test_mb_ecn_slow_start_local_brake(); + ret |= test_mb_ecn_slow_start_clean_ramp(); + ret |= test_mb_ecn_starved_decrease_escape(); + ret |= test_mb_ecn_starved_local_fallback(); + ret |= test_mb_ecn_decrease_dt_invariant(); + ret |= test_mb_ecn_decrease_subms_carry(); + ret |= test_mb_ecn_idle_resume_bounded(); + ret |= test_mb_ecn_rate_floor(); + ret |= test_mb_ecn_ece_staleness(); + ret |= test_mb_ecn_ece_ttl_tracks_rate(); + ret |= test_mb_ecn_sfq_pace(); + ret |= test_mb_ecn_probe_time_constant(); + ret |= test_mb_ecn_rcv_cap_window_min(); + ret |= test_mb_ecn_rcv_cap_onset_fresh(); + ret |= test_mb_ecn_ece_cap_derives_rates(); + ret |= test_mb_ecn_ece_cap_zero_keeps_rates(); + ret |= test_mb_ecn_ece_cap_clamps(); + ret |= test_mb_ecn_cap_ttl_reverts(); + ret |= test_mb_ecn_ctrl_per_ctx_ai(); + ret |= test_mb_ecn_ctrl_per_ctx_floor(); + ret |= test_mb_ecn_snd_slow_rate_paced(); + ret |= test_mb_ecn_snd_idle_burst_bound(); + ret |= test_mb_ecn_snd_wait_no_overflow(); + ret |= test_mb_ecn_backlogged_paced(); + ret |= test_mb_ecn_source_limited_ceiling(); + ret |= test_mb_ecn_max_filter(); + ret |= test_mb_ecn_idle_clears_backlogged(); + ret |= test_mb_ecn_first_send_warmup(); + ret |= test_mb_ecn_two_flow_converge(); + ret |= test_mb_ecn_ramp_overshoot_grows_with_rtt(); + ret |= test_mb_ecn_washout_bucket(); + + return ret; +} diff --git a/src/ipcpd/unicast/cap.c b/src/ipcpd/unicast/cap.c new file mode 100644 index 00000000..0d823dc6 --- /dev/null +++ b/src/ipcpd/unicast/cap.c @@ -0,0 +1,291 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Link capacity estimation + * + * 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/. + */ + +#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. + */ + +#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])); +} + +/* + * 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}; + uint64_t r = rate; /* copy halved to find band */ + unsigned e = 0; /* band: floor log2 rate */ + unsigned k = 0; /* quarter within band 0..3 */ + unsigned c; /* code = 4 * band + quarter */ + uint16_t top; /* rate scaled to [256, 512) */ + + if (rate == 0) + return 0; + + while (r > 1) { + r >>= 1; + e++; + } + + /* Top 9 bits: rate normalized to [256, 512). */ + top = e >= 8 ? (uint16_t) (rate >> (e - 8)) + : (uint16_t) (rate << (8 - e)); + + while (k < 3 && top >= thr[k]) + k++; + + c = 4 * e + k; + if (c == 0) + c = 1; /* 0 means unknown */ + + return (uint8_t) c; +} + +uint64_t cap_dec(uint8_t c) +{ + static const uint16_t m[4] = {256, 304, 362, 431}; + unsigned e = c >> 2; /* band = c >> 2 */ + unsigned k = c & 3; /* quarter = c & 3 */ + + if (c == 0) + return 0; + + if (e >= 8) + return (uint64_t) m[k] << (e - 8); + + return ((uint64_t) m[k] << e) >> 8; +} + +uint8_t cap_min(uint8_t a, + uint8_t b) +{ + if (a == 0) + return b; + + if (b == 0) + return a; + + return a < b ? a : b; +} + +void cap_stamp(uint8_t * pci, + uint8_t own) +{ + if (own == 0) + return; + + 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 new file mode 100644 index 00000000..df8c2ec1 --- /dev/null +++ b/src/ipcpd/unicast/cap.h @@ -0,0 +1,54 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Link capacity estimation + * + * 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/. + */ + +#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); + +uint64_t cap_dec(uint8_t c); + +uint8_t cap_min(uint8_t a, + uint8_t b); + +/* MIN-combine the own link code into the PCI byte. */ +void cap_stamp(uint8_t * pci, + uint8_t own); + +#endif /* OUROBOROS_IPCPD_UNICAST_CAP_H */ diff --git a/src/ipcpd/unicast/dir/dht.c b/src/ipcpd/unicast/dir/dht.c index d1d68e49..c2cd33aa 100644 --- a/src/ipcpd/unicast/dir/dht.c +++ b/src/ipcpd/unicast/dir/dht.c @@ -2238,7 +2238,7 @@ static int dht_send_msg(dht_msg_t * msg, dht_msg__pack(msg, ssm_pk_buff_head(spb)); - if (dt_write_packet(addr, QOS_CUBE_BE, dht.eid, spb) < 0) { + if (dt_write_packet(addr, QOS_CUBE_BE, dht.eid, spb, NULL) < 0) { log_warn("%s write failed", DHT_CODE(msg)); goto fail_send; } @@ -2954,7 +2954,7 @@ static dht_msg_t * do_dht_kv_find_value_req(const dht_find_req_msg_t * req) fail_msg: freebufs(vals, n_vals); fail_vals: - while (n_contacts-- > 0) + while (contacts != NULL && n_contacts-- > 0) dht_contact_msg__free_unpacked(contacts[n_contacts], NULL); free(contacts); @@ -3471,6 +3471,8 @@ static void value_list_destroy(struct list_head * vl) #define MUST_REPLICATE(v, now) ((now)->tv_sec > (v)->t_repl + dht.t_repl) #define MUST_REPUBLISH(v, now) /* Close to expiry deadline */ \ (((v)->t_exp - (now)->tv_sec) < (DHT_N_REPUB * dht.t_repl)) +/* A local value must be (re)stored if near expiry or never stored. */ +#define MUST_STORE_LVAL(v, now) (MUST_REPUBLISH(v, now) || (v)->t_repl == 0) static void dht_entry_get_repl_lists(const struct dht_entry * e, struct list_head * repl, struct list_head * rebl, @@ -3492,7 +3494,7 @@ static void dht_entry_get_repl_lists(const struct dht_entry * e, llist_for_each(p, &e->lvals) { struct val_entry * v = list_entry(p, struct val_entry, next); - if (MUST_REPLICATE(v, now) && MUST_REPUBLISH(v, now)) { + if (MUST_REPLICATE(v, now) && MUST_STORE_LVAL(v, now)) { /* Add expire time here, to allow creating val_entry */ n = val_entry_create(v->val, now->tv_sec + dht.t_exp); if (n == NULL) @@ -3782,6 +3784,9 @@ static void * work(void * o) log_dbg("DHT worker starting %ld seconds interval.", intv * n); + /* Flush names registered before we had peers to store them. */ + dht_kv_replicate(); + while (true) { int i = 0; while (tasks[i] != NULL) { diff --git a/src/ipcpd/unicast/dt.c b/src/ipcpd/unicast/dt.c index e89cb17e..bfc2ece7 100644 --- a/src/ipcpd/unicast/dt.c +++ b/src/ipcpd/unicast/dt.c @@ -36,6 +36,7 @@ #include <ouroboros/errno.h> #include <ouroboros/logs.h> #include <ouroboros/dev.h> +#include <ouroboros/ipcp-dev.h> #include <ouroboros/notifier.h> #include <ouroboros/rib.h> #ifdef IPCP_FLOW_STATS @@ -46,6 +47,7 @@ #include "common/comp.h" #include "common/connmgr.h" #include "ca.h" +#include "cap.h" #include "ipcp.h" #include "dt.h" #include "pff.h" @@ -78,12 +80,14 @@ struct comp_info { #define TTL_LEN 1 #define QOS_LEN 1 #define ECN_LEN 1 +#define CAP_LEN 1 struct dt_pci { uint64_t dst_addr; qoscube_t qc; uint8_t ttl; uint8_t ecn; + uint8_t cap; uint64_t eid; }; @@ -96,6 +100,7 @@ struct { size_t qc_o; size_t ttl_o; size_t ecn_o; + size_t cap_o; size_t eid_o; /* Initial TTL value */ @@ -115,6 +120,7 @@ static void dt_pci_ser(uint8_t * head, memcpy(head + dt_pci_info.qc_o, &dt_pci->qc, QOS_LEN); memcpy(head + dt_pci_info.ttl_o, &ttl, TTL_LEN); memcpy(head + dt_pci_info.ecn_o, &dt_pci->ecn, ECN_LEN); + memcpy(head + dt_pci_info.cap_o, &dt_pci->cap, CAP_LEN); memcpy(head + dt_pci_info.eid_o, &dt_pci->eid, dt_pci_info.eid_size); } @@ -133,6 +139,7 @@ static void dt_pci_des(uint8_t * head, memcpy(&dt_pci->qc, head + dt_pci_info.qc_o, QOS_LEN); memcpy(&dt_pci->ttl, head + dt_pci_info.ttl_o, TTL_LEN); memcpy(&dt_pci->ecn, head + dt_pci_info.ecn_o, ECN_LEN); + memcpy(&dt_pci->cap, head + dt_pci_info.cap_o, CAP_LEN); memcpy(&dt_pci->eid, head + dt_pci_info.eid_o, dt_pci_info.eid_size); } @@ -151,6 +158,7 @@ struct { struct pff * pff[QOS_CUBE_MAX]; struct routing_i * routing[QOS_CUBE_MAX]; #ifdef IPCP_FLOW_STATS + /* Flow stats use lock-free atomics; stamp is the validity flag. */ struct { time_t stamp; uint64_t addr; @@ -168,7 +176,6 @@ struct { size_t w_drp_bytes[QOS_CUBE_MAX]; size_t f_nhp_pkt[QOS_CUBE_MAX]; size_t f_nhp_bytes[QOS_CUBE_MAX]; - pthread_mutex_t lock; } stat[PROC_MAX_FLOWS]; size_t n_flows; @@ -180,10 +187,6 @@ struct { pthread_t listener; } dt; -/* - * Flow stats are lock-free relaxed atomics on the data path; the per-flow - * lock still guards the stamp/addr/n_flows lifecycle (see stat_used). - */ #ifdef IPCP_FLOW_STATS #define dt_stat_inc(idx, name, qc, len) \ do { \ @@ -204,6 +207,8 @@ static int dt_rib_read(const char * path, char tmstr[RIB_TM_STRLEN]; size_t rxqlen = 0; size_t txqlen = 0; + time_t stamp; + uint64_t addr; struct tm * tm; /* NOTE: we may need stronger checks. */ @@ -217,19 +222,18 @@ static int dt_rib_read(const char * path, buf[0] = '\0'; - pthread_mutex_lock(&dt.stat[fd].lock); - - if (dt.stat[fd].stamp == 0) { - pthread_mutex_unlock(&dt.stat[fd].lock); + stamp = LOAD_ACQUIRE(&dt.stat[fd].stamp); + if (stamp == 0) return 0; - } - if (dt.stat[fd].addr == dt.addr) + addr = LOAD_RELAXED(&dt.stat[fd].addr); + + if (addr == dt.addr) sprintf(addrstr, "%s", dt.comps[fd].name); else - sprintf(addrstr, ADDR_FMT32, ADDR_VAL32(&dt.stat[fd].addr)); + sprintf(addrstr, ADDR_FMT32, ADDR_VAL32(&addr)); - tm = gmtime(&dt.stat[fd].stamp); + tm = gmtime(&stamp); strftime(tmstr, sizeof(tmstr), RIB_TM_FORMAT, tm); if (fd >= PROC_RES_FDS) { @@ -279,8 +283,6 @@ static int dt_rib_read(const char * path, strcat(buf, str); } - pthread_mutex_unlock(&dt.stat[fd].lock); - return RIB_FILE_STRLEN; } @@ -301,15 +303,9 @@ static int dt_rib_readdir(char *** buf) if (*buf == NULL) goto fail_entries; - for (i = 0; i < PROC_MAX_FLOWS; ++i) { - pthread_mutex_lock(&dt.stat[i].lock); - - if (dt.stat[i].stamp == 0) { - pthread_mutex_unlock(&dt.stat[i].lock); - break; - } - - pthread_mutex_unlock(&dt.stat[i].lock); + for (i = 0; i < PROC_MAX_FLOWS && idx < (int) dt.n_flows; ++i) { + if (LOAD_RELAXED(&dt.stat[i].stamp) == 0) + continue; /* n-1 flows start at PROC_RES_FDS */ sprintf(entry, "%zu", i); @@ -328,6 +324,7 @@ static int dt_rib_readdir(char *** buf) fail_entry: while (idx-- > 0) free((*buf)[idx]); + free(*buf); fail_entries: pthread_rwlock_unlock(&dt.lock); @@ -339,24 +336,23 @@ static int dt_rib_getattr(const char * path, { int fd; char * entry; + time_t stamp; entry = strstr(path, RIB_SEPARATOR) + 1; assert(entry); fd = atoi(entry); - pthread_mutex_lock(&dt.stat[fd].lock); + stamp = LOAD_ACQUIRE(&dt.stat[fd].stamp); - if (dt.stat[fd].stamp != -1) { + if (stamp != -1) { attr->size = RIB_FILE_STRLEN; - attr->mtime = dt.stat[fd].stamp; + attr->mtime = stamp; } else { attr->size = 0; attr->mtime = 0; } - pthread_mutex_unlock(&dt.stat[fd].lock); - return 0; } @@ -366,30 +362,45 @@ static struct rib_ops r_ops = { .getattr = dt_rib_getattr }; -/* - * Hold dt.lock + per-stat together: dt_rib_readdir samples n_flows - * under rdlock and walks stamps under per-stat; updates must be - * atomic w.r.t. that snapshot or the malloc(n_flows) buffer can - * overflow. - */ static void stat_used(int fd, uint64_t addr) { struct timespec now; + int i; clock_gettime(CLOCK_REALTIME_COARSE, &now); pthread_rwlock_wrlock(&dt.lock); - pthread_mutex_lock(&dt.stat[fd].lock); - memset(&dt.stat[fd], 0, sizeof(dt.stat[fd])); + STORE_RELEASE(&dt.stat[fd].stamp, 0); - dt.stat[fd].stamp = (addr != INVALID_ADDR) ? now.tv_sec : 0; - dt.stat[fd].addr = addr; + /* Don't memset: incremented without locks in fast path. */ + for (i = 0; i < QOS_CUBE_MAX; ++i) { + STORE_RELAXED(&dt.stat[fd].snd_pkt[i], 0); + STORE_RELAXED(&dt.stat[fd].snd_bytes[i], 0); + STORE_RELAXED(&dt.stat[fd].rcv_pkt[i], 0); + STORE_RELAXED(&dt.stat[fd].rcv_bytes[i], 0); + STORE_RELAXED(&dt.stat[fd].lcl_r_pkt[i], 0); + STORE_RELAXED(&dt.stat[fd].lcl_r_bytes[i], 0); + STORE_RELAXED(&dt.stat[fd].lcl_w_pkt[i], 0); + STORE_RELAXED(&dt.stat[fd].lcl_w_bytes[i], 0); + STORE_RELAXED(&dt.stat[fd].r_drp_pkt[i], 0); + STORE_RELAXED(&dt.stat[fd].r_drp_bytes[i], 0); + STORE_RELAXED(&dt.stat[fd].w_drp_pkt[i], 0); + STORE_RELAXED(&dt.stat[fd].w_drp_bytes[i], 0); + STORE_RELAXED(&dt.stat[fd].f_nhp_pkt[i], 0); + STORE_RELAXED(&dt.stat[fd].f_nhp_bytes[i], 0); + } - (addr != INVALID_ADDR) ? ++dt.n_flows : --dt.n_flows; + STORE_RELAXED(&dt.stat[fd].addr, addr); + + if (addr != INVALID_ADDR) { + STORE_RELEASE(&dt.stat[fd].stamp, now.tv_sec); + ++dt.n_flows; + } else { + --dt.n_flows; + } - pthread_mutex_unlock(&dt.stat[fd].lock); pthread_rwlock_unlock(&dt.lock); } #else @@ -413,6 +424,7 @@ static void handle_event(void * self, #ifdef IPCP_FLOW_STATS stat_used(fd, c->conn_info.addr); #endif + cap_reset(fd); psched_add(dt.psched, fd); log_dbg("Added fd %d to packet scheduler.", fd); break; @@ -429,15 +441,17 @@ static void handle_event(void * self, } } -static void packet_handler(int fd, - qoscube_t qc, - struct ssm_pk_buff * spb) +static time_t packet_handler(int fd, + qoscube_t qc, + struct ssm_pk_buff * spb) { struct dt_pci dt_pci; int ret; int ofd; uint8_t * head; size_t len; + size_t qlen; + bool marks; len = ssm_pk_buff_len(spb); @@ -456,7 +470,7 @@ static void packet_handler(int fd, log_dbg("TTL was zero."); ipcp_spb_release(spb); dt_stat_inc(fd, r_drp, qc, len); - return; + return 0; } /* FIXME: Use qoscube from PCI instead of incoming flow. */ @@ -466,10 +480,16 @@ static void packet_handler(int fd, dt_pci.dst_addr); ipcp_spb_release(spb); dt_stat_inc(fd, f_nhp, qc, len); - return; + return 0; } - (void) ca_calc_ecn(ofd, head + dt_pci_info.ecn_o, qc, len); + marks = ca_marks_ecn(); + qlen = marks ? ipcp_flow_queued(ofd) : 0; + + (void) ca_calc_ecn(qlen, head + dt_pci_info.ecn_o, qc, len); + + if (marks) + cap_stamp(head + dt_pci_info.cap_o, cap_get(ofd)); ret = ipcp_flow_write(ofd, spb); if (ret < 0) { @@ -478,23 +498,27 @@ static void packet_handler(int fd, notifier_event(NOTIFY_DT_FLOW_DOWN, &ofd); ipcp_spb_release(spb); dt_stat_inc(ofd, w_drp, qc, len); - return; + return 0; } dt_stat_inc(ofd, snd, qc, len); + + if (marks) + cap_update(ofd, qlen, len); } else { dt_pci_shrink(spb); if (dt_pci.eid >= PROC_RES_FDS) { uint8_t ecn = *(head + dt_pci_info.ecn_o); - fa_np1_rcv(dt_pci.eid, ecn, spb); - return; + uint8_t cap = *(head + dt_pci_info.cap_o); + fa_np1_rcv(dt_pci.eid, ecn, cap, spb); + return 0; } if (dt.comps[dt_pci.eid].post_packet == NULL) { log_err("No registered component on eid %" PRIu64 ".", dt_pci.eid); ipcp_spb_release(spb); - return; + return 0; } dt_stat_inc(fd, lcl_r, qc, len); dt_stat_inc(dt_pci.eid, snd, qc, len); @@ -502,6 +526,8 @@ static void packet_handler(int fd, dt.comps[dt_pci.eid].post_packet(dt.comps[dt_pci.eid].comp, spb); } + + return 0; } static void * dt_conn_handle(void * o) @@ -560,10 +586,16 @@ int dt_init(struct dt_config cfg) dt_pci_info.qc_o = dt_pci_info.addr_size; dt_pci_info.ttl_o = dt_pci_info.qc_o + QOS_LEN; dt_pci_info.ecn_o = dt_pci_info.ttl_o + TTL_LEN; - dt_pci_info.eid_o = dt_pci_info.ecn_o + ECN_LEN; + dt_pci_info.cap_o = dt_pci_info.ecn_o + ECN_LEN; + 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 (connmgr_comp_init(COMPID_DT, &info)) { + 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; } @@ -603,14 +635,6 @@ int dt_init(struct dt_config cfg) #ifdef IPCP_FLOW_STATS memset(dt.stat, 0, sizeof(dt.stat)); - for (i = 0; i < PROC_MAX_FLOWS; ++i) - if (pthread_mutex_init(&dt.stat[i].lock, NULL)) { - log_err("Failed to init mutex for flow %d.", i); - for (j = 0; j < i; ++j) - pthread_mutex_destroy(&dt.stat[j].lock); - goto fail_stat_lock; - } - dt.n_flows = 0; sprintf(dtstr, "%s." ADDR_FMT32, DT, ADDR_VAL32(&dt.addr)); @@ -624,11 +648,8 @@ int dt_init(struct dt_config cfg) #ifdef IPCP_FLOW_STATS fail_rib_reg: - for (i = 0; i < PROC_MAX_FLOWS; ++i) - pthread_mutex_destroy(&dt.stat[i].lock); - fail_stat_lock: -#endif bmp_destroy(dt.res_fds); +#endif fail_res_fds: pthread_rwlock_destroy(&dt.lock); fail_rwlock_init: @@ -642,6 +663,8 @@ int dt_init(struct dt_config cfg) fail_routing: connmgr_comp_fini(COMPID_DT); fail_connmgr_comp_init: + cap_fini(); + fail_cap: return -1; } @@ -655,8 +678,6 @@ void dt_fini(void) #ifdef IPCP_FLOW_STATS sprintf(dtstr, "%s.%" PRIu64, DT, dt.addr); rib_unreg(dtstr); - for (i = 0; i < PROC_MAX_FLOWS; ++i) - pthread_mutex_destroy(&dt.stat[i].lock); #endif bmp_destroy(dt.res_fds); @@ -671,6 +692,8 @@ void dt_fini(void) routing_fini(); connmgr_comp_fini(COMPID_DT); + + cap_fini(); } int dt_start(void) @@ -773,13 +796,16 @@ void dt_unreg_comp(int eid) int dt_write_packet(uint64_t dst_addr, qoscube_t qc, uint64_t eid, - struct ssm_pk_buff * spb) + struct ssm_pk_buff * spb, + uint8_t * ecn) { struct dt_pci dt_pci; int fd; int ret; uint8_t * head; size_t len; + size_t qlen; + bool marks; assert(spb); assert(dst_addr != dt.addr); @@ -813,8 +839,18 @@ int dt_write_packet(uint64_t dst_addr, dt_pci.qc = qc; dt_pci.eid = eid; dt_pci.ecn = 0; + dt_pci.cap = 0; + + marks = ca_marks_ecn(); + qlen = marks ? ipcp_flow_queued(fd) : 0; + + (void) ca_calc_ecn(qlen, &dt_pci.ecn, qc, len); - (void) ca_calc_ecn(fd, &dt_pci.ecn, qc, len); + if (marks) + dt_pci.cap = cap_get(fd); + + if (ecn != NULL) + *ecn = dt_pci.ecn; dt_pci_ser(head, &dt_pci); @@ -828,14 +864,19 @@ int dt_write_packet(uint64_t dst_addr, #ifdef IPCP_FLOW_STATS if (dt_pci.eid < PROC_RES_FDS) dt_stat_inc(fd, lcl_w, qc, len); + dt_stat_inc(fd, snd, qc, len); #endif + if (marks) + cap_update(fd, qlen, len); + return 0; fail_write: #ifdef IPCP_FLOW_STATS if (eid < PROC_RES_FDS) dt_stat_inc(fd, lcl_w, qc, len); + dt_stat_inc(fd, w_drp, qc, len); #endif return -1; diff --git a/src/ipcpd/unicast/dt.h b/src/ipcpd/unicast/dt.h index a484377d..a055efea 100644 --- a/src/ipcpd/unicast/dt.h +++ b/src/ipcpd/unicast/dt.h @@ -48,6 +48,7 @@ void dt_unreg_comp(int eid); int dt_write_packet(uint64_t dst_addr, qoscube_t qc, uint64_t eid, - struct ssm_pk_buff * spb); + struct ssm_pk_buff * spb, + uint8_t * ecn); #endif /* OUROBOROS_IPCPD_UNICAST_DT_H */ diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c index c6eca175..596b101b 100644 --- a/src/ipcpd/unicast/fa.c +++ b/src/ipcpd/unicast/fa.c @@ -31,16 +31,19 @@ #define FA "flow-allocator" #define OUROBOROS_PREFIX FA +#include <ouroboros/atomics.h> +#include <ouroboros/dev.h> #include <ouroboros/endian.h> -#include <ouroboros/logs.h> -#include <ouroboros/fqueue.h> #include <ouroboros/errno.h> -#include <ouroboros/dev.h> +#include <ouroboros/fqueue.h> #include <ouroboros/ipcp-dev.h> +#include <ouroboros/logs.h> #include <ouroboros/np1_flow.h> -#include <ouroboros/rib.h> -#include <ouroboros/random.h> #include <ouroboros/pthread.h> +#include <ouroboros/qoscube.h> +#include <ouroboros/random.h> +#include <ouroboros/rib.h> +#include <ouroboros/time.h> #include "addr-auth.h" #include "dir.h" @@ -66,6 +69,10 @@ #define FLOW_REPLY 1 #define FLOW_UPDATE 2 #define FLOW_IRM_UPDATE 3 +#define FLOW_HB 4 +#define FLOW_ACK 5 + +#define HB_ID_LEN 16 /* 128-bit unguessable heartbeat nonce */ #define STAT_FILE_LEN 0 @@ -81,9 +88,11 @@ struct fa_msg { uint32_t max_gap; uint32_t timeout; uint16_t ece; + uint8_t cap; uint8_t code; uint8_t availability; uint8_t service; + uint8_t hb_id[HB_ID_LEN]; /* heartbeat / ack nonce */ } __attribute__((packed)); struct cmd { @@ -91,6 +100,22 @@ struct cmd { struct ssm_pk_buff * spb; }; +#define HB_TBL_TTL (4ULL * BILLION) /* drop unanswered heartbeats */ +#define HB_TBL_MAX 1024 /* cap outstanding heartbeats */ +#define HB_BUCKETS 256 /* nonce hash buckets (pow2) */ + +/* RIB flow entry: the CA stats string plus the flow header. */ +#define FA_RIB_STRLEN (CA_STATS_STRLEN + 512) + +/* Outstanding heartbeat: send time kept locally, keyed by the nonce. */ +struct hb_ent { + uint8_t id[HB_ID_LEN]; + uint64_t s_eid; /* originating flow (fd-reuse guard) */ + uint64_t t_snd; /* send timestamp (ns) */ + struct list_head hnext; /* nonce hash bucket chain */ + struct list_head qnext; /* expiry FIFO, oldest at head */ +}; + struct fa_flow { #ifdef IPCP_FLOW_STATS time_t stamp; /* Flow creation */ @@ -109,6 +134,8 @@ struct fa_flow { uint64_t r_eid; /* Remote endpoint id */ uint64_t r_addr; /* Remote address */ void * ctx; /* Congestion avoidance context */ + uint64_t fair; /* SFQ virtual finish tag (bytes) */ + uint8_t l_ecn; /* Local first-hop mark (relaxed) */ }; struct { @@ -124,6 +151,11 @@ struct { pthread_mutex_t mtx; pthread_t worker; + struct list_head hb_bkt[HB_BUCKETS]; /* nonce hash buckets */ + struct list_head hb_q; /* expiry FIFO, oldest at head */ + size_t n_hbs; + pthread_mutex_t hb_mtx; + struct psched * psched; } fa; @@ -138,7 +170,7 @@ static int fa_rib_read(const char * path, char s_eidstr[21]; char r_eidstr[21]; char tmstr[RIB_TM_STRLEN]; - char castr[1024]; + char castr[CA_STATS_STRLEN]; char * entry; struct tm * tm; @@ -150,7 +182,7 @@ static int fa_rib_read(const char * path, if (fd < 0 || fd >= PROC_MAX_FLOWS) return -1; - if (len < 1536) + if (len < FA_RIB_STRLEN) return 0; flow = &fa.flows[fd]; @@ -171,7 +203,7 @@ static int fa_rib_read(const char * path, tm = gmtime(&flow->stamp); strftime(tmstr, sizeof(tmstr), RIB_TM_FORMAT, tm); - ca_print_stats(flow->ctx, castr, 1024); + ca_print_stats(flow->ctx, castr, CA_STATS_STRLEN); sprintf(buf, "Flow established at: %20s\n" @@ -245,6 +277,7 @@ static int fa_rib_readdir(char *** buf) fail_entry: while (idx-- > 0) free((*buf)[idx]); + free(*buf); fail_entries: pthread_rwlock_unlock(&fa.flows_lock); @@ -268,7 +301,7 @@ static int fa_rib_getattr(const char * path, pthread_rwlock_rdlock(&fa.flows_lock); if (flow->stamp != 0) { - attr->size = 1536; + attr->size = FA_RIB_STRLEN; attr->mtime = flow->stamp; } else { attr->size = 0; @@ -317,18 +350,140 @@ static uint64_t gen_eid(int fd) return ((uint64_t) rnd << 32) + fd; } -static void packet_handler(int fd, - qoscube_t qc, - struct ssm_pk_buff * spb) +/* The nonce is uniformly random, so its low word is a fine hash. */ +static size_t fa_hb_hash(const uint8_t * id) +{ + uint32_t h; + + memcpy(&h, id, sizeof(h)); + + return h & (HB_BUCKETS - 1); +} + +/* Record an outstanding heartbeat; expire stale entries as we go. */ +static void fa_hb_record(const uint8_t * id, + uint64_t s_eid, + uint64_t t_snd) +{ + struct hb_ent * ent; + struct list_head * p; + struct list_head * h; + + ent = malloc(sizeof(*ent)); + if (ent == NULL) + return; + + memcpy(ent->id, id, HB_ID_LEN); + ent->s_eid = s_eid; + ent->t_snd = t_snd; + + pthread_mutex_lock(&fa.hb_mtx); + + /* The FIFO is time-ordered; stop at the first fresh entry. */ + list_for_each_safe(p, h, &fa.hb_q) { + struct hb_ent * e = list_entry(p, struct hb_ent, qnext); + if (t_snd - e->t_snd <= HB_TBL_TTL) + break; + list_del(&e->hnext); + list_del(&e->qnext); + free(e); + fa.n_hbs--; + } + + if (fa.n_hbs >= HB_TBL_MAX) { + pthread_mutex_unlock(&fa.hb_mtx); + free(ent); + return; + } + + list_add(&ent->hnext, &fa.hb_bkt[fa_hb_hash(id)]); + list_add_tail(&ent->qnext, &fa.hb_q); + fa.n_hbs++; + + pthread_mutex_unlock(&fa.hb_mtx); +} + +/* Consume a heartbeat nonce, returning the flow and send time it maps to. */ +static int fa_hb_match(const uint8_t * id, + uint64_t * s_eid, + uint64_t * t_snd) +{ + struct list_head * bkt; + struct list_head * p; + struct list_head * h; + + pthread_mutex_lock(&fa.hb_mtx); + + bkt = &fa.hb_bkt[fa_hb_hash(id)]; + list_for_each_safe(p, h, bkt) { + struct hb_ent * e = list_entry(p, struct hb_ent, hnext); + if (memcmp(e->id, id, HB_ID_LEN) == 0) { + *s_eid = e->s_eid; + *t_snd = e->t_snd; + list_del(&e->hnext); + list_del(&e->qnext); + free(e); + fa.n_hbs--; + pthread_mutex_unlock(&fa.hb_mtx); + return 0; + } + } + + pthread_mutex_unlock(&fa.hb_mtx); + + return -1; +} + +/* Send a bare control message (heartbeat or ack) carrying only a nonce. */ +static int fa_send_ctrl(uint64_t r_addr, + uint8_t code, + const uint8_t * id) +{ + struct fa_msg * msg; + struct ssm_pk_buff * spb; + qoscube_t qc = QOS_CUBE_BE; + + if (ipcp_spb_reserve(&spb, sizeof(*msg))) + return -1; + + msg = (struct fa_msg *) ssm_pk_buff_head(spb); + memset(msg, 0, sizeof(*msg)); + + msg->code = code; + msg->s_addr = hton64(addr_auth_address()); + memcpy(msg->hb_id, id, HB_ID_LEN); + + if (dt_write_packet(r_addr, qc, fa.eid, spb, NULL)) { + ipcp_spb_release(spb); + return -1; + } + + return 0; +} + +static time_t packet_handler(int fd, + qoscube_t qc, + struct ssm_pk_buff * spb) { struct fa_flow * flow; + struct timespec tv; + uint64_t now; uint64_t r_addr; uint64_t r_eid; - ca_wnd_t wnd; + uint64_t s_eid; + bool hb; + uint8_t nonce[HB_ID_LEN]; + time_t wait; size_t len; + uint8_t ecn; flow = &fa.flows[fd]; + ecn = 0; + + clock_gettime(PTHREAD_COND_CLOCK, &tv); + now = TS_TO_UINT64(tv); + pthread_rwlock_wrlock(&fa.flows_lock); len = ssm_pk_buff_len(spb); @@ -337,16 +492,23 @@ static void packet_handler(int fd, ++flow->p_snd; flow->b_snd += len; #endif - wnd = ca_ctx_update_snd(flow->ctx, len); + wait = ca_ctx_update_snd(flow->ctx, len, + LOAD_RELAXED(&flow->l_ecn), &flow->fair); + hb = ca_ctx_hb_due(flow->ctx, now); r_addr = flow->r_addr; r_eid = flow->r_eid; + s_eid = flow->s_eid; pthread_rwlock_unlock(&fa.flows_lock); - ca_wnd_wait(wnd); + if (hb && random_buffer(nonce, HB_ID_LEN) == 0) { + fa_hb_record(nonce, s_eid, now); + fa_send_ctrl(r_addr, FLOW_HB, nonce); + } - if (dt_write_packet(r_addr, qc, r_eid, spb)) { + if (dt_write_packet(r_addr, qc, r_eid, spb, &ecn)) { + STORE_RELAXED(&flow->l_ecn, ecn); ipcp_spb_release(spb); log_dbg("Failed to forward packet."); #ifdef IPCP_FLOW_STATS @@ -355,8 +517,12 @@ static void packet_handler(int fd, flow->b_snd_f += len; pthread_rwlock_unlock(&fa.flows_lock); #endif - return; + return wait; } + + STORE_RELAXED(&flow->l_ecn, ecn); + + return wait; } static int fa_flow_init(struct fa_flow * flow) @@ -370,9 +536,7 @@ static int fa_flow_init(struct fa_flow * flow) flow->s_eid = -1; flow->r_addr = INVALID_ADDR; - flow->ctx = ca_ctx_create(); - if (flow->ctx == NULL) - return -1; + /* ctx is acquired once (r_addr, qc) are known (ca_ctx_get). */ #ifdef IPCP_FLOW_STATS clock_gettime(CLOCK_REALTIME_COARSE, &now); @@ -386,7 +550,8 @@ static int fa_flow_init(struct fa_flow * flow) static void fa_flow_fini(struct fa_flow * flow) { - ca_ctx_destroy(flow->ctx); + if (flow->ctx != NULL) + ca_ctx_put(flow->ctx); memset(flow, 0, sizeof(*flow)); @@ -488,6 +653,9 @@ static int fa_handle_flow_req(struct fa_msg * msg, qs.max_gap = ntoh32(msg->max_gap); qs.timeout = ntoh32(msg->timeout); + /* Ack the seed nonce now: a clean RTT, before any accept delay. */ + fa_send_ctrl(ntoh64(msg->s_addr), FLOW_ACK, msg->hb_id); + fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_UNICAST_MPL, IPCP_UNICAST_MTU, &data); if (fd < 0) @@ -503,6 +671,13 @@ static int fa_handle_flow_req(struct fa_msg * msg, flow->r_eid = ntoh64(msg->s_eid); flow->r_addr = ntoh64(msg->s_addr); + flow->ctx = ca_ctx_get(flow->r_addr, qos_spec_to_cube(qs)); + if (flow->ctx == NULL) { + fa_flow_fini(flow); + pthread_rwlock_unlock(&fa.flows_lock); + return -ENOMEM; + } + pthread_rwlock_unlock(&fa.flows_lock); return fd; @@ -580,7 +755,50 @@ static int fa_handle_flow_update(struct fa_msg * msg, #ifdef IPCP_FLOW_STATS flow->u_rcv++; #endif - ca_ctx_update_ece(flow->ctx, ntoh16(msg->ece)); + ca_ctx_update_ece(flow->ctx, ntoh16(msg->ece), msg->cap); + + pthread_rwlock_unlock(&fa.flows_lock); + + return 0; +} + +/* Heartbeat: reflect the nonce straight back to the sender's address. */ +static int fa_handle_flow_hb(struct fa_msg * msg, + size_t len) +{ + if (len < sizeof(*msg)) + return -EINVAL; + + return fa_send_ctrl(ntoh64(msg->s_addr), FLOW_ACK, msg->hb_id); +} + +/* Ack: the reflected nonce yields an RTT sample for its path. */ +static int fa_handle_flow_ack(struct fa_msg * msg, + size_t len) +{ + struct timespec tv; + struct fa_flow * flow; + uint64_t now; + uint64_t t_snd; + uint64_t s_eid; + int fd; + + if (len < sizeof(*msg)) + return -EINVAL; + + if (fa_hb_match(msg->hb_id, &s_eid, &t_snd) < 0) + return 0; /* unknown or stale nonce */ + + clock_gettime(PTHREAD_COND_CLOCK, &tv); + now = TS_TO_UINT64(tv); + + pthread_rwlock_wrlock(&fa.flows_lock); + + fd = eid_to_fd(s_eid); + if (fd >= 0 && now > t_snd) { + flow = &fa.flows[fd]; + ca_ctx_rtt(flow->ctx, now, now - t_snd); + } pthread_rwlock_unlock(&fa.flows_lock); @@ -656,6 +874,14 @@ static void * fa_handle_packet(void * o) if (fa_handle_flow_irm_update(msg, len) < 0) log_err("Error handling flow update."); break; + case FLOW_HB: + if (fa_handle_flow_hb(msg, len) < 0) + log_err("Error handling heartbeat."); + break; + case FLOW_ACK: + if (fa_handle_flow_ack(msg, len) < 0) + log_err("Error handling heartbeat ack."); + break; default: log_warn("Recieved unknown flow allocation message."); break; @@ -668,24 +894,28 @@ static void * fa_handle_packet(void * o) int fa_init(void) { pthread_condattr_t cattr; + size_t i; - if (pthread_rwlock_init(&fa.flows_lock, NULL)) + if (pthread_rwlock_init(&fa.flows_lock, NULL) != 0) goto fail_rwlock; - if (pthread_mutex_init(&fa.mtx, NULL)) + if (pthread_mutex_init(&fa.mtx, NULL) != 0) goto fail_mtx; - if (pthread_condattr_init(&cattr)) + if (pthread_mutex_init(&fa.hb_mtx, NULL) != 0) + goto fail_hb_mtx; + + if (pthread_condattr_init(&cattr) != 0) goto fail_cattr; #ifndef __APPLE__ pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK); #endif - if (pthread_cond_init(&fa.cond, &cattr)) + if (pthread_cond_init(&fa.cond, &cattr) != 0) goto fail_cond; #ifdef IPCP_FLOW_STATS - if (rib_reg(FA, &r_ops)) + if (rib_reg(FA, &r_ops) != 0) goto fail_rib_reg; #endif @@ -694,6 +924,11 @@ int fa_init(void) goto fail_dt_reg; list_head_init(&fa.cmds); + for (i = 0; i < HB_BUCKETS; i++) + list_head_init(&fa.hb_bkt[i]); + + list_head_init(&fa.hb_q); + fa.n_hbs = 0; pthread_condattr_destroy(&cattr); @@ -708,6 +943,8 @@ int fa_init(void) fail_cond: pthread_condattr_destroy(&cattr); fail_cattr: + pthread_mutex_destroy(&fa.hb_mtx); + fail_hb_mtx: pthread_mutex_destroy(&fa.mtx); fail_mtx: pthread_rwlock_destroy(&fa.flows_lock); @@ -717,10 +954,20 @@ int fa_init(void) void fa_fini(void) { + struct list_head * p; + struct list_head * h; + #ifdef IPCP_FLOW_STATS rib_unreg(FA); #endif + list_for_each_safe(p, h, &fa.hb_q) { + struct hb_ent * e = list_entry(p, struct hb_ent, qnext); + list_del(&e->qnext); + free(e); + } + pthread_cond_destroy(&fa.cond);; + pthread_mutex_destroy(&fa.hb_mtx); pthread_mutex_destroy(&fa.mtx); pthread_rwlock_destroy(&fa.flows_lock); } @@ -803,6 +1050,8 @@ int fa_alloc(int fd, qoscube_t qc = QOS_CUBE_BE; size_t len; uint64_t eid; + struct timespec tv; + uint8_t nonce[HB_ID_LEN]; addr = dir_query(dst); if (addr == 0) @@ -830,11 +1079,18 @@ int fa_alloc(int fd, msg->max_gap = hton32(qs.max_gap); msg->timeout = hton32(qs.timeout); + /* Seed an early RTT sample: peer acks this nonce on arrival. */ + if (random_buffer(nonce, HB_ID_LEN) == 0) { + memcpy(msg->hb_id, nonce, HB_ID_LEN); + clock_gettime(PTHREAD_COND_CLOCK, &tv); + fa_hb_record(nonce, eid, TS_TO_UINT64(tv)); + } + memcpy(msg + 1, dst, ipcp_dir_hash_len()); if (data->len > 0) memcpy(ssm_pk_buff_head(spb) + len, data->data, data->len); - if (dt_write_packet(addr, qc, fa.eid, spb)) { + if (dt_write_packet(addr, qc, fa.eid, spb, NULL)) { log_err("Failed to send flow allocation request packet."); ipcp_spb_release(spb); return -1; @@ -848,6 +1104,13 @@ int fa_alloc(int fd, flow->r_addr = addr; flow->s_eid = eid; + flow->ctx = ca_ctx_get(addr, qos_spec_to_cube(qs)); + if (flow->ctx == NULL) { + fa_flow_fini(flow); + pthread_rwlock_unlock(&fa.flows_lock); + return -1; + } + pthread_rwlock_unlock(&fa.flows_lock); return 0; @@ -890,7 +1153,7 @@ int fa_alloc_resp(int fd, pthread_rwlock_unlock(&fa.flows_lock); - if (dt_write_packet(flow->r_addr, qc, fa.eid, spb)) { + if (dt_write_packet(flow->r_addr, qc, fa.eid, spb, NULL)) { log_err("Failed to send flow allocation response packet."); goto fail_packet; } @@ -944,7 +1207,7 @@ int fa_irm_update(int fd, pthread_rwlock_unlock(&fa.flows_lock); - if (dt_write_packet(r_addr, qc, fa.eid, spb)) { + if (dt_write_packet(r_addr, qc, fa.eid, spb, NULL)) { log_err("Failed to send flow update packet."); ipcp_spb_release(spb); return -1; @@ -972,7 +1235,8 @@ int fa_dealloc(int fd) } static int fa_update_remote(int fd, - uint16_t ece) + uint16_t ece, + uint8_t cap) { struct fa_msg * msg; struct ssm_pk_buff * spb; @@ -996,6 +1260,7 @@ static int fa_update_remote(int fd, msg->code = FLOW_UPDATE; msg->r_eid = hton64(flow->r_eid); msg->ece = hton16(ece); + msg->cap = cap; r_addr = flow->r_addr; #ifdef IPCP_FLOW_STATS @@ -1004,7 +1269,7 @@ static int fa_update_remote(int fd, pthread_rwlock_unlock(&fa.flows_lock); - if (dt_write_packet(r_addr, qc, fa.eid, spb)) { + if (dt_write_packet(r_addr, qc, fa.eid, spb, NULL)) { log_err("Failed to send flow update packet."); ipcp_spb_release(spb); return -1; @@ -1015,11 +1280,13 @@ static int fa_update_remote(int fd, void fa_np1_rcv(uint64_t eid, uint8_t ecn, + uint8_t cap, struct ssm_pk_buff * spb) { struct fa_flow * flow; bool update; uint16_t ece; + uint8_t fcap; int fd; size_t len; @@ -1041,7 +1308,7 @@ void fa_np1_rcv(uint64_t eid, ++flow->p_rcv; flow->b_rcv += len; #endif - update = ca_ctx_update_rcv(flow->ctx, len, ecn, &ece); + update = ca_ctx_update_rcv(flow->ctx, len, ecn, cap, &ece, &fcap); pthread_rwlock_unlock(&fa.flows_lock); @@ -1057,5 +1324,5 @@ void fa_np1_rcv(uint64_t eid, } if (update) - fa_update_remote(eid, ece); + fa_update_remote(eid, ece, fcap); } diff --git a/src/ipcpd/unicast/fa.h b/src/ipcpd/unicast/fa.h index f31b40e9..1d0012b0 100644 --- a/src/ipcpd/unicast/fa.h +++ b/src/ipcpd/unicast/fa.h @@ -50,6 +50,7 @@ int fa_irm_update(int fd, void fa_np1_rcv(uint64_t eid, uint8_t ecn, + uint8_t cap, struct ssm_pk_buff * spb); #endif /* OUROBOROS_IPCPD_UNICAST_FA_H */ diff --git a/src/ipcpd/unicast/main.c b/src/ipcpd/unicast/main.c index 1155b88b..86cb1f06 100644 --- a/src/ipcpd/unicast/main.c +++ b/src/ipcpd/unicast/main.c @@ -67,7 +67,7 @@ static int initialize_components(struct ipcp_config * conf) log_info("IPCP got address %" PRIu64 ".", addr_auth_address()); - if (ca_init(conf->unicast.cong_avoid)) { + if (ca_init(conf->unicast.cong_avoid, conf->unicast.dt.max_rtt)) { log_err("Failed to initialize congestion avoidance."); goto fail_ca; } diff --git a/src/ipcpd/unicast/pff/multipath.c b/src/ipcpd/unicast/pff/multipath.c index 9ba59592..c2c21078 100644 --- a/src/ipcpd/unicast/pff/multipath.c +++ b/src/ipcpd/unicast/pff/multipath.c @@ -49,7 +49,7 @@ struct pff_ops multipath_pff_ops = { .del = multipath_pff_del, .flush = multipath_pff_flush, .nhop = multipath_pff_nhop, - .flow_state_change = NULL + .flow_state_change = multipath_pff_flow_state_change }; struct pff_i * multipath_pff_create(void) @@ -170,6 +170,24 @@ void multipath_pff_flush(struct pff_i * pff_i) pft_flush(pff_i->pft); } +int multipath_pff_flow_state_change(struct pff_i * pff_i, + int fd, + bool up) +{ + assert(pff_i); + + if (up) + return 0; + + pthread_rwlock_wrlock(&pff_i->lock); + + pft_del_fd(pff_i->pft, fd); + + pthread_rwlock_unlock(&pff_i->lock); + + return 0; +} + int multipath_pff_nhop(struct pff_i * pff_i, uint64_t addr) { diff --git a/src/ipcpd/unicast/pff/multipath.h b/src/ipcpd/unicast/pff/multipath.h index 5329f7fc..123030b6 100644 --- a/src/ipcpd/unicast/pff/multipath.h +++ b/src/ipcpd/unicast/pff/multipath.h @@ -53,6 +53,10 @@ void multipath_pff_flush(struct pff_i * pff_i); int multipath_pff_nhop(struct pff_i * pff_i, uint64_t addr); +int multipath_pff_flow_state_change(struct pff_i * pff_i, + int fd, + bool up); + extern struct pff_ops multipath_pff_ops; #endif /* OUROBOROS_IPCPD_UNICAST_MULTIPATH_PFF_H */ diff --git a/src/ipcpd/unicast/pff/pft.c b/src/ipcpd/unicast/pff/pft.c index d0e562d6..2a295a40 100644 --- a/src/ipcpd/unicast/pff/pft.c +++ b/src/ipcpd/unicast/pff/pft.c @@ -91,6 +91,43 @@ void pft_destroy(struct pft * pft) free(pft); } +void pft_del_fd(struct pft * pft, + int fd) +{ + unsigned int i; + struct list_head * p; + struct list_head * h; + struct pft_entry * entry; + size_t j; + size_t n; + + assert(pft); + + for (i = 0; i < pft->buckets_size; i++) { + list_for_each_safe(p, h, &(pft->buckets[i])) { + entry = list_entry(p, struct pft_entry, next); + + n = 0; + for (j = 0; j < entry->len; j++) { + if (entry->fds[j] != fd) + entry->fds[n++] = entry->fds[j]; + } + + if (n == entry->len) + continue; + + if (n > 0) { + entry->len = n; + continue; + } + + list_del(&entry->next); + free(entry->fds); + free(entry); + } + } +} + void pft_flush(struct pft * pft) { unsigned int i; diff --git a/src/ipcpd/unicast/pff/pft.h b/src/ipcpd/unicast/pff/pft.h index 15bbe451..3517e0ef 100644 --- a/src/ipcpd/unicast/pff/pft.h +++ b/src/ipcpd/unicast/pff/pft.h @@ -35,6 +35,9 @@ void pft_destroy(struct pft * table); void pft_flush(struct pft * table); +void pft_del_fd(struct pft * table, + int fd); + /* Passes ownership of the block of memory */ int pft_insert(struct pft * pft, uint64_t dst, diff --git a/src/ipcpd/unicast/pff/simple.c b/src/ipcpd/unicast/pff/simple.c index 7befa42f..4347dcba 100644 --- a/src/ipcpd/unicast/pff/simple.c +++ b/src/ipcpd/unicast/pff/simple.c @@ -47,7 +47,7 @@ struct pff_ops simple_pff_ops = { .del = simple_pff_del, .flush = simple_pff_flush, .nhop = simple_pff_nhop, - .flow_state_change = NULL + .flow_state_change = simple_pff_flow_state_change }; struct pff_i * simple_pff_create(void) @@ -170,6 +170,24 @@ void simple_pff_flush(struct pff_i * pff_i) pft_flush(pff_i->pft); } +int simple_pff_flow_state_change(struct pff_i * pff_i, + int fd, + bool up) +{ + assert(pff_i); + + if (up) + return 0; + + pthread_rwlock_wrlock(&pff_i->lock); + + pft_del_fd(pff_i->pft, fd); + + pthread_rwlock_unlock(&pff_i->lock); + + return 0; +} + int simple_pff_nhop(struct pff_i * pff_i, uint64_t addr) { diff --git a/src/ipcpd/unicast/pff/simple.h b/src/ipcpd/unicast/pff/simple.h index 1046e4c4..b72aba21 100644 --- a/src/ipcpd/unicast/pff/simple.h +++ b/src/ipcpd/unicast/pff/simple.h @@ -52,6 +52,10 @@ void simple_pff_flush(struct pff_i * pff_i); int simple_pff_nhop(struct pff_i * pff_i, uint64_t addr); +int simple_pff_flow_state_change(struct pff_i * pff_i, + int fd, + bool up); + extern struct pff_ops simple_pff_ops; #endif /* OUROBOROS_IPCPD_UNICAST_SIMPLE_PFF_H */ diff --git a/src/ipcpd/unicast/pff/tests/pft_test.c b/src/ipcpd/unicast/pff/tests/pft_test.c index 20e73a94..0b4a165b 100644 --- a/src/ipcpd/unicast/pff/tests/pft_test.c +++ b/src/ipcpd/unicast/pff/tests/pft_test.c @@ -22,97 +22,321 @@ #include "pft.c" +#include <test/test.h> + #include <stdio.h> #define TBL_SIZE 256 #define INT_TEST 4 -int pft_test(int argc, - char ** argv) +/* Next hops used by the del_fd tests. */ +#define FD_GONE 7 +#define FD_KEEP 8 +#define FD_OTHER 9 + +static int pft_add(struct pft * pft, + uint64_t dst, + const int * fds, + size_t len) +{ + int * blk; + size_t i; + + blk = malloc(sizeof(*blk) * len); + if (blk == NULL) + return -1; + + for (i = 0; i < len; i++) + blk[i] = fds[i]; + + if (pft_insert(pft, dst, blk, len)) { + free(blk); + return -1; + } + + return 0; +} + +static int test_pft_create_destroy(void) +{ + struct pft * pft; + + TEST_START(); + + pft = pft_create(TBL_SIZE); + if (pft == NULL) { + printf("Failed to create.\n"); + goto fail; + } + + pft_destroy(pft); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_pft_insert_lookup(void) { struct pft * pft; - int i; int * j; size_t len; + int i; - (void) argc; - (void) argv; + TEST_START(); pft = pft_create(TBL_SIZE); if (pft == NULL) { printf("Failed to create.\n"); - return -1; + goto fail; } for (i = 0; i < TBL_SIZE + INT_TEST + 2; i++) { - j = malloc(sizeof(*j)); - if (j == NULL) { - printf("Failed to malloc.\n"); - pft_destroy(pft); - return -1; - } - *j = i; - - if (pft_insert(pft, i, j, 1)) { + if (pft_add(pft, i, &i, 1)) { printf("Failed to insert.\n"); - pft_destroy(pft); - free(j); - return -1; + goto fail_pft; } } if (pft_lookup(pft, INT_TEST, &j, &len)) { printf("Failed to lookup.\n"); - pft_destroy(pft); - return -1; + goto fail_pft; } if (*j != INT_TEST) { printf("Lookup returned wrong value (%d != %d).\n", INT_TEST, *j); - pft_destroy(pft); - return -1; + goto fail_pft; } if (pft_lookup(pft, TBL_SIZE + INT_TEST, &j, &len)) { - printf("Failed to lookup.\n"); - pft_destroy(pft); - return -1; + printf("Failed to lookup on a shared bucket.\n"); + goto fail_pft; } if (*j != TBL_SIZE + INT_TEST) { printf("Lookup returned wrong value (%d != %d).\n", - INT_TEST, *j); - pft_destroy(pft); - return -1; + TBL_SIZE + INT_TEST, *j); + goto fail_pft; + } + + pft_destroy(pft); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_pft: + pft_destroy(pft); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_pft_delete(void) +{ + struct pft * pft; + int * j; + size_t len; + int i; + + TEST_START(); + + pft = pft_create(TBL_SIZE); + if (pft == NULL) { + printf("Failed to create.\n"); + goto fail; + } + + for (i = 0; i < TBL_SIZE + INT_TEST + 2; i++) { + if (pft_add(pft, i, &i, 1)) { + printf("Failed to insert.\n"); + goto fail_pft; + } } if (pft_delete(pft, INT_TEST)) { printf("Failed to delete.\n"); - pft_destroy(pft); - return -1; + goto fail_pft; } if (pft_lookup(pft, INT_TEST, &j, &len) == 0) { printf("Failed to delete properly.\n"); - pft_destroy(pft); - return -1; + goto fail_pft; } if (pft_lookup(pft, TBL_SIZE + INT_TEST, &j, &len)) { printf("Failed to lookup after deletion.\n"); - pft_destroy(pft); - return -1; + goto fail_pft; } if (*j != TBL_SIZE + INT_TEST) { printf("Lookup returned wrong value (%d != %d).\n", - INT_TEST, *j); - pft_destroy(pft); - return -1; + TBL_SIZE + INT_TEST, *j); + goto fail_pft; } pft_destroy(pft); - return 0; + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_pft: + pft_destroy(pft); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_pft_del_fd_sole(void) +{ + struct pft * pft; + int * j; + size_t len; + int fds[] = {FD_GONE}; + + TEST_START(); + + pft = pft_create(TBL_SIZE); + if (pft == NULL) { + printf("Failed to create.\n"); + goto fail; + } + + if (pft_add(pft, INT_TEST, fds, 1)) { + printf("Failed to insert.\n"); + goto fail_pft; + } + + pft_del_fd(pft, FD_GONE); + + if (pft_lookup(pft, INT_TEST, &j, &len) == 0) { + printf("Route without a next hop survived.\n"); + goto fail_pft; + } + + pft_destroy(pft); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_pft: + pft_destroy(pft); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_pft_del_fd_shared(void) +{ + struct pft * pft; + int * j; + size_t len; + int fds[] = {FD_KEEP, FD_GONE, FD_OTHER}; + + TEST_START(); + + pft = pft_create(TBL_SIZE); + if (pft == NULL) { + printf("Failed to create.\n"); + goto fail; + } + + if (pft_add(pft, INT_TEST, fds, 3)) { + printf("Failed to insert.\n"); + goto fail_pft; + } + + pft_del_fd(pft, FD_GONE); + + if (pft_lookup(pft, INT_TEST, &j, &len)) { + printf("Route with next hops left was dropped.\n"); + goto fail_pft; + } + + if (len != 2) { + printf("Expected 2 next hops, got %zu.\n", len); + goto fail_pft; + } + + if (j[0] != FD_KEEP || j[1] != FD_OTHER) { + printf("Next hops not preserved in order (%d, %d).\n", + j[0], j[1]); + goto fail_pft; + } + + pft_destroy(pft); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_pft: + pft_destroy(pft); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_pft_del_fd_untouched(void) +{ + struct pft * pft; + int * j; + size_t len; + int fds[] = {FD_KEEP}; + + TEST_START(); + + pft = pft_create(TBL_SIZE); + if (pft == NULL) { + printf("Failed to create.\n"); + goto fail; + } + + if (pft_add(pft, INT_TEST, fds, 1)) { + printf("Failed to insert.\n"); + goto fail_pft; + } + + pft_del_fd(pft, FD_GONE); + + if (pft_lookup(pft, INT_TEST, &j, &len)) { + printf("Unrelated route was dropped.\n"); + goto fail_pft; + } + + if (len != 1 || *j != FD_KEEP) { + printf("Unrelated route was modified.\n"); + goto fail_pft; + } + + pft_destroy(pft); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_pft: + pft_destroy(pft); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +int pft_test(int argc, + char ** argv) +{ + int ret = 0; + + (void) argc; + (void) argv; + + ret |= test_pft_create_destroy(); + ret |= test_pft_insert_lookup(); + ret |= test_pft_delete(); + ret |= test_pft_del_fd_sole(); + ret |= test_pft_del_fd_shared(); + ret |= test_pft_del_fd_untouched(); + + return ret; } diff --git a/src/ipcpd/unicast/psched.c b/src/ipcpd/unicast/psched.c index 21e23617..01c4b066 100644 --- a/src/ipcpd/unicast/psched.c +++ b/src/ipcpd/unicast/psched.c @@ -30,6 +30,7 @@ #include <ouroboros/errno.h> #include <ouroboros/notifier.h> +#include <ouroboros/time.h> #include "common/connmgr.h" #include "ipcp.h" @@ -50,7 +51,7 @@ static int qos_prio [] = { #endif struct psched { - fset_t * set[QOS_CUBE_MAX]; + fset_t * set[QOS_CUBE_MAX * IPCP_SCHED_THR_MUL]; next_packet_fn_t callback; read_fn_t read; pthread_t readers[QOS_CUBE_MAX * IPCP_SCHED_THR_MUL]; @@ -59,23 +60,155 @@ struct psched { struct sched_info { struct psched * sch; qoscube_t qc; + size_t idx; }; +/* Map an FD to one reader's set: one FD, one thread (no shared FDs). */ +static size_t fd_set_idx(int fd, qoscube_t qc) +{ + return qc + ((size_t) fd % IPCP_SCHED_THR_MUL) * QOS_CUBE_MAX; +} + static void cleanup_reader(void * o) { fqueue_destroy((fqueue_t *) o); } +/* + * Per-reader deadline scheduler: a paced flow is served, then deferred + * to its next-send deadline instead of blocking the thread, so it never + * stalls its thread-mates. + */ +struct dsched { + uint64_t deadline[PROC_MAX_FLOWS]; /* absolute ns, per tracked fd */ + int active[PROC_MAX_FLOWS]; /* compact list of tracked fds */ + int posn[PROC_MAX_FLOWS]; /* fd -> active index, -1 = none */ + size_t n; +}; + +static void cleanup_dsched(void * o) +{ + free(o); +} + +static void dsched_track(struct dsched * d, + int fd, + uint64_t deadline) +{ + if (d->posn[fd] >= 0) + return; + + d->deadline[fd] = deadline; + d->posn[fd] = (int) d->n; + d->active[d->n++] = fd; +} + +static void dsched_untrack(struct dsched * d, + int fd) +{ + int i = d->posn[fd]; + + if (i < 0) + return; + + d->active[i] = d->active[--d->n]; + d->posn[d->active[i]] = i; + d->posn[fd] = -1; +} + +/* Fold a deadline into the earliest pending one (0 = none yet). */ +static uint64_t dmin_fold(uint64_t dmin, + uint64_t deadline) +{ + if (dmin == 0 || deadline < dmin) + return deadline; + + return dmin; +} + +static uint64_t dsched_serve(struct dsched * d, + struct psched * sched, + qoscube_t qc, + uint64_t now) +{ + struct ssm_pk_buff * spb; + uint64_t dmin = 0; + size_t i; + int fd; + time_t wait; + bool served; + + /* Round-robin one packet per flow so none monopolises egress. */ + do { + served = false; + + for (i = 0; i < d->n; ) { + fd = d->active[i]; + + if (d->deadline[fd] > now) { + dmin = dmin_fold(dmin, d->deadline[fd]); + ++i; + continue; + } + + if (sched->read(fd, &spb) < 0) { + dsched_untrack(d, fd); + continue; + } + + wait = sched->callback(fd, qc, spb); + served = true; + + if (wait > 0) { + d->deadline[fd] = now + (uint64_t) wait; + dmin = dmin_fold(dmin, d->deadline[fd]); + } + + ++i; + } + } while (served); + + return dmin; +} + +static void dsched_events(struct dsched * d, + fqueue_t * fq, + uint64_t now) +{ + int fd; + + while ((fd = fqueue_next(fq)) >= 0) { + switch (fqueue_type(fq)) { + case FLOW_DEALLOC: + dsched_untrack(d, fd); + notifier_event(NOTIFY_DT_FLOW_DEALLOC, &fd); + break; + case FLOW_DOWN: + notifier_event(NOTIFY_DT_FLOW_DOWN, &fd); + break; + case FLOW_UP: + notifier_event(NOTIFY_DT_FLOW_UP, &fd); + break; + case FLOW_PKT: + dsched_track(d, fd, now); + break; + default: + break; + } + } +} + static void * packet_reader(void * o) { - struct psched * sched; - struct ssm_pk_buff * spb; - int fd; - fqueue_t * fq; - qoscube_t qc; + struct psched * sched; + struct dsched * d; + fqueue_t * fq; + qoscube_t qc; + size_t idx; sched = ((struct sched_info *) o)->sch; qc = ((struct sched_info *) o)->qc; + idx = ((struct sched_info *) o)->idx; ipcp_lock_to_core(); @@ -85,37 +218,51 @@ static void * packet_reader(void * o) if (fq == NULL) return (void *) -1; + d = malloc(sizeof(*d)); + if (d == NULL) { + fqueue_destroy(fq); + return (void *) -1; + } + + memset(d, 0, sizeof(*d)); + memset(d->posn, 0xFF, sizeof(d->posn)); /* -1: nothing tracked yet */ + + pthread_cleanup_push(cleanup_dsched, d); pthread_cleanup_push(cleanup_reader, fq); while (true) { - int ret = fevent(sched->set[qc], fq, NULL); + struct timespec now_ts; + struct timespec to; + struct timespec * timeo; + uint64_t now; + uint64_t dmin; + uint64_t delta; + int ret; + + clock_gettime(PTHREAD_COND_CLOCK, &now_ts); + + now = TS_TO_UINT64(now_ts); + + dmin = dsched_serve(d, sched, qc, now); + + if (dmin == 0) { + timeo = NULL; + } else { + delta = dmin > now ? dmin - now : 1; + to.tv_sec = (time_t) (delta / BILLION); + to.tv_nsec = (long) (delta % BILLION); + timeo = &to; + } + + ret = fevent(sched->set[idx], fq, timeo); if (ret < 0) continue; - while ((fd = fqueue_next(fq)) >= 0) { - switch (fqueue_type(fq)) { - case FLOW_DEALLOC: - notifier_event(NOTIFY_DT_FLOW_DEALLOC, &fd); - break; - case FLOW_DOWN: - notifier_event(NOTIFY_DT_FLOW_DOWN, &fd); - break; - case FLOW_UP: - notifier_event(NOTIFY_DT_FLOW_UP, &fd); - break; - case FLOW_PKT: - if (sched->read(fd, &spb) < 0) - continue; - - sched->callback(fd, qc, spb); - break; - default: - break; - } - } + dsched_events(d, fq, now); } pthread_cleanup_pop(true); + pthread_cleanup_pop(true); return (void *) 0; } @@ -137,7 +284,7 @@ struct psched * psched_create(next_packet_fn_t callback, psched->callback = callback; psched->read = read; - for (i = 0; i < QOS_CUBE_MAX; ++i) { + for (i = 0; i < QOS_CUBE_MAX * IPCP_SCHED_THR_MUL; ++i) { psched->set[i] = fset_create(); if (psched->set[i] == NULL) { for (j = 0; j < i; ++j) @@ -155,6 +302,7 @@ struct psched * psched_create(next_packet_fn_t callback, } infos[i]->sch = psched; infos[i]->qc = i % QOS_CUBE_MAX; + infos[i]->idx = i; } for (i = 0; i < QOS_CUBE_MAX * IPCP_SCHED_THR_MUL; ++i) { @@ -196,11 +344,12 @@ struct psched * psched_create(next_packet_fn_t callback, fail_sched: for (j = 0; j < QOS_CUBE_MAX * IPCP_SCHED_THR_MUL; ++j) pthread_cancel(psched->readers[j]); + for (j = 0; j < QOS_CUBE_MAX * IPCP_SCHED_THR_MUL; ++j) pthread_join(psched->readers[j], NULL); #endif fail_infos: - for (j = 0; j < QOS_CUBE_MAX; ++j) + for (j = 0; j < QOS_CUBE_MAX * IPCP_SCHED_THR_MUL; ++j) fset_destroy(psched->set[j]); fail_flow_set: free(psched); @@ -219,7 +368,7 @@ void psched_destroy(struct psched * psched) pthread_join(psched->readers[i], NULL); } - for (i = 0; i < QOS_CUBE_MAX; ++i) + for (i = 0; i < QOS_CUBE_MAX * IPCP_SCHED_THR_MUL; ++i) fset_destroy(psched->set[i]); free(psched); @@ -233,7 +382,7 @@ void psched_add(struct psched * psched, assert(psched); ipcp_flow_get_qoscube(fd, &qc); - fset_add(psched->set[qc], fd); + fset_add(psched->set[fd_set_idx(fd, qc)], fd); } void psched_del(struct psched * psched, @@ -244,5 +393,5 @@ void psched_del(struct psched * psched, assert(psched); ipcp_flow_get_qoscube(fd, &qc); - fset_del(psched->set[qc], fd); + fset_del(psched->set[fd_set_idx(fd, qc)], fd); } diff --git a/src/ipcpd/unicast/psched.h b/src/ipcpd/unicast/psched.h index d83bb793..8c2914b3 100644 --- a/src/ipcpd/unicast/psched.h +++ b/src/ipcpd/unicast/psched.h @@ -26,9 +26,9 @@ #include <ouroboros/ipcp-dev.h> #include <ouroboros/fqueue.h> -typedef void (* next_packet_fn_t)(int fd, - qoscube_t qc, - struct ssm_pk_buff * spb); +typedef time_t (* next_packet_fn_t)(int fd, + qoscube_t qc, + struct ssm_pk_buff * spb); typedef int (* read_fn_t)(int fd, struct ssm_pk_buff ** spb); diff --git a/src/ipcpd/unicast/tests/CMakeLists.txt b/src/ipcpd/unicast/tests/CMakeLists.txt new file mode 100644 index 00000000..2e35ed66 --- /dev/null +++ b/src/ipcpd/unicast/tests/CMakeLists.txt @@ -0,0 +1,34 @@ +get_filename_component(CURRENT_SOURCE_PARENT_DIR + ${CMAKE_CURRENT_SOURCE_DIR} DIRECTORY) +get_filename_component(CURRENT_BINARY_PARENT_DIR + ${CMAKE_CURRENT_BINARY_DIR} DIRECTORY) + +get_filename_component(PARENT_PATH ${CMAKE_CURRENT_SOURCE_DIR} DIRECTORY) +get_filename_component(PARENT_DIR ${PARENT_PATH} NAME) + +compute_test_prefix() + +create_test_sourcelist(${PARENT_DIR}_tests test_suite.c + # Add new tests here + cap_test.c + ) + +add_executable(${PARENT_DIR}_test ${${PARENT_DIR}_tests}) + +target_include_directories(${PARENT_DIR}_test PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ${CMAKE_CURRENT_BINARY_DIR} + ${CURRENT_SOURCE_PARENT_DIR} + ${CURRENT_BINARY_PARENT_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}_test) +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}) diff --git a/src/ipcpd/unicast/tests/cap_test.c b/src/ipcpd/unicast/tests/cap_test.c new file mode 100644 index 00000000..7867b490 --- /dev/null +++ b/src/ipcpd/unicast/tests/cap_test.c @@ -0,0 +1,593 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Unit tests for link capacity estimation + * + * 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 "cap.c" + +#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) +{ + unsigned c; + + TEST_START(); + + for (c = 32; c <= 255; c++) { + if (cap_enc(cap_dec((uint8_t) c)) != c) { + printf("Code %u does not roundtrip.\n", c); + goto fail; + } + + if (cap_dec((uint8_t) c) <= cap_dec((uint8_t) (c - 1))) { + printf("Decode not monotone at %u.\n", c); + goto fail; + } + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_cap_codec_bounds(void) +{ + TEST_START(); + + if (cap_enc(0) != 0 || cap_dec(0) != 0) { + printf("Zero is not unknown.\n"); + goto fail; + } + + if (cap_enc(1) != 1) { + printf("Rate 1 encoded as %u.\n", cap_enc(1)); + goto fail; + } + + if (cap_enc(UINT64_MAX) != 255) { + printf("Max rate encoded as %u.\n", cap_enc(UINT64_MAX)); + goto fail; + } + + if (cap_dec(255) <= cap_dec(254)) { + printf("Top code does not decode.\n"); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_cap_min(void) +{ + TEST_START(); + + if (cap_min(0, 42) != 42 || cap_min(42, 0) != 42) { + printf("Unknown not skipped in min.\n"); + goto fail; + } + + if (cap_min(0, 0) != 0) { + printf("Two unknowns not unknown.\n"); + goto fail; + } + + if (cap_min(97, 42) != 42 || cap_min(42, 97) != 42) { + printf("Min not taken.\n"); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_cap_stamp(void) +{ + uint8_t pci; + + 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; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + 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) +{ + int ret = 0; + + (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/irmd/configfile.c b/src/irmd/configfile.c index 35cf4292..e2e1e554 100644 --- a/src/irmd/configfile.c +++ b/src/irmd/configfile.c @@ -457,7 +457,7 @@ static int toml_congestion(toml_table_t * table, if (congestion.ok) { if (strcmp(congestion.u.s, "none") == 0) conf->cong_avoid = CA_NONE; - else if (strcmp(congestion.u.s, "lfa") == 0) + else if (strcmp(congestion.u.s, "mb-ecn") == 0) conf->cong_avoid = CA_MB_ECN; else conf->cong_avoid = CA_INVALID; @@ -477,6 +477,7 @@ static int toml_dt(toml_table_t * table, toml_datum_t addr; toml_datum_t eid; toml_datum_t ttl; + toml_datum_t max_rtt; addr = toml_int_in(table, "addr_size"); if (addr.ok) @@ -490,6 +491,10 @@ static int toml_dt(toml_table_t * table, if (ttl.ok) conf->max_ttl = ttl.u.i; + max_rtt = toml_int_in(table, "max_rtt"); + if (max_rtt.ok) + conf->max_rtt = max_rtt.u.i; + if (toml_routing(table, conf) < 0) { log_err("Invalid routing option."); return -1; diff --git a/src/irmd/oap/cli.c b/src/irmd/oap/cli.c index 3518b4d1..ebfcd71f 100644 --- a/src/irmd/oap/cli.c +++ b/src/irmd/oap/cli.c @@ -291,6 +291,13 @@ int oap_cli_prepare(void ** ctx, goto fail_kex; } + /* A re-keyed flow is encrypted; absent config must fail closed. */ + if (rekey && !IS_KEX_ALGO_SET(&s->scfg)) { + log_err_id(s->id.data, "Refusing re-key without KEX for %s.", + info->name); + goto fail_kex; + } + /* Re-key forces server-encap: client-encap forfeits FS/PCS. */ if (rekey && s->scfg.x.mode == KEM_MODE_CLIENT_ENCAP) { s->scfg.x.mode = KEM_MODE_SERVER_ENCAP; @@ -385,26 +392,9 @@ static int do_client_kex_complete_kem(struct oap_cli_ctx * s, struct sec_config * scfg = &s->scfg; uint8_t * id = s->id.data; uint8_t key_buf[SYMMKEYSZ]; + buffer_t ct; - if (scfg->x.mode == KEM_MODE_SERVER_ENCAP) { - buffer_t ct; - - if (peer_hdr->kex.len == 0) { - log_err_id(id, "Server did not send KEM CT."); - return -ECRYPT; - } - - ct.data = peer_hdr->kex.data; - ct.len = peer_hdr->kex.len; - - if (kex_kem_decap(s->pkp, ct, scfg->k.nid, key_buf) < 0) { - log_err_id(id, "Failed to decapsulate KEM."); - return -ECRYPT; - } - - log_dbg_id(id, "Client decapsulated server CT."); - - } else if (scfg->x.mode == KEM_MODE_CLIENT_ENCAP) { + if (scfg->x.mode == KEM_MODE_CLIENT_ENCAP) { /* Key already derived during prepare */ memcpy(sk->key, s->key, SYMMKEYSZ); sk->nid = scfg->c.nid; @@ -413,6 +403,22 @@ static int do_client_kex_complete_kem(struct oap_cli_ctx * s, return 0; } + /* KEM_MODE_SERVER_ENCAP */ + if (peer_hdr->kex.len == 0) { + log_err_id(id, "Server did not send KEM CT."); + return -ECRYPT; + } + + ct.data = peer_hdr->kex.data; + ct.len = peer_hdr->kex.len; + + if (kex_kem_decap(s->pkp, ct, scfg->k.nid, key_buf) < 0) { + log_err_id(id, "Failed to decapsulate KEM."); + return -ECRYPT; + } + + log_dbg_id(id, "Client decapsulated server CT."); + memcpy(sk->key, key_buf, SYMMKEYSZ); sk->nid = scfg->c.nid; crypt_secure_clear(key_buf, SYMMKEYSZ); diff --git a/src/irmd/oap/hdr.c b/src/irmd/oap/hdr.c index 6f355133..0cff345c 100644 --- a/src/irmd/oap/hdr.c +++ b/src/irmd/oap/hdr.c @@ -292,9 +292,9 @@ static void write_oap_fixed(uint8_t * buf, kex_len |= OAP_KEX_ROLE_BIT; } - kex_len = hton16(kex_len); - memcpy(buf + offset, &kex_len, sizeof(kex_len)); - offset += sizeof(kex_len); + v = hton16(kex_len); + memcpy(buf + offset, &v, sizeof(v)); + offset += sizeof(v); v = hton16((uint16_t) data_len); memcpy(buf + offset, &v, sizeof(v)); diff --git a/src/irmd/oap/srv.c b/src/irmd/oap/srv.c index cc3dec5b..d78fc8d4 100644 --- a/src/irmd/oap/srv.c +++ b/src/irmd/oap/srv.c @@ -473,6 +473,12 @@ int oap_srv_process(const struct name_info * info, peer_crt->len = peer_hdr.crt.len; } + /* A re-keyed flow is encrypted; refuse a plaintext re-key. */ + if (rekey && peer_hdr.kex.len == 0) { + log_err_id(id, "Re-key request without KEX."); + goto fail_kex; + } + if (do_server_kex(info, &peer_hdr, &scfg, &local_hdr.kex, sk) < 0) goto fail_kex; diff --git a/src/irmd/oap/tests/oap_test.c b/src/irmd/oap/tests/oap_test.c index 145b68c7..b24bb786 100644 --- a/src/irmd/oap/tests/oap_test.c +++ b/src/irmd/oap/tests/oap_test.c @@ -269,6 +269,76 @@ static int test_oap_rekey_badcache_all(void) return ret; } +/* Absent sec config (ENOENT) clears the KEX; a re-key must fail closed. */ +static int test_oap_cli_rejects_rekey_no_kex(void) +{ + struct oap_test_ctx ctx; + + TEST_START(); + + test_enc_noauth_cfg(); + test_cfg.cli.kex = NID_undef; + + if (oap_test_setup(&ctx, root_ca_crt_ec, im_ca_crt_ec) < 0) + goto fail; + + ctx.rekey = true; + + if (oap_cli_prepare_ctx(&ctx) == 0) { + printf("Client prepared a re-key without KEX.\n"); + goto fail_cleanup; + } + + oap_test_teardown(&ctx); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_cleanup: + oap_test_teardown(&ctx); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_oap_srv_rejects_rekey_no_kex(void) +{ + struct oap_test_ctx ctx; + + TEST_START(); + + test_enc_noauth_cfg(); + test_cfg.cli.kex = NID_undef; + test_cfg.srv.kex = NID_undef; + + if (oap_test_setup(&ctx, root_ca_crt_ec, im_ca_crt_ec) < 0) + goto fail; + + /* First-contact plaintext request, replayed as a re-key. */ + if (oap_cli_prepare_ctx(&ctx) < 0) { + printf("Client prepare failed.\n"); + goto fail_cleanup; + } + + ctx.rekey = true; + + if (oap_srv_process_ctx(&ctx) == 0) { + printf("Server accepted a re-key without KEX.\n"); + goto fail_cleanup; + } + + oap_test_teardown(&ctx); + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail_cleanup: + oap_test_teardown(&ctx); + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + static int test_oap_roundtrip_kex_only(void) { test_enc_noauth_cfg(); @@ -1945,6 +2015,8 @@ int oap_test(int argc, ret |= test_oap_rekey_all(); ret |= test_oap_rekey_badcache_all(); ret |= test_oap_rekey_srv_badcache_all(); + ret |= test_oap_cli_rejects_rekey_no_kex(); + ret |= test_oap_srv_rejects_rekey_no_kex(); ret |= test_oap_roundtrip_all(); ret |= test_oap_roundtrip_md_all(); @@ -1987,6 +2059,11 @@ int oap_test(int argc, (void) test_oap_roundtrip_auth_only; (void) test_oap_roundtrip_kex_only; (void) test_oap_piggyback_data; + (void) test_oap_rekey_all; + (void) test_oap_rekey_badcache_all; + (void) test_oap_rekey_srv_badcache_all; + (void) test_oap_cli_rejects_rekey_no_kex; + (void) test_oap_srv_rejects_rekey_no_kex; (void) test_oap_roundtrip; (void) test_oap_roundtrip_all; (void) test_oap_roundtrip_md; @@ -1999,14 +2076,15 @@ int oap_test(int argc, (void) test_oap_deflated_length_field; (void) test_oap_nid_without_kex; (void) test_oap_unsupported_nid; + (void) test_oap_unsupported_nid_undefined; + (void) test_oap_unsupported_nid_all; (void) test_oap_cipher_mismatch; (void) test_oap_srv_enc_cli_none; (void) test_oap_cli_enc_srv_none; (void) test_oap_cli_rejects_downgrade; (void) test_oap_cli_rejects_suite_swap; (void) test_oap_srv_rejects_weak_kex; - (void) test_oap_outdated_packet; - (void) test_oap_future_packet; + (void) test_oap_ts_reject_all; (void) test_oap_replay_packet; (void) test_oap_replay_generations; (void) test_oap_missing_root_ca; @@ -2018,6 +2096,8 @@ int oap_test(int argc, (void) test_oap_srv_rejects_md_mismatch; (void) test_oap_server_cert_hidden; (void) test_oap_sealed_tamper; + (void) test_oap_cleartext_echo_tamper; + (void) test_oap_response_id_tamper; (void) test_oap_rekey; (void) test_oap_rekey_badcache; diff --git a/src/irmd/reg/name.c b/src/irmd/reg/name.c index 61a328ec..a3621fc3 100644 --- a/src/irmd/reg/name.c +++ b/src/irmd/reg/name.c @@ -157,6 +157,8 @@ static struct prog_entry * __reg_name_get_prog(const struct reg_name * name, llist_for_each(p, &name->progs) { struct prog_entry * entry; entry = list_entry(p, struct prog_entry, next); + assert(entry->exec != NULL); + assert(entry->exec[0] != NULL); if (strcmp(entry->exec[0], prog) == 0) return entry; } diff --git a/src/irmd/reg/proc.c b/src/irmd/reg/proc.c index 8a7e24c9..94ac6b8f 100644 --- a/src/irmd/reg/proc.c +++ b/src/irmd/reg/proc.c @@ -119,6 +119,7 @@ static struct name_entry * __reg_proc_get_name(const struct reg_proc * proc, llist_for_each(p, &proc->names) { struct name_entry * entry; entry = list_entry(p, struct name_entry, next); + assert(entry->name != NULL); if (strcmp(entry->name, name) == 0) return entry; } @@ -140,7 +141,7 @@ int reg_proc_add_name(struct reg_proc * proc, } entry->name = strdup(name); - if (entry == NULL) { + if (entry->name == NULL) { log_err("Failed to strdup name."); goto fail_name; } diff --git a/src/irmd/reg/prog.c b/src/irmd/reg/prog.c index 2d7f9f8d..1e977c89 100644 --- a/src/irmd/reg/prog.c +++ b/src/irmd/reg/prog.c @@ -107,6 +107,7 @@ static struct name_entry * __reg_prog_get_name(const struct reg_prog * prog, llist_for_each(p, &prog->names) { struct name_entry * entry; entry = list_entry(p, struct name_entry, next); + assert(entry->name != NULL); if (strcmp(entry->name, name) == 0) return entry; } @@ -128,7 +129,7 @@ int reg_prog_add_name(struct reg_prog * prog, } entry->name = strdup(name); - if (entry == NULL) { + if (entry->name == NULL) { log_err("Failed to strdup name."); goto fail_name; } diff --git a/src/lib/dev.c b/src/lib/dev.c index 166aba5c..3fb8d831 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -98,14 +98,14 @@ struct flow { ssize_t part_idx; struct crypt_ctx * crypt; - int headsz; /* Selector */ - int tailsz; /* Tag + CRC */ + int headsz; /* Selector */ + int tailsz; /* Tag + CRC */ - struct timespec rk_grace; /* TX-promote deadline (0 = none) */ - struct timespec rk_attempt; /* Last re-key attempt (backoff) */ - bool rk_wm_inflight; /* Re-key trigger in flight */ - uint32_t rk_wm_ctr; /* Throttles the consult */ - bool rk_initiator; /* OAP initiator this re-key */ + struct timespec rk_grace; /* TX-promote deadline */ + struct timespec rk_attempt; /* Last re-key attempt */ + bool rk_wm_inflight; /* Re-key trigger in flight */ + uint32_t rk_wm_ctr; /* Throttles the consult */ + bool rk_initiator; /* OAP initiator this re-key */ struct timespec snd_act; struct timespec rcv_act; @@ -1739,8 +1739,7 @@ static ssize_t flow_write_frag(struct flow * flow, memcpy(ptr, src + off, clen); - ret = flow_tx_spb(flow, spb, flow_frag_role(i, n), - block, dl); + ret = flow_tx_spb(flow, spb, flow_frag_role(i, n), block, dl); if (ret < 0) { ssm_pool_remove(proc.pool, idx); if (off > 0) @@ -2887,19 +2886,13 @@ int ipcp_flow_get_qoscube(int fd, return 0; } +/* Not a snapshot: two atomic loads; caller keeps the fd live. */ size_t ipcp_flow_queued(int fd) { - size_t q; - - pthread_rwlock_rdlock(&proc.lock); - + assert(fd >= 0 && fd < PROC_MAX_FLOWS); assert(proc.flows[fd].info.id >= 0); - q = ssm_rbuff_queued(proc.flows[fd].tx_rb); - - pthread_rwlock_unlock(&proc.lock); - - return q; + return ssm_rbuff_queued(proc.flows[fd].tx_rb); } int local_flow_transfer(int src_fd, diff --git a/src/lib/frct.c b/src/lib/frct.c index c055433d..efd50b9a 100644 --- a/src/lib/frct.c +++ b/src/lib/frct.c @@ -193,6 +193,8 @@ struct frcti_stat { size_t rxm_due_unowned; /* bail: slot.rxm replaced */ size_t rxm_due_aged; /* bail: r->t0 + t_r < now */ size_t rxm_due_defer; /* bail: non-HoL, deferred to HoL */ + size_t rxm_fast_skip; /* SACK skips: slot has FAST_RXM */ + size_t rxm_fast_stuck; /* those skips with age > rto */ size_t rxm_arm_fail; /* rxm_arm: malloc failed */ size_t rxm_cancel; /* entries cancelled at teardown */ size_t rxm_tx_dead; /* RXM tx into terminal flow */ @@ -549,6 +551,8 @@ static int frct_rib_read(const char * path, " bail (unowned): %20zu\n" " bail (aged): %20zu\n" " bail (defer): %20zu\n" + " skip (fast-rxm set): %20zu\n" + " skip (stuck past rto): %20zu\n" "RXM-arm malloc failures: %20zu\n" "RXM cancels (teardown): %20zu\n" "RXM tx into dead flow: %20zu\n" @@ -597,6 +601,7 @@ static int frct_rib_read(const char * path, s.stat.rxm_due_count, s.stat.rxm_due_acked, s.stat.rxm_due_unowned, s.stat.rxm_due_aged, s.stat.rxm_due_defer, + s.stat.rxm_fast_skip, s.stat.rxm_fast_stuck, s.stat.rxm_arm_fail, s.stat.rxm_cancel, s.stat.rxm_tx_dead, s.stat.tx_drop, @@ -1857,6 +1862,7 @@ struct frcti * frcti_create(int fd, / SACK_BLOCK_SIZE; if (bb > SACK_MAX_BLOCKS) bb = SACK_MAX_BLOCKS; + frcti->sack_n_max = (uint16_t) bb; frcti->max_rcv_sdu = FRCT_MAX_SDU; @@ -1948,10 +1954,13 @@ void frcti_destroy(struct frcti * frcti) printf("[FRCT teardown] pid=%d fd=%d " "sdu_snd=%zu sdu_reasm=%zu sdu_sole=%zu " "frag_snd=%zu frag_rcv=%zu frag_drop=%zu " - "rxm_rto=%zu rxm_sack=%zu rxm_dup=%zu " + "rxm_rto=%zu rxm_sack=%zu rxm_rack=%zu rxm_dup=%zu " "rxm_due=%zu acked=%zu unowned=%zu aged=%zu defer=%zu " + "fast_skip=%zu fast_stuck=%zu " "cancel=%zu arm_fail=%zu inflight=%u " "nack_snd=%zu nack_rcv=%zu inact_drop=%zu " + "tlp_snd=%zu sack_snd=%zu sack_rcv=%zu ack_supp=%zu " + "out_rcv=%zu rqo_rcv=%zu dup_rcv=%zu rxm_dup_rcv=%zu " "drf_rebase=%zu rq_released=%zu\n", (int) getpid(), frcti->fd, frcti->stat.sdu_snd_frag, frcti->stat.sdu_reasm, @@ -1959,14 +1968,19 @@ void frcti_destroy(struct frcti * frcti) frcti->stat.frag_snd, frcti->stat.frag_rcv, frcti->stat.frag_drop, frcti->stat.rxm_rto, frcti->stat.rxm_sack, - frcti->stat.rxm_dupthresh, + frcti->stat.rxm_rack, frcti->stat.rxm_dupthresh, frcti->stat.rxm_due_count, frcti->stat.rxm_due_acked, frcti->stat.rxm_due_unowned, frcti->stat.rxm_due_aged, frcti->stat.rxm_due_defer, + frcti->stat.rxm_fast_skip, frcti->stat.rxm_fast_stuck, frcti->stat.rxm_cancel, frcti->stat.rxm_arm_fail, frcti->snd_cr.seqno - frcti->snd_cr.lwe, frcti->stat.nack_snd, frcti->stat.nack_rcv, frcti->stat.inact_drop, + frcti->stat.tlp_snd, frcti->stat.sack_snd, + frcti->stat.sack_rcv, frcti->stat.ack_supp_seqno, + frcti->stat.out_rcv, frcti->stat.rqo_rcv, + frcti->stat.dup_rcv, frcti->stat.rxm_dup_rcv, frcti->stat.drf_rebase, frcti->stat.rq_released); #endif @@ -2062,6 +2076,7 @@ int frcti_set_rcv_ring_sz(struct frcti * frcti, if (!frcti->stream) return -ENOTSUP; + if (!stream_ring_sz_ok(frcti, n)) return -EINVAL; @@ -2930,9 +2945,6 @@ static void tlp_due(void * arg) if (frcti->snd_cr.seqno == frcti->snd_cr.lwe) goto unlock; - if (!before(frcti->snd_cr.seqno, frcti->snd_cr.rwe)) - goto unlock; /* FC-blocked: RDV handles it. */ - /* RFC 8985 §7.3: one outstanding probe, MAX_TLP_PER_EP per ep. */ if (frcti->tlp_high_seq != 0) goto unlock; @@ -2965,10 +2977,10 @@ static void tlp_due(void * arg) memcpy(pkt_copy, rxm->pkt, rxm->len); pkt_len = rxm->len; frcti->snd_slots[hp].time = now_ns; - frcti->snd_slots[hp].flags |= SND_TLP | SND_FAST_RXM; + frcti->snd_slots[hp].flags |= SND_TLP; frcti->rtt_lwe = frcti->snd_cr.lwe + 1; - /* §7.3 outstanding-probe marker; ack_rcv/rxm_snd clear. */ - frcti->tlp_high_seq = frcti->snd_cr.seqno; + /* Probe is the HoL: any cum-ACK resolves the episode. */ + frcti->tlp_high_seq = frcti->snd_cr.lwe + 1; frcti->tlp_count++; STAT_BUMP(frcti, tlp_snd); } @@ -2998,8 +3010,10 @@ static int tlp_arm(struct frcti * frcti) /* §7.3: one outstanding probe, MAX_TLP_PER_EP per recovery ep. */ if (LOAD_RELAXED(&frcti->tlp_high_seq) != 0) return 0; + if (LOAD_RELAXED(&frcti->tlp_count) >= MAX_TLP_PER_EP) return 0; + if (__atomic_test_and_set(&frcti->tlp_pending, __ATOMIC_RELAXED)) return 0; @@ -3082,15 +3096,20 @@ static bool rtt_sample_eligible(struct frcti * frcti, { if (flags & FRCT_RXM) return false; + if (frcti->snd_slots[p].flags & (SND_RTX | SND_TLP)) return false; + if (LOAD_ACQUIRE(&frcti->snd_slots[p].rxm) == NULL) return false; + if (before(lwe, frcti->rtt_lwe)) return false; + /* Don't seed srtt from a cum-ACK; let probes seed. */ if (frcti->srtt == 0) return false; + return true; } @@ -3182,8 +3201,10 @@ static void frcti_ack_rcv(struct frcti * frcti, /* §7.3: cum-ACK past the probed seqno resolves the TLP. */ if (frcti->tlp_high_seq != 0 - && !before(ackno, frcti->tlp_high_seq)) + && !before(ackno, frcti->tlp_high_seq)) { frcti->tlp_high_seq = 0; + frcti->tlp_count = 0; + } /* §7.3: end the probe episode once inflight drains. */ if (ackno == frcti->snd_cr.seqno) @@ -3297,8 +3318,22 @@ static void sack_queue_rxm(struct frcti * frcti, if (rxm == NULL) continue; - if (frcti->snd_slots[kp].flags & SND_FAST_RXM) - continue; + /* + * A fast-retx still outstanding after its own RTO is + * presumed lost; clear the flag so RACK can repair it + * again instead of stranding it until the HoL timer. + */ + if (frcti->snd_slots[kp].flags & SND_FAST_RXM) { + if (!ts_aged_ns(now_ns, + frcti->snd_slots[kp].time, + LOAD_RELAXED(&frcti->rto))) { + STAT_BUMP(frcti, rxm_fast_skip); + continue; + } + + STAT_BUMP(frcti, rxm_fast_stuck); + frcti->snd_slots[kp].flags &= ~SND_FAST_RXM; + } if (RXM_AGED_OUT(rxm->t0, now_ns, frcti->t_r)) continue; @@ -3584,13 +3619,10 @@ static bool sack_check(struct frcti * frcti, n = dsack_consume(frcti, out->blocks); if (n == 1) out->dsack = true; + n += sack_blocks_build(frcti, out->blocks + n, frcti->sack_n_max - n); - if (!out->dsack - && rcv_cr->lwe == frcti->sack_lwe && n == frcti->sack_n) - return false; - out->n = n; out->ack = rcv_cr->lwe; out->rwe = frcti_advert_rwe(frcti); @@ -3646,6 +3678,7 @@ static void seqno_rotate(struct frcti * frcti, if (!ts_aged_ns(now_ns, snd_cr->act, snd_cr->inact)) return; + /* Idle-on-wire ≠ idle e2e: don't orphan in-flight rxm. */ if (snd_cr->seqno != snd_cr->lwe) return; @@ -3709,9 +3742,6 @@ static int frcti_snd(struct frcti * frcti, memset(pci, 0, FRCT_PCILEN); - if (frcti->stream) - spci = FRCT_SPCI(pci); - clock_gettime(PTHREAD_COND_CLOCK, &now); now_ns = TS_TO_UINT64(now); @@ -3727,6 +3757,8 @@ static int frcti_snd(struct frcti * frcti, STAT_BUMP(frcti, frag_snd); if (frcti->stream) { + spci = FRCT_SPCI(pci); + if (flags & FRCT_FIN) pci_flags |= FRCT_FIN; diff --git a/src/lib/pb/ipcp_config.proto b/src/lib/pb/ipcp_config.proto index eac4da37..a111b6d8 100644 --- a/src/lib/pb/ipcp_config.proto +++ b/src/lib/pb/ipcp_config.proto @@ -42,6 +42,7 @@ message dt_config_msg { required uint32 eid_size = 2; required uint32 max_ttl = 3; required routing_config_msg routing = 4; + optional uint32 max_rtt = 5; } message dir_dht_config_msg { diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c index a824d357..6beae000 100644 --- a/src/lib/protobuf.c +++ b/src/lib/protobuf.c @@ -362,6 +362,8 @@ dt_config_msg_t * dt_config_s_to_msg(const struct dt_config * s) msg->addr_size = s->addr_size; msg->eid_size = s->eid_size; msg->max_ttl = s->max_ttl; + msg->has_max_rtt = true; + msg->max_rtt = s->max_rtt; msg->routing = routing_config_s_to_msg(&s->routing); if (msg->routing == NULL) goto fail_routing; @@ -381,6 +383,7 @@ struct dt_config dt_config_msg_to_s(const dt_config_msg_t * msg) s.addr_size = msg->addr_size; s.eid_size = msg->eid_size; s.max_ttl = msg->max_ttl; + s.max_rtt = msg->has_max_rtt ? msg->max_rtt : 200; /* ms */ s.routing = routing_config_msg_to_s(msg->routing); return s; diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c index 7886b5c2..04978d82 100644 --- a/src/lib/ssm/rbuff.c +++ b/src/lib/ssm/rbuff.c @@ -74,7 +74,7 @@ struct ssm_rbuff { ssize_t * shm_base; /* start of shared memory */ size_t * head; /* start of ringbuffer */ size_t * tail; - size_t * flags; /* out-of-band flags (RB_*) */ + size_t * flags; /* out-of-band flags (RB_*) */ pthread_mutex_t * mtx; /* lock for cond vars only */ pthread_cond_t * add; /* signal when new data */ pthread_cond_t * del; /* signal when data removed */ diff --git a/src/lib/tests/crypt_test.c b/src/lib/tests/crypt_test.c index f00618d8..50b7268a 100644 --- a/src/lib/tests/crypt_test.c +++ b/src/lib/tests/crypt_test.c @@ -51,12 +51,20 @@ static int test_crypt_create_destroy(void) memset(key, 0, sizeof(key)); ctx = crypt_create_ctx(&sk); +#ifdef HAVE_OPENSSL if (ctx == NULL) { printf("Failed to initialize cryptography.\n"); goto fail; } crypt_destroy_ctx(ctx); +#else + if (ctx != NULL) { + printf("Created cipher context without a backend.\n"); + crypt_destroy_ctx(ctx); + goto fail; + } +#endif TEST_SUCCESS(); diff --git a/src/tools/irm/irm_ipcp_bootstrap.c b/src/tools/irm/irm_ipcp_bootstrap.c index de73b076..cc8bf8fa 100644 --- a/src/tools/irm/irm_ipcp_bootstrap.c +++ b/src/tools/irm/irm_ipcp_bootstrap.c @@ -105,7 +105,8 @@ static char * usage_str = \ " [Data Transfer Constants]\n" " [addr <address size> (default: %d)]\n" " [eid <eid size> (default: %d)]\n" - " [ttl <max time-to-live>, default: %d)]\n\n" + " [ttl <max time-to-live>, default: %d)]\n" + " [rtt <max layer RTT in ms> (default: %d)]\n\n" "if DIRECTORY_POLICY == " DHT_DIR "\n" " [dht_alpha <search factor> (default: %u)]\n" " [dht_k <replication factor> (default: %u)]\n" @@ -152,7 +153,7 @@ static void usage(void) /* unicast */ FLAT_RANDOM, DHT_DIR, SHA3_256, LINK_STATE, MB_ECN_CA, /* dt */ - DT(addr_size), DT(eid_size), DT(max_ttl), + DT(addr_size), DT(eid_size), DT(max_ttl), DT(max_rtt), /* dht */ DHT(alpha), DHT(k), DHT(t_expire), DHT(t_refresh), DHT(t_replicate), @@ -184,6 +185,7 @@ int do_bootstrap_ipcp(int argc, uint8_t addr_size = DT(addr_size); uint8_t eid_size = DT(eid_size); uint8_t max_ttl = DT(max_ttl); + uint16_t max_rtt = DT(max_rtt); struct routing_config routing = default_routing_config; enum pol_addr_auth addr_auth_type = UNI(addr_auth_type); enum pol_cong_avoid cong_avoid = UNI(cong_avoid); @@ -250,6 +252,8 @@ int do_bootstrap_ipcp(int argc, eid_size = atoi(*(argv + 1)); } else if (matches(*argv, "ttl") == 0) { max_ttl = atoi(*(argv + 1)); + } else if (matches(*argv, "rtt") == 0) { + max_rtt = atoi(*(argv + 1)); } else if (matches(*argv, "port") == 0) { port4 = atoi(*(argv + 1)); port6 = port4; @@ -422,6 +426,7 @@ int do_bootstrap_ipcp(int argc, conf.unicast.dt.addr_size = addr_size; conf.unicast.dt.eid_size = eid_size; conf.unicast.dt.max_ttl = max_ttl; + conf.unicast.dt.max_rtt = max_rtt; conf.unicast.dt.routing = routing; conf.unicast.addr_auth_type = addr_auth_type; conf.unicast.cong_avoid = cong_avoid; diff --git a/src/tools/ocbr/ocbr_server.c b/src/tools/ocbr/ocbr_server.c index c98b33e9..f9f7e3c8 100644 --- a/src/tools/ocbr/ocbr_server.c +++ b/src/tools/ocbr/ocbr_server.c @@ -50,7 +50,7 @@ #include <stdlib.h> #include <pthread.h> -#define THREADS_SIZE 10 +#define THREADS_SIZE 64 pthread_t listen_thread; pthread_t threads[THREADS_SIZE]; diff --git a/src/tools/oftp/oftp.c b/src/tools/oftp/oftp.c index 1ae99403..2504393a 100644 --- a/src/tools/oftp/oftp.c +++ b/src/tools/oftp/oftp.c @@ -65,9 +65,11 @@ static void apply_rto_min_env(int fd) env = getenv("OFTP_FRCT_RTO_MIN"); if (env == NULL) return; + v = strtol(env, NULL, 10); if (v <= 0) return; + if (fccntl(fd, FRCTSRTOMIN, (time_t) v) < 0) fprintf(stderr, "oftp: failed to set RTO_MIN=%ld ns\n", v); @@ -81,9 +83,11 @@ static void apply_stream_ring_sz_env(int fd) env = getenv("OFTP_FRCT_STREAM_RING_SZ"); if (env == NULL) return; + v = strtol(env, NULL, 10); if (v <= 0) return; + if (fccntl(fd, FRCTSRRINGSZ, (size_t) v) < 0) fprintf(stderr, "oftp: failed to set STREAM_RING_SZ=%ld\n", v); @@ -301,7 +305,7 @@ static int server_main(const char * outpath) fd = flow_accept(&qs, NULL); if (fd < 0) { fprintf(stderr, "flow_accept failed: %d\n", fd); - if (out != stdout) + if (outpath != NULL) fclose(out); return 1; } @@ -311,7 +315,7 @@ static int server_main(const char * outpath) "oftp: rejecting non-stream flow (service=%u)\n", qs.service); flow_dealloc(fd); - if (out != stdout) { + if (outpath != NULL) { fclose(out); unlink(outpath); } @@ -325,7 +329,7 @@ static int server_main(const char * outpath) flow_dealloc(fd); - if (out != stdout) { + if (outpath != NULL) { fclose(out); /* Drop the half-written file on abort/interrupt. */ if (rc != 0) @@ -358,7 +362,7 @@ static int client_main(const char * name, fd = flow_alloc(name, &qs, NULL); if (fd < 0) { fprintf(stderr, "flow_alloc failed: %d\n", fd); - if (in != stdin) + if (inpath != NULL) fclose(in); return 2; } @@ -370,7 +374,7 @@ static int client_main(const char * name, flow_dealloc(fd); - if (in != stdin) + if (inpath != NULL) fclose(in); return rc; |