diff options
Diffstat (limited to 'src/ipcpd/unicast')
| -rw-r--r-- | src/ipcpd/unicast/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca.c | 178 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca.h | 28 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/mb-ecn.c | 826 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/mb-ecn.h | 17 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/nop.c | 35 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/nop.h | 17 | ||||
| -rw-r--r-- | src/ipcpd/unicast/ca/ops.h | 20 | ||||
| -rw-r--r-- | src/ipcpd/unicast/cap.c | 291 | ||||
| -rw-r--r-- | src/ipcpd/unicast/cap.h | 54 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dir/dht.c | 2 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dt.c | 81 | ||||
| -rw-r--r-- | src/ipcpd/unicast/dt.h | 3 | ||||
| -rw-r--r-- | src/ipcpd/unicast/fa.c | 72 | ||||
| -rw-r--r-- | src/ipcpd/unicast/fa.h | 1 | ||||
| -rw-r--r-- | src/ipcpd/unicast/psched.c | 199 | ||||
| -rw-r--r-- | src/ipcpd/unicast/psched.h | 6 |
17 files changed, 1532 insertions, 299 deletions
diff --git a/src/ipcpd/unicast/CMakeLists.txt b/src/ipcpd/unicast/CMakeLists.txt index d3388112..d51546a3 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 diff --git a/src/ipcpd/unicast/ca.c b/src/ipcpd/unicast/ca.c index a1751672..d0ee2f73 100644 --- a/src/ipcpd/unicast/ca.c +++ b/src/ipcpd/unicast/ca.c @@ -22,17 +22,47 @@ #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) { +#ifndef IPCP_CA_PER_FLOW + size_t i; +#endif switch(pol) { case CA_NONE: log_dbg("Disabling congestion control."); @@ -46,63 +76,167 @@ int ca_init(enum pol_cong_avoid pol) 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; -void ca_ctx_update_ece(void * ctx, - uint16_t ece) -{ - return ca.ops->ctx_update_ece(ctx, ece); + return ca.ops->ctx_update_rcv(ctx->pol, len, ecn, cap, ece, fcap); } -void ca_wnd_wait(ca_wnd_t wnd) +void ca_ctx_update_ece(void * _ctx, + uint16_t ece, + uint8_t cap) { - return ca.ops->wnd_wait(wnd); + struct ca_ctx * ctx = _ctx; + + return ca.ops->ctx_update_ece(ctx->pol, ece, cap); } -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..d73d35f5 100644 --- a/src/ipcpd/unicast/ca.h +++ b/src/ipcpd/unicast/ca.h @@ -29,38 +29,40 @@ #include <stdbool.h> #include <sys/types.h> -typedef union { - time_t wait; -} ca_wnd_t; - int ca_init(enum pol_cong_avoid ca); 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); -void ca_wnd_wait(ca_wnd_t wnd); - -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..e59aac88 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,31 +39,125 @@ #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) +/* + * The sender paces with a token bucket at a rate driven by Δt-scaled + * AIMD: each step changes the rate proportional to elapsed wall-clock + * time, so the per-second dynamics do not depend on how often packets + * arrive. The receiver's averaging window and the sender's feedback + * staleness both stretch with the flow's byte rate, so a slow flow + * is measured and controlled like a fast one; CA_RATE_MIN only + * bounds those horizons (window <= CA_TW_ABSMAX, TTL ~8 s). There + * is no per-flow timer; the control runs on packet sends. + * + * 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 */ +/* Must stay >= 1 ms: the MD term scales by dtc / MILLION (truncates). */ +#define CA_DT_CTRL (BILLION / 1000) +#define CA_DT_CAP (BILLION / 20) /* idle-resume Δt clamp 50ms */ +/* Floor of the rate-relative feedback staleness (ctx->ece_ttl). */ +#define CA_ECE_TTL ((1 << CA_TW_GAP_SHFT) * CA_TW_INIT) +#define CA_SS_TC (BILLION / 50) +#define CA_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) */ + uint8_t tx_cap; /* path capacity code fed back to us */ + bool tx_cav; /* past slow start */ + 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 r_bkt; /* rate snapshot at last washout bucket */ + uint64_t wash_acc; /* washout bucket time accumulator (ns) */ + uint64_t inv_rate; /* fixed-point 1/rate for pacing */ + uint64_t vt; /* virtual service clock (bytes) */ + uint64_t lead; /* pacer lead of last send (bytes) */ + uint64_t last_ts; /* last clock advance (ns) */ + uint64_t last_ctrl; /* last rate update (ns) */ + uint64_t last_fb; /* last feedback applied (ns) */ + uint64_t last_loc; /* last local mark seen (ns) */ + uint64_t last_cap; /* last capacity applied (ns) */ + + uint64_t n_ctrl; /* control steps taken */ + uint64_t t_ctrl; /* wall time covered by steps (ns) */ + uint64_t t_bank; /* increase time banked in steps (ns) */ + uint64_t n_fb; /* feedback updates received */ + uint64_t n_ttl; /* feedback aged out (TTL) */ + uint64_t n_cap; /* capacity updates applied */ + uint64_t ss_peak; /* peak rate in slow start (bytes/s) */ + + uint64_t snd_byt; /* bytes offered this window (capped) */ + uint64_t snd_win; /* utilisation window start (ns) */ + uint64_t snd_r0; /* rate at window start */ + 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 */ }; struct ca_ops mb_ecn_ca_ops = { @@ -71,14 +166,34 @@ struct ca_ops mb_ecn_ca_ops = { .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, .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; +} + void * mb_ecn_ctx_create(void) { struct timespec now; + uint64_t t; struct mb_ecn_ctx * ctx; ctx = malloc(sizeof(*ctx)); @@ -89,10 +204,27 @@ 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->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_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 +234,519 @@ 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 / CA_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; + + ctx->rate += ctx->ai_rate * dta / BILLION; + ctx->rate += ctx->rate * dta / CA_PROBE_TC; +} + +/* Multiplicative decrease: proportional cut plus a one-sided lead. */ +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->tx_ecp = 0; + return; + } + + mark = MIN(m, CA_ECE_REF); + rise = 0; + if (m > ctx->tx_ecp) + rise = MIN(m - ctx->tx_ecp, CA_ECE_REF); + + cut = ctx->rate * rise / (CA_ECE_REF * CA_MD_KD_DIV); + + /* Honest elapsed ms, so a starved sender still cuts. */ + dtm = dtc / 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; + + offered = ctx->snd_byt * BILLION / elapsed; + + ctx->backlogged = offered * CA_USE_DEN >= ctx->snd_r0 * CA_USE_NUM; -ca_wnd_t mb_ecn_ctx_update_snd(void * _ctx, - size_t len) + 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. */ +static void mb_ecn_age(struct mb_ecn_ctx * ctx, + uint64_t t) +{ + 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->snd_win = t; + ctx->snd_r0 = ctx->rate; + } + + dt = t - ctx->last_ts; + ctx->last_ts = t; + + 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; + + if (dt > (uint64_t) CA_DT_CAP) + ctx->backlogged = false; + + if (t - ctx->snd_win >= (uint64_t) CA_SND_WIN) + mb_ecn_win(ctx, t); + + /* 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; - nanosleep(&s, NULL); + dt = t - ctx->rx_ts; + ctx->rx_ts = t; + + 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; + ctx->last_fb = 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 mb_ecn_ctx* ctx = _ctx; + struct timespec now; + struct mb_ecn_ctx * ctx = _ctx; - ctx->tx_ece = ece; - ctx->tx_ctr = 0; - ctx->tx_cav = true; + clock_gettime(PTHREAD_COND_CLOCK, &now); + + mb_ecn_ece(ctx, ece, cap, TS_TO_UINT64(now)); } -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 +755,64 @@ 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) 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->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", "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); return strlen(buf); } diff --git a/src/ipcpd/unicast/ca/mb-ecn.h b/src/ipcpd/unicast/ca/mb-ecn.h index 1be27764..7bf0b29f 100644 --- a/src/ipcpd/unicast/ca/mb-ecn.h +++ b/src/ipcpd/unicast/ca/mb-ecn.h @@ -29,20 +29,23 @@ 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); - -void mb_ecn_wnd_wait(ca_wnd_t wnd); + uint16_t ece, + uint8_t cap); -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..a86c1b3b 100644 --- a/src/ipcpd/unicast/ca/ops.h +++ b/src/ipcpd/unicast/ca/ops.h @@ -30,24 +30,30 @@ 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); - - void (* wnd_wait)(ca_wnd_t wnd); + uint16_t ece, + uint8_t cap); - 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/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 b17218c4..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; } diff --git a/src/ipcpd/unicast/dt.c b/src/ipcpd/unicast/dt.c index 0acae05e..6afaa1f9 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); } @@ -301,12 +308,12 @@ static int dt_rib_readdir(char *** buf) if (*buf == NULL) goto fail_entries; - for (i = 0; i < PROC_MAX_FLOWS; ++i) { + for (i = 0; i < PROC_MAX_FLOWS && idx < (int) dt.n_flows; ++i) { pthread_mutex_lock(&dt.stat[i].lock); if (dt.stat[i].stamp == 0) { pthread_mutex_unlock(&dt.stat[i].lock); - break; + continue; /* n-1 flows start at PROC_RES_FDS */ } pthread_mutex_unlock(&dt.stat[i].lock); @@ -328,6 +335,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); @@ -413,6 +421,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 +438,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 +467,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 +477,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 +495,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 +523,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,9 +583,15 @@ 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 (cap_init() < 0) { + log_err("Failed to init capacity estimator."); + goto fail_cap; + } + if (connmgr_comp_init(COMPID_DT, &info)) { log_err("Failed to register with connmgr."); goto fail_connmgr_comp_init; @@ -642,6 +671,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; } @@ -671,6 +702,8 @@ void dt_fini(void) routing_fini(); connmgr_comp_fini(COMPID_DT); + + cap_fini(); } int dt_start(void) @@ -773,13 +806,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 +849,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); + + if (marks) + dt_pci.cap = cap_get(fd); - (void) ca_calc_ecn(fd, &dt_pci.ecn, qc, len); + if (ecn != NULL) + *ecn = dt_pci.ecn; dt_pci_ser(head, &dt_pci); @@ -828,14 +874,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..ac2ecaea 100644 --- a/src/ipcpd/unicast/fa.c +++ b/src/ipcpd/unicast/fa.c @@ -31,6 +31,7 @@ #define FA "flow-allocator" #define OUROBOROS_PREFIX FA +#include <ouroboros/atomics.h> #include <ouroboros/endian.h> #include <ouroboros/logs.h> #include <ouroboros/fqueue.h> @@ -38,6 +39,7 @@ #include <ouroboros/dev.h> #include <ouroboros/ipcp-dev.h> #include <ouroboros/np1_flow.h> +#include <ouroboros/qoscube.h> #include <ouroboros/rib.h> #include <ouroboros/random.h> #include <ouroboros/pthread.h> @@ -81,6 +83,7 @@ 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; @@ -109,6 +112,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 { @@ -245,6 +250,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); @@ -317,18 +323,21 @@ 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) +static time_t packet_handler(int fd, + qoscube_t qc, + struct ssm_pk_buff * spb) { struct fa_flow * flow; uint64_t r_addr; uint64_t r_eid; - ca_wnd_t wnd; + time_t wait; size_t len; + uint8_t ecn; flow = &fa.flows[fd]; + ecn = 0; + pthread_rwlock_wrlock(&fa.flows_lock); len = ssm_pk_buff_len(spb); @@ -337,16 +346,16 @@ 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); r_addr = flow->r_addr; r_eid = flow->r_eid; pthread_rwlock_unlock(&fa.flows_lock); - ca_wnd_wait(wnd); - - 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 +364,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 +383,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 +397,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)); @@ -503,6 +515,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 +599,7 @@ 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); @@ -834,7 +853,7 @@ int fa_alloc(int fd, 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 +867,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 +916,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 +970,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 +998,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 +1023,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 +1032,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 +1043,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 +1071,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 +1087,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/psched.c b/src/ipcpd/unicast/psched.c index 21e23617..2b535d67 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,139 @@ 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; +} + +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; + + for (i = 0; i < d->n; ) { + fd = d->active[i]; + + if (d->deadline[fd] > now) { + if (dmin == 0 || d->deadline[fd] < dmin) + dmin = d->deadline[fd]; + ++i; + continue; + } + + if (sched->read(fd, &spb) < 0) { + dsched_untrack(d, fd); + continue; + } + + wait = sched->callback(fd, qc, spb); + if (wait == 0) + continue; + + d->deadline[fd] = now + (uint64_t) wait; + if (dmin == 0 || d->deadline[fd] < dmin) + dmin = d->deadline[fd]; + ++i; + } + + 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 +202,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 +268,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 +286,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 +328,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 +352,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 +366,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 +377,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); |