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