diff options
| -rw-r--r-- | cmake/config/lib/ssm.cmake | 44 | ||||
| -rw-r--r-- | doc/man/fccntl.3 | 7 | ||||
| -rw-r--r-- | include/ouroboros/atomics.h | 2 | ||||
| -rw-r--r-- | include/ouroboros/flow.h | 2 | ||||
| -rw-r--r-- | include/ouroboros/ipcp-dev.h | 116 | ||||
| -rw-r--r-- | include/ouroboros/ssm_rbuff.h | 10 | ||||
| -rw-r--r-- | src/ipcpd/unicast/fa.c | 4 | ||||
| -rw-r--r-- | src/ipcpd/unicast/fa.h | 2 | ||||
| -rw-r--r-- | src/ipcpd/unicast/main.c | 2 | ||||
| -rw-r--r-- | src/irmd/reg/flow.c | 5 | ||||
| -rw-r--r-- | src/irmd/reg/reg.c | 12 | ||||
| -rw-r--r-- | src/lib/dev.c | 219 | ||||
| -rw-r--r-- | src/lib/frct.c | 489 | ||||
| -rw-r--r-- | src/lib/pb/model.proto | 1 | ||||
| -rw-r--r-- | src/lib/poa/poa.c | 4 | ||||
| -rw-r--r-- | src/lib/protobuf.c | 2 | ||||
| -rw-r--r-- | src/lib/ssm/rbuff.c | 430 | ||||
| -rw-r--r-- | src/lib/ssm/ssm.h.in | 2 | ||||
| -rw-r--r-- | src/lib/ssm/tests/rbuff_test.c | 98 |
19 files changed, 1050 insertions, 401 deletions
diff --git a/cmake/config/lib/ssm.cmake b/cmake/config/lib/ssm.cmake index a9ac35c9..aa9c4f75 100644 --- a/cmake/config/lib/ssm.cmake +++ b/cmake/config/lib/ssm.cmake @@ -22,6 +22,10 @@ set(SSM_RBUFF_SIZE 1024 CACHE STRING "Number of slots in a flow's rbuff ring; must be a power of 2") set(SSM_RBUFF_TXQ_DELAY 10 CACHE STRING "Queueing delay a flow's tx ring may hold (ms); 0 is unlimited") +set(SSM_RBUFF_TXQ_PRIO_MUL 2 CACHE STRING + "Retransmissions get this multiple of the tx occupancy limit") +set(SSM_RBUFF_TXQ_PRIO_DIV 16 CACHE STRING + "Ring fraction (1/N) new data leaves to retransmissions") set(SSM_RBUFF_PREFIX "/${SHM_PREFIX}.rbuff." CACHE INTERNAL "Prefix for rbuff POSIX shared memory filenames") set(SSM_FLOW_SET_PREFIX "/${SHM_PREFIX}.set." CACHE INTERNAL @@ -148,6 +152,46 @@ message(STATUS " Blocks: ${SSM_PUP_256_BLOCKS}, ${SSM_PUP_512_BLOCKS}, " "${SSM_PUP_16K_BLOCKS}, ${SSM_PUP_64K_BLOCKS}, ${SSM_PUP_256K_BLOCKS}, " "${SSM_PUP_1M_BLOCKS}") +math(EXPR SSM_RBUFF_TXQ_RESERVE + "${SSM_RBUFF_SIZE} / ${SSM_RBUFF_TXQ_PRIO_DIV}") + +set(SSM_RBUFF_TXQ_RESERVE ${SSM_RBUFF_TXQ_RESERVE} CACHE INTERNAL + "Top-of-ring slots new data may not use") + +math(EXPR SSM_RBUFF_TXQ_DATA_MAX + "${SSM_RBUFF_SIZE} - 1 - ${SSM_RBUFF_TXQ_RESERVE}") + +# The limiter asserts its target fits SSM_RBUFF_TXQ_MAX_DELAY (1 s). +if(SSM_RBUFF_TXQ_DELAY LESS 0 OR SSM_RBUFF_TXQ_DELAY GREATER 1000) + message(FATAL_ERROR + "SSM_RBUFF_TXQ_DELAY (${SSM_RBUFF_TXQ_DELAY}) must be in " + "[0, 1000] ms; a longer target trips the limiter's own bound.") +endif() + +# What the reserve leaves new data must still clear the 4-slot floor. +if(SSM_RBUFF_TXQ_DATA_MAX LESS 4) + message(FATAL_ERROR + "SSM_RBUFF_TXQ_PRIO_DIV (${SSM_RBUFF_TXQ_PRIO_DIV}) leaves new " + "data ${SSM_RBUFF_TXQ_DATA_MAX} of ${SSM_RBUFF_SIZE} slots, " + "below the 4-slot floor: every write would be refused.") +endif() + +# Both bounds keep the retransmission headroom above zero: the divisor +# reserves it at saturation, the multiplier below it. +if(SSM_RBUFF_TXQ_PRIO_DIV LESS 2 OR SSM_RBUFF_TXQ_RESERVE LESS 1) + message(FATAL_ERROR + "SSM_RBUFF_TXQ_PRIO_DIV (${SSM_RBUFF_TXQ_PRIO_DIV}) must be in " + "[2, SSM_RBUFF_SIZE (${SSM_RBUFF_SIZE})]: it reserves " + "${SSM_RBUFF_TXQ_RESERVE} of ${SSM_RBUFF_SIZE} slots for " + "retransmissions.") +endif() + +if(SSM_RBUFF_TXQ_PRIO_MUL LESS 2) + message(FATAL_ERROR + "SSM_RBUFF_TXQ_PRIO_MUL (${SSM_RBUFF_TXQ_PRIO_MUL}) must be >= 2: " + "at 1 a retransmission gets the ceiling new data already has.") +endif() + # FRCT reorder queue must fit in every enabled size class. If RQ_SIZE >= any backing pool, the # receiver advertises a window the pool cannot back; np1_flow_write fails under load and a single # dropped fragment wedges the flow. Auto-zeroed classes are skipped. diff --git a/doc/man/fccntl.3 b/doc/man/fccntl.3 index 61022837..3dcfc2d9 100644 --- a/doc/man/fccntl.3 +++ b/doc/man/fccntl.3 @@ -80,7 +80,8 @@ buffer. Takes a \fBsize_t \fIqlen\fR * as third argument. occupancy limiter. Takes a \fBstruct timespec * \fItimeo\fR as third argument. A zero delay leaves the ring unlimited, bounded only by its slot count. The build-time default is applied when the flow is -created. +created. The delay must be a normalised, non-negative timespec no +longer than the maximum the limiter accepts. \fBFLOWGTXQDLY\fR - retrieve the current target queueing delay for the transmit occupancy limiter. Takes a \fBstruct timespec * \fItimeo\fR @@ -111,7 +112,9 @@ On success, \fBfccntl\fR() returns 0. \fBfccntl\fR() can return .B -EINVAL -An invalid argument was passed. +An invalid argument was passed. This is also returned when the delay +passed to FLOWSTXQDLY is negative, is not normalised, or exceeds the +maximum the transmit occupancy limiter accepts. .B -EPERM Operation not permitted. This is returned when requesting the value of diff --git a/include/ouroboros/atomics.h b/include/ouroboros/atomics.h index 8e667522..9783fc78 100644 --- a/include/ouroboros/atomics.h +++ b/include/ouroboros/atomics.h @@ -35,5 +35,7 @@ #define FETCH_SUB_RELAXED(p, v) (__atomic_fetch_sub(p, v, __ATOMIC_RELAXED)) #define FETCH_ADD(p, v) (__atomic_fetch_add(p, v, __ATOMIC_SEQ_CST)) #define FETCH_SUB(p, v) (__atomic_fetch_sub(p, v, __ATOMIC_SEQ_CST)) +#define FETCH_OR(p, v) (__atomic_fetch_or(p, v, __ATOMIC_SEQ_CST)) +#define FETCH_AND(p, v) (__atomic_fetch_and(p, v, __ATOMIC_SEQ_CST)) #endif /* OUROBOROS_LIB_ATOMICS_H */ diff --git a/include/ouroboros/flow.h b/include/ouroboros/flow.h index 8b096410..3d3c74ec 100644 --- a/include/ouroboros/flow.h +++ b/include/ouroboros/flow.h @@ -53,6 +53,8 @@ struct flow_info { uint32_t mtu; /* n-1 layer MTU in bytes, 0 = unknown */ + uint32_t max_rtt; /* declared max path RTT (ms), 0 = unknown */ + struct qos_spec qs; enum flow_state state; diff --git a/include/ouroboros/ipcp-dev.h b/include/ouroboros/ipcp-dev.h index ab99d950..c0899c2b 100644 --- a/include/ouroboros/ipcp-dev.h +++ b/include/ouroboros/ipcp-dev.h @@ -33,109 +33,111 @@ #include <stdint.h> #include <time.h> -int ipcp_create_r(const struct ipcp_info * info); +int ipcp_create_r(const struct ipcp_info * info); -int ipcp_flow_req_arr(const buffer_t * dst, - qosspec_t qs, - time_t mpl, - uint32_t mtu, - const buffer_t * data); +void ipcp_flow_set_max_rtt(uint32_t max_rtt); -int ipcp_flow_update_arr(int flow_id, - const buffer_t * data); +int ipcp_flow_req_arr(const buffer_t * dst, + qosspec_t qs, + time_t mpl, + uint32_t mtu, + const buffer_t * data); -int ipcp_flow_alloc_reply(int fd, - int response, - time_t mpl, - uint32_t mtu, +int ipcp_flow_update_arr(int flow_id, const buffer_t * data); -int ipcp_flow_read(int fd, - struct ssm_pk_buff ** spb); +int ipcp_flow_alloc_reply(int fd, + int response, + time_t mpl, + uint32_t mtu, + const buffer_t * data); + +int ipcp_flow_read(int fd, + struct ssm_pk_buff ** spb); -int ipcp_flow_write(int fd, - struct ssm_pk_buff * spb); +int ipcp_flow_write(int fd, + struct ssm_pk_buff * spb); -int np1_flow_read(int fd, - struct ssm_pk_buff ** spb, - struct ssm_pool * pool); +int np1_flow_read(int fd, + struct ssm_pk_buff ** spb, + struct ssm_pool * pool); -int np1_flow_write(int fd, - struct ssm_pk_buff * spb, - struct ssm_pool * pool); +int np1_flow_write(int fd, + struct ssm_pk_buff * spb, + struct ssm_pool * pool); -int ipcp_flow_dealloc(int fd); +int ipcp_flow_dealloc(int fd); -int ipcp_flow_fini(int fd); +int ipcp_flow_fini(int fd); -int ipcp_flow_get_qoscube(int fd, - qoscube_t * cube); +int ipcp_flow_get_qoscube(int fd, + qoscube_t * cube); /* Bytes queued in the transmit path of the flow. */ -size_t ipcp_flow_queued(int fd); +size_t ipcp_flow_queued(int fd); /* Mean size of a packet written to the flow, bytes. */ -size_t ipcp_flow_mean_len(int fd); +size_t ipcp_flow_mean_len(int fd); /* * Capacity estimator of the transmit queue behind the flow. * Flows attached to one point of attachment share a queue and * share the estimator. Armed by the IPCP; unarmed flows read 0. */ -int ipcp_flow_cap_arm(int fd); +int ipcp_flow_cap_arm(int fd); -void ipcp_flow_cap_update(int fd, - size_t qlen, - size_t len); +void ipcp_flow_cap_update(int fd, + size_t qlen, + size_t len); /* Estimated capacity of the transmit queue (bytes/s), 0 = unknown. */ uint64_t ipcp_flow_cap(int fd); -int ipcp_spb_reserve(struct ssm_pk_buff ** spb, - size_t len); +int ipcp_spb_reserve(struct ssm_pk_buff ** spb, + size_t len); -void ipcp_spb_release(struct ssm_pk_buff * spb); +void ipcp_spb_release(struct ssm_pk_buff * spb); /* PoA lifecycle; an IPCP owns its PoAs, applications have none. */ -int poa_init(const char * name); +int poa_init(const char * name); -int poa_start(void); +int poa_start(void); -void poa_stop(void); +void poa_stop(void); -void poa_fini(void); +void poa_fini(void); /* Also answer name queries for the layer; set at bootstrap/enroll. */ -int poa_set_layer(const char * layer); +int poa_set_layer(const char * layer); /* Attach or release a PoA; the type in the spec picks the transport. */ -int poa_attach(const struct poa_spec * poa); +int poa_attach(const struct poa_spec * poa); -int poa_detach(const struct poa_spec * poa); +int poa_detach(const struct poa_spec * poa); /* The PoAs this process has attached. */ -ssize_t poa_list(struct poa_spec * specs, - size_t max); +ssize_t poa_list(struct poa_spec * specs, + size_t max); /* Allocate a flow to dst over the PoA that carries addr. */ -int poa_flow_alloc(const char * dst, - const struct poa_addr * addr, - qosspec_t * qs, - const struct timespec * timeo); +int poa_flow_alloc(const char * dst, + const struct poa_addr * addr, + qosspec_t * qs, + const struct timespec * timeo); /* Resolve dst on the attached PoAs; no flow is created. */ -int poa_query(const char * dst, - const struct timespec * timeo, - struct poa_addr * addr); +int poa_query(const char * dst, + const struct timespec * timeo, + struct poa_addr * addr); /* PoA counterparts of the ipcp_flow_* operations */ -int poa_flow_alloc_resp(int flow_id, - int response, - const buffer_t * data); +int poa_flow_alloc_resp(int flow_id, + int response, + const buffer_t * data); -int poa_flow_update(int flow_id, - const buffer_t * data); +int poa_flow_update(int flow_id, + const buffer_t * data); -int poa_flow_dealloc(int flow_id); +int poa_flow_dealloc(int flow_id); #endif /* OUROBOROS_LIB_IPCP_DEV_H */ diff --git a/include/ouroboros/ssm_rbuff.h b/include/ouroboros/ssm_rbuff.h index 9ff880e9..297aad9f 100644 --- a/include/ouroboros/ssm_rbuff.h +++ b/include/ouroboros/ssm_rbuff.h @@ -35,6 +35,8 @@ #define RB_FLOWPEER 0010 #define RB_REKEY 0020 /* re-key seed parked (out-of-band signal) */ +#define SSM_RBUFF_TXQ_MAX_DELAY 1000000000LL + struct ssm_rbuff; struct ssm_rbuff * ssm_rbuff_create(pid_t pid, @@ -47,11 +49,11 @@ struct ssm_rbuff * ssm_rbuff_open(pid_t pid, void ssm_rbuff_close(struct ssm_rbuff * rb); -void ssm_rbuff_set_bits(struct ssm_rbuff * rb, - uint32_t bits); +void ssm_rbuff_set_flags(struct ssm_rbuff * rb, + uint32_t flags); -void ssm_rbuff_clr_bits(struct ssm_rbuff * rb, - uint32_t bits); +void ssm_rbuff_clr_flags(struct ssm_rbuff * rb, + uint32_t flags); uint32_t ssm_rbuff_get_flags(struct ssm_rbuff * rb); diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c index 744eb316..1c939fab 100644 --- a/src/ipcpd/unicast/fa.c +++ b/src/ipcpd/unicast/fa.c @@ -891,11 +891,13 @@ static void * fa_handle_packet(void * o) return (void *) 0; } -int fa_init(void) +int fa_init(uint16_t max_rtt) { pthread_condattr_t cattr; size_t i; + ipcp_flow_set_max_rtt(max_rtt); + if (pthread_rwlock_init(&fa.flows_lock, NULL) != 0) goto fail_rwlock; diff --git a/src/ipcpd/unicast/fa.h b/src/ipcpd/unicast/fa.h index 1d0012b0..504d67d3 100644 --- a/src/ipcpd/unicast/fa.h +++ b/src/ipcpd/unicast/fa.h @@ -26,7 +26,7 @@ #include <ouroboros/qos.h> #include <ouroboros/utils.h> -int fa_init(void); +int fa_init(uint16_t max_rtt); void fa_fini(void); diff --git a/src/ipcpd/unicast/main.c b/src/ipcpd/unicast/main.c index 3ee53401..320ce165 100644 --- a/src/ipcpd/unicast/main.c +++ b/src/ipcpd/unicast/main.c @@ -85,7 +85,7 @@ static int initialize_components(struct ipcp_config * conf) goto fail_dir; } - if (fa_init()) { + if (fa_init(conf->unicast.dt.max_rtt)) { log_err("Failed to initialize flow allocator component."); goto fail_fa; } diff --git a/src/irmd/reg/flow.c b/src/irmd/reg/flow.c index 63c9199b..61e2a14b 100644 --- a/src/irmd/reg/flow.c +++ b/src/irmd/reg/flow.c @@ -172,8 +172,9 @@ int reg_flow_update(struct reg_flow * flow, assert(flow->info.n_pid != 0); assert(info->mpl != 0); - flow->info.mpl = info->mpl; - flow->info.mtu = info->mtu; + flow->info.mpl = info->mpl; + flow->info.mtu = info->mtu; + flow->info.max_rtt = info->max_rtt; if (flow->info.state == FLOW_ALLOC_PENDING) break; diff --git a/src/irmd/reg/reg.c b/src/irmd/reg/reg.c index a302fa15..e19566d4 100644 --- a/src/irmd/reg/reg.c +++ b/src/irmd/reg/reg.c @@ -2425,7 +2425,7 @@ int reg_flow_store_pending(int flow_id, flow->rk.has_pending = true; /* Doorbell raised only after the seed is parked. */ if (flow->n_rb != NULL) - ssm_rbuff_set_bits(flow->n_rb, RB_REKEY); + ssm_rbuff_set_flags(flow->n_rb, RB_REKEY); ret = 0; } else { ret = -ENOMEM; @@ -2484,10 +2484,10 @@ int reg_flow_store_pending_direct(int flow_id, flow->rk.pulled |= RK_N_1_PID; if (flow->n_rb != NULL && !(flow->rk.pulled & RK_N_PID)) - ssm_rbuff_set_bits(flow->n_rb, RB_REKEY); + ssm_rbuff_set_flags(flow->n_rb, RB_REKEY); if (flow->n_1_rb != NULL && !(flow->rk.pulled & RK_N_1_PID)) - ssm_rbuff_set_bits(flow->n_1_rb, RB_REKEY); + ssm_rbuff_set_flags(flow->n_1_rb, RB_REKEY); ret = 0; out: @@ -2526,11 +2526,11 @@ static void __take_pending_direct(struct reg_flow * flow, if (allocator) { flow->rk.pulled |= RK_N_1_PID; if (flow->n_1_rb != NULL) - ssm_rbuff_clr_bits(flow->n_1_rb, RB_REKEY); + ssm_rbuff_clr_flags(flow->n_1_rb, RB_REKEY); } else { flow->rk.pulled |= RK_N_PID; if (flow->n_rb != NULL) - ssm_rbuff_clr_bits(flow->n_rb, RB_REKEY); + ssm_rbuff_clr_flags(flow->n_rb, RB_REKEY); } if ((flow->rk.pulled & RK_PID_MASK) != RK_PID_MASK) @@ -2575,7 +2575,7 @@ int reg_flow_take_pending(int flow_id, flow->rk.has_pending = false; crypt_secure_clear(flow->rk.pending_seed, SYMMKEYSZ); if (flow->n_rb != NULL) - ssm_rbuff_clr_bits(flow->n_rb, RB_REKEY); + ssm_rbuff_clr_flags(flow->n_rb, RB_REKEY); ret = 0; out: diff --git a/src/lib/dev.c b/src/lib/dev.c index bce64c37..3b0363da 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -156,6 +156,8 @@ struct { struct flow * flows; struct fmap * id_to_fd; + uint32_t max_rtt; /* IPCPs: declared layer RTT (ms) */ + pthread_mutex_t mtx; pthread_cond_t cond; @@ -284,7 +286,7 @@ static int spb_encrypt(struct flow * flow, in.len = ssm_pk_buff_len(spb); if (crypt_encrypt(flow->crypt, in, &out) < 0) - goto fail_encrypt; + return -ECRYPT; head = ssm_pk_buff_push(spb, flow->headsz); if (head == NULL) @@ -301,7 +303,7 @@ static int spb_encrypt(struct flow * flow, return 0; fail_alloc: freebuf(out); - fail_encrypt: + return -ECRYPT; } @@ -703,10 +705,10 @@ static void flow_quiesce(int fd) struct ssm_rbuff * tx_rb = proc.flows[fd].tx_rb; if (rx_rb != NULL) - ssm_rbuff_set_bits(rx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(rx_rb, RB_FLOWDOWN); if (tx_rb != NULL) - ssm_rbuff_set_bits(tx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(tx_rb, RB_FLOWDOWN); } static void do_flow_fini(int fd) @@ -778,11 +780,15 @@ static int flow_init(struct flow_info * info, time_t rtt_hint, struct poa_flow * pf) { - struct timespec now; - struct timespec txq; - struct flow * flow; - int fd; - int err = -ENOMEM; + struct timespec now; + struct timespec txq; + struct flow * flow; + struct ssm_rbuff * tx_rb = NULL; + int fd; + int err = -ENOMEM; + + if (info->id < 0 || info->id >= SYS_MAX_FLOWS) + return -EBADF; clock_gettime(PTHREAD_COND_CLOCK, &now); @@ -807,10 +813,7 @@ 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); + tx_rb = flow->tx_rb; flow->set = ssm_flow_set_open(info->n_1_pid); if (flow->set == NULL) @@ -842,7 +845,7 @@ static int flow_init(struct flow_info * info, uint32_t frct_mtu = flow_user_mtu(flow, info->mtu); flow->frcti = frcti_create(fd, DELT_A, DELT_R, - info->mpl, rtt_hint, + info->mpl, rtt_hint, info->max_rtt, info->qs, frct_mtu); if (flow->frcti == NULL) goto fail_frcti; @@ -857,6 +860,13 @@ static int flow_init(struct flow_info * info, pthread_rwlock_unlock(&proc.lock); + if (tx_rb != NULL) { + txq.tv_sec = SSM_RBUFF_TXQ_DELAY / 1000; + txq.tv_nsec = (SSM_RBUFF_TXQ_DELAY % 1000) * MILLION; + + ssm_rbuff_set_txq_target(tx_rb, &txq); + } + return fd; fail_frcti: @@ -1305,6 +1315,8 @@ int flow_dealloc(int fd) pthread_rwlock_unlock(&proc.lock); + frcti_set_draining(flow->frcti); + flow_read(fd, buf, SOCK_BUF_SIZE); pthread_rwlock_rdlock(&proc.lock); @@ -1395,6 +1407,18 @@ int ipcp_flow_dealloc(int fd) return err; } +/* A settable delay is a normalised, non-negative timespec. */ +static bool delay_is_valid(const struct timespec * ts) +{ + if (ts->tv_sec < 0 || ts->tv_nsec < 0) + return false; + + if (ts->tv_nsec >= BILLION) + return false; + + return TS_TO_UINT64(*ts) <= SSM_RBUFF_TXQ_MAX_DELAY; +} + int fccntl(int fd, int cmd, ...) @@ -1417,6 +1441,7 @@ int fccntl(int fd, time_t * rtop; int rc; bool emit_eos = false; + bool set_txq = false; if (fd < 0 || fd >= PROC_MAX_FLOWS) return -EBADF; @@ -1500,7 +1525,10 @@ int fccntl(int fd, if (flow->tx_rb == NULL) goto eperm; - ssm_rbuff_set_txq_target(flow->tx_rb, timeo); + if (!delay_is_valid(timeo)) + goto einval; + + set_txq = true; break; case FLOWGTXQDLY: timeo = va_arg(l, struct timespec *); @@ -1525,21 +1553,21 @@ int fccntl(int fd, /* Our flow write-only -> peer's read-only; restore on RDWR. */ if (flow->oflags & FLOWFWRONLY) - ssm_rbuff_clr_bits(flow->rx_rb, RB_WR); + ssm_rbuff_clr_flags(flow->rx_rb, RB_WR); else - ssm_rbuff_set_bits(flow->rx_rb, RB_WR); + ssm_rbuff_set_flags(flow->rx_rb, RB_WR); if (flow->oflags & FLOWFDOWN) { - ssm_rbuff_set_bits(flow->rx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(flow->rx_rb, RB_FLOWDOWN); if (flow->tx_rb != NULL) - ssm_rbuff_set_bits(flow->tx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(flow->tx_rb, RB_FLOWDOWN); if (flow->set != NULL) ssm_flow_set_notify(flow->set, flow->info.id, FLOW_DOWN); } else { - ssm_rbuff_clr_bits(flow->rx_rb, RB_FLOWDOWN); + ssm_rbuff_clr_flags(flow->rx_rb, RB_FLOWDOWN); if (flow->tx_rb != NULL) - ssm_rbuff_clr_bits(flow->tx_rb, RB_FLOWDOWN); + ssm_rbuff_clr_flags(flow->tx_rb, RB_FLOWDOWN); if (flow->set != NULL) ssm_flow_set_notify(flow->set, flow->info.id, FLOW_UP); @@ -1631,6 +1659,9 @@ int fccntl(int fd, if (emit_eos) frcti_fin_snd(flow->frcti); + if (set_txq) + ssm_rbuff_set_txq_target(flow->tx_rb, timeo); + va_end(l); return 0; @@ -1856,7 +1887,9 @@ static ssize_t flow_write_frag(struct flow * flow, /* * Initiator promotes on the install grace (it holds the key-confirm - * tag); responder waits for peer_synced, with a near-exhaustion floor. + * tag); responder waits for peer_synced. The near-exhaustion floor + * backstops both roles: the receiver selects the epoch by the wire + * selector, so promoting beats wedging TX on a spent keyring. */ static void flow_tx_promote(struct flow * flow) { @@ -1877,7 +1910,7 @@ static void flow_tx_promote(struct flow * flow) promote = ts_diff_ns(&now, &flow->rk_grace) >= 0; } - if (!promote && !flow->rk_initiator) { + if (!promote) { nodes_left = crypt_nodes_left(flow->crypt); promote = nodes_left >= 0 && nodes_left <= REKEY_PROMOTE_FLOOR; } @@ -2636,6 +2669,12 @@ int ipcp_create_r(const struct ipcp_info * info) return irm__irm_result_des(&msg); } +/* Layer-wide bound for flow_info; set once before flows are served. */ +void ipcp_flow_set_max_rtt(uint32_t max_rtt) +{ + proc.max_rtt = max_rtt; +} + int ipcp_flow_req_arr(const buffer_t * dst, qosspec_t qs, time_t mpl, @@ -2657,6 +2696,7 @@ int ipcp_flow_req_arr(const buffer_t * dst, flow.qs = qs; flow.mpl = mpl; flow.mtu = mtu; + flow.max_rtt = proc.max_rtt; if (ipcp_flow_req_arr__irm_req_ser(&msg, dst, &flow, data) < 0) return -ENOMEM; @@ -2732,6 +2772,7 @@ int ipcp_flow_alloc_reply(int fd, flow.mpl = mpl; flow.mtu = mtu; + flow.max_rtt = proc.max_rtt; if (ipcp_flow_alloc_reply__irm_msg_ser(&msg, &flow, response, data) < 0) return -ENOMEM; @@ -2746,8 +2787,14 @@ int ipcp_flow_alloc_reply(int fd, int ipcp_flow_read(int fd, struct ssm_pk_buff ** spb) { - struct flow * flow; - ssize_t idx = -1; + struct flow * flow; + struct ssm_pk_buff * out; + uint8_t * ptr; + ssize_t idx = -1; + ssize_t fret; + size_t len; + size_t nfrags; + int ret; assert(fd >= 0 && fd < PROC_MAX_FLOWS); assert(spb); @@ -2761,6 +2808,24 @@ int ipcp_flow_read(int fd, return -ENOTALLOC; } + if (FRCTI_IS_STREAM(flow->frcti)) { + pthread_rwlock_unlock(&proc.lock); + return -ENOTSUP; + } + + pthread_rwlock_unlock(&proc.lock); + + if (flow->crypt != NULL + && (ssm_rbuff_get_flags(flow->rx_rb) & RB_REKEY)) + flow_rekey(flow); + + /* Advance TX off a stale epoch even on recv-mostly flows. */ + flow_tx_promote(flow); + + tw_move_safe(); + + pthread_rwlock_rdlock(&proc.lock); + /* Raw flow: deliver the popped pkt directly (no FRCT rq). */ if (flow->frcti == NULL) { idx = flow_rx_spb(flow, spb, false, NULL); @@ -2780,14 +2845,59 @@ int ipcp_flow_read(int fd, pthread_rwlock_unlock(&proc.lock); + /* + * A hand-back of the fed spb would leave it double-owned by + * the reorder queue; frcti_consume is the only safe way to + * take it. A write can also complete a PDU, so PDU_READY may + * be true with no loop-local spb to fall back on anyway. + */ + + ret = FRCTI_PDU_INFO(flow->frcti, &len, &nfrags); + if (ret < 0) + return ret; + + /* + * Oversize (over frcti's own cap, or too big for any pool + * class): force frcti_consume's total > count drop branch + * now, so the run leaves the delivery edge instead of + * stalling every read after this one. + */ + if (len > frcti_get_max_rcv_sdu(flow->frcti)) { + (void) FRCTI_CONSUME(flow->frcti, NULL, 0); + return -EMSGSIZE; + } + + idx = ssm_pool_alloc_b(proc.pool, len, &ptr, &out, NULL); + if (idx < 0) { + if (idx == -EMSGSIZE) + (void) FRCTI_CONSUME(flow->frcti, NULL, 0); + return (int) idx; + } + + fret = FRCTI_CONSUME(flow->frcti, ptr, len); + if (fret < 0 || (size_t) fret != len) { + ssm_pool_remove(proc.pool, idx); + return fret < 0 ? (int) fret : -EIO; + } + + *spb = out; + return 0; } +/* + * Writes an spb to an IPCP-internal flow, splitting it over multiple + * FRCT fragments when it exceeds the flow's fragment payload cap. + * Consumes spb on success; on failure spb is left to the caller. + */ int ipcp_flow_write(int fd, struct ssm_pk_buff * spb) { - struct flow * flow; - int ret; + struct flow * flow; + int oflags; + size_t len; + ssize_t fret; + int ret; assert(fd >= 0 && fd < PROC_MAX_FLOWS); assert(spb); @@ -2806,8 +2916,54 @@ int ipcp_flow_write(int fd, return -EPERM; } + if (FRCTI_IS_STREAM(flow->frcti)) { + pthread_rwlock_unlock(&proc.lock); + return -ENOTSUP; + } + + oflags = flow->oflags; + pthread_rwlock_unlock(&proc.lock); + if (flow->crypt != NULL + && (ssm_rbuff_get_flags(flow->rx_rb) & RB_REKEY)) + flow_rekey(flow); + + flow_tx_promote(flow); + + /* Pre-empt TX key exhaustion; the timer is the backstop. */ + if (flow_wm_due(flow)) { + STORE_RELAXED(&flow->rk_wm_inflight, true); + + if (flow_rekey_trigger(flow) < 0) + STORE_RELAXED(&flow->rk_wm_inflight, false); + } + + tw_move_safe(); + + len = ssm_pk_buff_len(spb); + if (FRCTI_NEEDS_FRAG(flow->frcti, len)) { + fret = flow_write_frag(flow, ssm_pk_buff_head(spb), len, + oflags, NULL); + + if (fret < 0) + return (int) fret; + + /* Partial: flow_write_frag swallowed the real cause. */ + if (fret != (ssize_t) len) { + /* PoA flows have no tx_rb flag to consult. */ + if (flow->tx_rb != NULL + && (ssm_rbuff_get_flags(flow->tx_rb) + & RB_FLOWDOWN)) + return -EFLOWDOWN; + return -EIO; + } + + ipcp_spb_release(spb); + + return 0; + } + ret = flow_tx_spb(flow, spb, FRCT_FR_SOLE, true, NULL); return ret; @@ -2875,6 +3031,11 @@ int np1_flow_read(int fd, return 0; } +/* + * An N-1 flow gets no flow_write to advance its TX epoch off a rotated + * key. Promoting is local; a re-key request here would block on the + * IRMd. + */ int np1_flow_write(int fd, struct ssm_pk_buff * spb, struct ssm_pool * pool) @@ -2953,10 +3114,10 @@ int ipcp_flow_fini(int fd) return -1; } - ssm_rbuff_set_bits(proc.flows[fd].rx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(proc.flows[fd].rx_rb, RB_FLOWDOWN); if (proc.flows[fd].tx_rb != NULL) - ssm_rbuff_set_bits(proc.flows[fd].tx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(proc.flows[fd].tx_rb, RB_FLOWDOWN); if (proc.flows[fd].set != NULL) ssm_flow_set_notify(proc.flows[fd].set, proc.flows[fd].info.id, diff --git a/src/lib/frct.c b/src/lib/frct.c index efd50b9a..ecec2543 100644 --- a/src/lib/frct.c +++ b/src/lib/frct.c @@ -25,16 +25,18 @@ #define DELT_RDV (100 * MILLION) /* ns */ #define MAX_RDV (1 * BILLION) /* ns */ -#define MAX_RTO_MUL 8 /* caps the RTO backoff shift */ +#define RXM_TRIES_SHIFT 5 /* >= 32 HoL tries within t_r */ +#define MAX_RTO_MUL 16 /* guard; rxm_backoff clamps */ #define MAX_TLP_PER_EP 2 /* RFC 8985 §7.3: up to 2 TLPs */ -#define INITIAL_RTO (1 * BILLION) /* RFC 6298 §2.1: 1 s default */ #define RTT_BOOT_NS (10 * MILLION) /* rtt_hint floor + initial mdev */ #define SRTT_FLOOR_NS 1000L /* 1 us; smoothed RTT floor */ #define MDEV_FLOOR_NS 100L /* 100 ns; mdev sanity floor */ #define RTT_CLAMP_MUL 16 /* probe sample cap = N * srtt */ #define MIN_RTT_WIN_NS (300ULL * BILLION) /* 5 min, Linux tcp default */ +#define MIN_RTT_SLOTS 3 /* windowed-min sample slots */ #define NACK_COOLDOWN_NS (100 * MILLION) /* pre-DRF NACK cooldown */ #define FRCT_TX_TIMEO_NS (250 * 1000) /* tx ring write deadline */ +#define RTT_LOUD_NS (500 * MILLION) /* diagnostic sample threshold */ #define ACK_DELAY_NS (2ULL * TICTIME) /* delayed-ACK fire delay */ #define FRCT "frct" @@ -49,11 +51,13 @@ #define SACK_MIN_GAP_NS (250u * 1000u) /* 250 us SACK gap */ #define MIN_REORDER_NS (250u * 1000u) /* 250 us RACK floor */ #define SACK_RXM_MAX 32 /* Cap on retransmits staged from single SACK.*/ -#define DUP_THRESH 3 /* RFC 8985 §6.2 step 2.2 SACK count gate. */ +#define DUP_THRESH 3 /* RFC 8985 §6.2 step 4 SACK count gate. */ +/* Repair budget: burst cap on SACK-driven retransmits (tokens). */ +#define RXM_BUDGET_MAX (2 * SACK_RXM_MAX) -/* RFC 8985 §7.2 RACK reorder-window scaling cap. */ +/* RFC 8985 §6.2 RACK reorder-window scaling cap. */ #define REO_WND_MULT_MAX 20 -/* RFC 8985 §7.2 step 5: round trips of no DSACK before halving. */ +/* RFC 8985 §6.2: fresh-ACKed seqnos before decaying the scale. */ #define REO_DECAY_PKTS 16 /* DSACK seqno sanity: reject reports older/farther than one rcv window. */ #define MAX_DSACK_LAG RQ_SIZE @@ -186,15 +190,17 @@ struct frcti_stat { size_t rxm_dup_rcv; /* RXM dups (peer already had it) */ size_t rxm_sack; /* SACK-mechanism retransmits */ size_t rxm_rack; /* RACK-driven retransmits */ - size_t rxm_dupthresh; /* DupThresh-driven retransmits */ + size_t rxm_zero_reo; /* repairs at zero reorder wnd */ size_t rxm_nack; /* NACK-pulled retransmits */ size_t rxm_due_count; /* rxm_due entries (pre-bail) */ size_t rxm_due_acked; /* bail: seqno < snd_lwe */ size_t rxm_due_unowned; /* bail: slot.rxm replaced */ size_t rxm_due_aged; /* bail: r->t0 + t_r < now */ size_t rxm_due_defer; /* bail: non-HoL, deferred to HoL */ + size_t rxm_hol_gone; /* defers with no rxm at HoL slot */ size_t rxm_fast_skip; /* SACK skips: slot has FAST_RXM */ size_t rxm_fast_stuck; /* those skips with age > rto */ + size_t rxm_no_budget; /* SACKs cut short: no repair token*/ size_t rxm_arm_fail; /* rxm_arm: malloc failed */ size_t rxm_cancel; /* entries cancelled at teardown */ size_t rxm_tx_dead; /* RXM tx into terminal flow */ @@ -304,6 +310,11 @@ struct frct_cr { uint64_t inact; /* Inactivity threshold (ns) */ }; +struct rtt_min { + time_t v; /* measured RTT (ns) */ + uint64_t t; /* when it was measured (ns) */ +}; + struct frcti { /* IMM: set once in frcti_create; read-only thereafter. */ int fd; @@ -324,18 +335,17 @@ struct frcti { struct frct_cr rcv_cr; /* RTT/RACK estimator */ - time_t srtt; /* smoothed RTT */ - time_t mdev; /* mean deviation */ - time_t min_rtt; /* RACK base, ns */ - uint64_t t_min_rtt; /* min_rtt last set */ - time_t rto; /* retransmit TO */ - time_t rto_min; /* RTO floor (ns) */ - uint8_t rto_mul; /* RTO backoff bits */ - uint32_t rtt_lwe; /* RTT-sample fence */ - uint64_t t_rcv_rtt; /* last RTT feed */ - uint64_t t_snd_probe; /* last probe sent */ - uint64_t t_latest_ack; /* RACK.fack snd-ts */ - uint32_t probe_id_next; + time_t srtt; /* smoothed RTT */ + time_t mdev; /* mean deviation */ + struct rtt_min min_rtt[MIN_RTT_SLOTS]; + time_t rto; /* retransmit TO */ + time_t rto_min; /* RTO floor (ns) */ + uint8_t rto_mul; /* RTO backoff bits */ + uint32_t rtt_lwe; /* RTT-sample fence */ + uint64_t t_rcv_rtt; /* last RTT feed */ + uint64_t t_snd_probe; /* last probe sent */ + uint64_t t_latest_ack; /* RACK.fack snd-ts */ + uint32_t probe_id_next; struct { uint32_t id; uint64_t ts; /* ts_to_ns send */ @@ -344,6 +354,7 @@ struct frcti { /* rcv reassembly */ size_t max_rcv_sdu; /* max reasm bytes */ + bool draining; /* dealloc drain */ uint8_t * rcv_ring; /* lazy alloc */ size_t rcv_ring_sz; /* power of 2 */ uint32_t ring_seq_cap; /* ring/per_pkt */ @@ -368,12 +379,13 @@ struct frcti { uint32_t dsack_seqno; bool dsack_valid; - /* RFC 8985 §7.2 RACK reorder-window scaling. */ + /* RFC 8985 §6.2 RACK reorder-window scaling. */ uint8_t reo_wnd_mult; /* REO_WND_MULT_MAX */ uint32_t dsack_lwe_snap; /* lwe @ last DSACK */ uint64_t t_last_reo_widen; /* once-per-RTT */ uint32_t dup_thresh; /* RFC 8985 */ + uint32_t rxm_budget; /* repair tokens */ uint32_t tlp_high_seq; /* §7.3: 0 = none */ uint8_t tlp_count; /* §7.3 per-episode */ uint64_t t_nack; @@ -467,7 +479,7 @@ static int frct_rib_read(const char * path, s.srtt = frcti->srtt; s.mdev = frcti->mdev; s.rto = frcti->rto; - s.min_rtt = frcti->min_rtt; + s.min_rtt = frcti->min_rtt[0].v; s.snd_cr = frcti->snd_cr; s.rcv_cr = frcti->rcv_cr; s.stat = frcti->stat; @@ -496,7 +508,7 @@ static int frct_rib_read(const char * path, " duplicates received: %20zu\n" "RXM (SACK mechanism) sent: %20zu\n" "RXM (RACK-driven) sent: %20zu\n" - "RXM (DupThresh-driven) sent: %20zu\n" + "RXM (zero reorder wnd) sent: %20zu\n" "RXM (NACK-driven) sent: %20zu\n" "ACK packets sent: %20zu\n" "Delayed-ACK timer fires: %20zu\n" @@ -551,8 +563,10 @@ static int frct_rib_read(const char * path, " bail (unowned): %20zu\n" " bail (aged): %20zu\n" " bail (defer): %20zu\n" + " defer, no rxm at HoL: %20zu\n" " skip (fast-rxm set): %20zu\n" " skip (stuck past rto): %20zu\n" + " skip (no repair budget): %20zu\n" "RXM-arm malloc failures: %20zu\n" "RXM cancels (teardown): %20zu\n" "RXM tx into dead flow: %20zu\n" @@ -574,7 +588,7 @@ static int frct_rib_read(const char * path, (long long)(now_ns - s.rcv_cr.act), s.rcv_cr.seqno, s.stat.rxm_rto, s.stat.rxm_rcv, s.stat.rxm_dup_rcv, - s.stat.rxm_sack, s.stat.rxm_rack, s.stat.rxm_dupthresh, + s.stat.rxm_sack, s.stat.rxm_rack, s.stat.rxm_zero_reo, s.stat.rxm_nack, s.stat.ack_snd, s.stat.ack_fire, s.stat.ack_supp_seqno, s.stat.ack_supp_inact, @@ -601,7 +615,9 @@ static int frct_rib_read(const char * path, s.stat.rxm_due_count, s.stat.rxm_due_acked, s.stat.rxm_due_unowned, s.stat.rxm_due_aged, s.stat.rxm_due_defer, + s.stat.rxm_hol_gone, s.stat.rxm_fast_skip, s.stat.rxm_fast_stuck, + s.stat.rxm_no_budget, s.stat.rxm_arm_fail, s.stat.rxm_cancel, s.stat.rxm_tx_dead, s.stat.tx_drop, @@ -694,15 +710,15 @@ static __inline__ bool same_epoch_drf(uint32_t seq, /* * RACK reorder window R (RFC 8985 §6.2): * R = MIN(reo_wnd_mult * RACK.min_RTT / 4, SRTT) - * reo_wnd_mult scales on D-SACK evidence of under-tolerance (§7.2). + * reo_wnd_mult scales on D-SACK evidence of under-tolerance (§6.2). * Fall back to srtt when no min_rtt sample exists yet; MIN_REORDER_NS * floor guards collapse below the timer-tick resolution. */ static __inline__ uint64_t rack_reorder_window(struct frcti * frcti) { uint64_t mult = frcti->reo_wnd_mult > 0 ? frcti->reo_wnd_mult : 1; - uint64_t base = frcti->min_rtt > 0 ? (uint64_t) frcti->min_rtt - : (uint64_t) frcti->srtt; + time_t min = frcti->min_rtt[0].v; + uint64_t base = min > 0 ? (uint64_t) min : (uint64_t) frcti->srtt; uint64_t R = mult * (base / 4); R = MAX(R, (uint64_t) MIN_REORDER_NS); @@ -711,6 +727,24 @@ static __inline__ uint64_t rack_reorder_window(struct frcti * frcti) return R; } +/* + * RFC 8985 §6.2 RACK_update_reo_wnd(): as long as no reordering has + * been observed, a repair episode or DupThresh SACKs above the head + * drop the reordering tolerance to zero. This removes the tolerance + * only; the RACK time test still gates every repair. + */ +static __inline__ uint64_t rack_reo_wnd(struct frcti * frcti, + uint64_t R) +{ + if (frcti->reo_wnd_mult > 1) + return R; + + if (frcti->in_recovery || frcti->dup_thresh >= DUP_THRESH) + return 0; + + return R; +} + static __inline__ int frct_spb_reserve(size_t len, struct ssm_pk_buff ** spb) { @@ -827,7 +861,9 @@ static void frct_tx_drop_bump(struct frcti * frcti, STAT_BUMP(frcti, tx_drop_other); } -static int frct_tx(struct frcti * frcti, struct ssm_pk_buff * spb) +static int frct_tx(struct frcti * frcti, + struct ssm_pk_buff * spb, + bool prio) { struct flow * f = frcti_to_flow(frcti); const struct frct_pci * pci; @@ -854,16 +890,33 @@ static int frct_tx(struct frcti * frcti, struct ssm_pk_buff * spb) if (spb_encrypt(f, spb) < 0) goto fail; - idx = ssm_pk_buff_get_off(spb); - - /* DATA blocks; control times out so a full ring can't stall wheel. */ + /* Control times out so a full queue cannot stall the wheel. */ if (!(flags & FRCT_DATA)) { clock_gettime(PTHREAD_COND_CLOCK, &now); ts_add(&now, &intv, &deadline); + dl = &deadline; } - ret = ssm_rbuff_write_b(f->tx_rb, idx, dl); + if (f->poa != NULL) { + ret = poa_flow_tx(f->poa, spb, true, dl); + if (ret < 0) + goto fail; + + return 0; + } + + idx = ssm_pk_buff_get_off(spb); + + /* + * The peer is already waiting on a retransmission, so it skips + * the occupancy limit and never waits: the timer that sent it + * must not block, and the r-timer retries what does not fit. + */ + if (prio) + ret = ssm_rbuff_write_prio(f->tx_rb, idx); + else + ret = ssm_rbuff_write_b(f->tx_rb, idx, dl); if (ret < 0) goto fail; @@ -883,10 +936,10 @@ static void frct_mark_flow_down(struct frcti * frcti) struct flow * f = frcti_to_flow(frcti); if (f->rx_rb != NULL) - ssm_rbuff_set_bits(f->rx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(f->rx_rb, RB_FLOWDOWN); if (f->tx_rb != NULL) - ssm_rbuff_set_bits(f->tx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(f->tx_rb, RB_FLOWDOWN); } __attribute__((cold)) @@ -895,7 +948,7 @@ static void frct_mark_peer_dead(struct frcti * frcti) struct flow * f = frcti_to_flow(frcti); if (f->rx_rb != NULL) - ssm_rbuff_set_bits(f->rx_rb, RB_FLOWPEER); + ssm_rbuff_set_flags(f->rx_rb, RB_FLOWPEER); if (proc.fqset != NULL) ssm_flow_set_notify(proc.fqset, f->info.id, FLOW_PEER); @@ -955,14 +1008,30 @@ static void frcti_pkt_snd(struct frcti * frcti, frct_hcs_set(pci, false); - frct_tx(frcti, spb); + frct_tx(frcti, spb, false); +} + +/* Restart the window from a single sample. */ +static __inline__ void min_rtt_seed(struct frcti * frcti, + time_t mrtt, + uint64_t now_ns) +{ + size_t i; + + for (i = 0; i < MIN_RTT_SLOTS; i++) { + frcti->min_rtt[i].v = mrtt; + frcti->min_rtt[i].t = now_ns; + } } /* RTO floor scales with srtt; hard floor rto_min guards sub-ms RTT. */ static void rtt_init(struct frcti * frcti, - time_t rtt_hint) + time_t rtt_hint, + uint32_t max_rtt, + uint64_t now_ns) { time_t floor; + time_t cap; if (rtt_hint > 0) { rtt_hint = MAX(rtt_hint, (time_t) RTT_BOOT_NS); @@ -970,42 +1039,85 @@ static void rtt_init(struct frcti * frcti, frcti->mdev = rtt_hint >> 3; floor = MAX(frcti->rto_min, 2 * frcti->srtt); frcti->rto = MAX(floor, rtt_hint + (frcti->mdev << MDEV_MUL)); - frcti->min_rtt = rtt_hint; + + min_rtt_seed(frcti, rtt_hint, now_ns); } else { - /* Boot from first ACK. */ + /* Boot from first ACK; declared max path RTT caps RTO. */ + cap = (time_t) (frcti->t_r >> RXM_TRIES_SHIFT); + + if (max_rtt > 0) + cap = MIN(cap, (time_t) max_rtt * 2 * MILLION); frcti->srtt = 0; frcti->mdev = RTT_BOOT_NS; - frcti->rto = MAX((time_t) INITIAL_RTO, frcti->rto_min); - frcti->min_rtt = 0; + frcti->rto = MAX(cap, frcti->rto_min); + + min_rtt_seed(frcti, 0, now_ns); } frcti->rto_mul = 0; } -/* RFC 8985 §6.2: replace min_RTT on unset, smaller sample, or expiry. */ -static __inline__ bool min_rtt_stale(struct frcti * frcti, - time_t mrtt, - uint64_t now_ns) +/* Promote the runners-up as the window slides past each slot. */ +static __inline__ void min_rtt_subwin(struct frcti * frcti, + const struct rtt_min * val) { - if (frcti->min_rtt == 0) - return true; + struct rtt_min * s = frcti->min_rtt; + int64_t dt = ts_age_ns(val->t, s[0].t); + int64_t win = (int64_t) MIN_RTT_WIN_NS; - if (mrtt < frcti->min_rtt) - return true; + /* A clock step or an out-of-order stamp: hold the window. */ + if (dt < 0) + return; - return ts_aged_ns(now_ns, frcti->t_min_rtt, MIN_RTT_WIN_NS); + if (dt > win) { + /* Slot 0 fell out; slot 1 may be stale in turn. */ + s[0] = s[1]; + s[1] = s[2]; + s[2] = *val; + if (ts_aged_ns(val->t, s[0].t, MIN_RTT_WIN_NS)) { + s[0] = s[1]; + s[1] = s[2]; + s[2] = *val; + } + } else if (s[1].t == s[0].t && dt > win / 4) { + s[2] = s[1] = *val; + } else if (s[2].t == s[1].t && dt > win / 2) { + s[2] = *val; + } } -/* Linux-style windowed-min refresh of RACK.min_RTT. */ +/* + * Windowed minimum of RACK.min_RTT over MIN_RTT_WIN_NS, after Linux + * lib/minmax.c. Slots 1 and 2 hold minima over the trailing 3/4 and + * 1/2 of the window, so when slot 0 ages out the estimate drops back + * to a true minimum over what remains rather than to a spot sample. + */ static __inline__ void min_rtt_update(struct frcti * frcti, time_t mrtt, uint64_t now_ns) { - if (!min_rtt_stale(frcti, mrtt, now_ns)) + struct rtt_min * s = frcti->min_rtt; + struct rtt_min val; + + if (mrtt <= 0) return; - frcti->min_rtt = mrtt; - frcti->t_min_rtt = now_ns; + val.v = mrtt; + val.t = now_ns; + + /* New min, unseeded, or nothing left in the window. */ + if (s[0].v == 0 || mrtt <= s[0].v + || ts_aged_ns(now_ns, s[2].t, MIN_RTT_WIN_NS)) { + min_rtt_seed(frcti, mrtt, now_ns); + return; + } + + if (mrtt <= s[1].v) + s[1] = s[2] = val; + else if (mrtt <= s[2].v) + s[2] = val; + + min_rtt_subwin(frcti, &val); } static void rtt_update(struct frcti * frcti, @@ -1040,8 +1152,15 @@ static void rtt_update(struct frcti * frcti, floor = MAX(frcti->rto_min, 2 * frcti->srtt); rto = MAX(floor, frcti->srtt + (frcti->mdev << MDEV_MUL)); + /* FIXME: align with t_r; an rto that spans it retries nothing. */ STORE_RELEASE(&frcti->rto, rto); STORE_RELEASE(&frcti->rto_mul, 0); + + /* Diagnostic: a sample this large is not a path RTT. */ + if (mrtt > RTT_LOUD_NS) + log_warn("RTT sample %lld ms, srtt %lld ms on fd %d.", + (long long) mrtt / MILLION, + (long long) frcti->srtt / MILLION, frcti->fd); } /* Fill probes[pos], return new probe_id; 0 on entropy failure. Wrlock. */ @@ -1116,7 +1235,7 @@ static void frcti_rttp_snd(struct frcti * frcti, rttp->echo_id = hton32(echo_id); memcpy(rttp->nonce, nonce, sizeof(rttp->nonce)); - frct_tx(frcti, spb); + frct_tx(frcti, spb, false); } struct rxm_entry { @@ -1142,6 +1261,29 @@ static bool rxm_still_owned(struct frcti * frcti, } /* + * Backoff clamped to a fixed fraction of t_r, so the ladder always + * leaves room for 1 << RXM_TRIES_SHIFT tries inside the flow's life + * whatever t_r is. Never returns less than the RTO estimate itself: + * on a path whose RTT is large against t_r that many tries do not + * fit, and retrying faster than the estimate only duplicates. + */ +static uint64_t rxm_backoff(struct frcti * frcti, + time_t rto, + uint8_t rto_mul) +{ + uint64_t cap = frcti->t_r >> RXM_TRIES_SHIFT; + + if (cap < (uint64_t) rto) + return (uint64_t) rto; + + /* Compare before shifting; the product can overflow at large t_r. */ + if (rto_mul >= 64 || (uint64_t) rto > (cap >> rto_mul)) + return cap; + + return (uint64_t) rto << rto_mul; +} + +/* * All in-flight slots share the HoL backoff; otherwise non-HoL timers * cycle at base RTO and storm the wire while HoL is still backing off. */ @@ -1151,7 +1293,7 @@ static uint64_t rxm_next_deadline(struct frcti * frcti, time_t rto = LOAD_RELAXED(&frcti->rto); uint8_t rto_mul = LOAD_RELAXED(&frcti->rto_mul); - return now_ns + ((uint64_t) rto << rto_mul); + return now_ns + rxm_backoff(frcti, rto, rto_mul); } /* Copy pkt, set FRCT_RXM, refresh ackno, re-seal HCS. */ @@ -1216,7 +1358,7 @@ static void rxm_snd(struct frcti * frcti, if (seqno == snd_lwe && frcti->rto_mul < MAX_RTO_MUL) STORE_RELEASE(&frcti->rto_mul, frcti->rto_mul + 1); - /* RFC 8985 §7.2 step 4: RTO on HoL resets RACK reo scaling. */ + /* RFC 8985 §6.3: RTO on HoL resets RACK reo scaling. */ if (seqno == snd_lwe) frcti->reo_wnd_mult = 1; @@ -1229,7 +1371,7 @@ static void rxm_snd(struct frcti * frcti, return; /* ETIMEDOUT/ENOMEM: let r-timer drive teardown. */ - ret = frct_tx(frcti, spb); + ret = frct_tx(frcti, spb, true); if (ret == -EFLOWDOWN || ret == -ENOTALLOC) STAT_BUMP(frcti, rxm_tx_dead); } @@ -1265,6 +1407,24 @@ static void rxm_due(void * arg) /* R-timer expired: peer unreachable. */ if (RXM_AGED_OUT(r->t0, now_ns, frcti->t_r)) { STAT_BUMP(frcti, rxm_due_aged); + log_warn("Flow down: rxm seq=%u aged out (hol=%u) " + "age_ms=%llu t_r_ms=%llu rto_ms=%llu mul=%u " + "ack_age_ms=%lld hol_rxm=%s hol_flags=0x%x " + "budget=%u tlp_hi=%u tlp_n=%u on fd %d.", + r->seqno, snd_lwe, + (unsigned long long)(now_ns - r->t0) / MILLION, + (unsigned long long) frcti->t_r / MILLION, + (unsigned long long) LOAD_RELAXED(&frcti->rto) + / MILLION, + (unsigned) LOAD_RELAXED(&frcti->rto_mul), + (long long)(now_ns - frcti->t_latest_ack) / MILLION, + LOAD_ACQUIRE(&frcti->snd_slots[RQ_SLOT(snd_lwe)].rxm) + == NULL ? "none" : "live", + (unsigned) frcti->snd_slots[RQ_SLOT(snd_lwe)].flags, + (unsigned) frcti->rxm_budget, + frcti->tlp_high_seq, + (unsigned) frcti->tlp_count, + frcti->fd); frct_mark_flow_down(frcti); goto cleanup; } @@ -1272,8 +1432,11 @@ static void rxm_due(void * arg) /* HoL-only retx; defer at base rto so HoL transitions react. */ if (r->seqno != snd_lwe) { STAT_BUMP(frcti, rxm_due_defer); - tw_post(&r->tw, now_ns + LOAD_RELAXED(&frcti->rto), - rxm_due, r); + + if (LOAD_ACQUIRE(&frcti->snd_slots[RQ_SLOT(snd_lwe)].rxm) + == NULL) + STAT_BUMP(frcti, rxm_hol_gone); + tw_post(&r->tw, now_ns + LOAD_RELAXED(&frcti->rto), rxm_due, r); return; } @@ -1340,7 +1503,7 @@ static void rxm_arm(struct frcti * frcti, rto = LOAD_RELAXED(&frcti->rto); rto_mul = LOAD_RELAXED(&frcti->rto_mul); - deadline = r->t0 + ((uint64_t) rto << rto_mul); + deadline = r->t0 + rxm_backoff(frcti, rto, rto_mul); pthread_rwlock_wrlock(&frcti->lock); @@ -1476,7 +1639,7 @@ static void frcti_sack_snd(struct frcti * frcti, for (i = 0; i < sa->n; ++i) sack_block_put(buf.data, i, sa->blocks[i][0], sa->blocks[i][1]); - frct_tx(frcti, spb); + frct_tx(frcti, spb, false); } static void ack_snd(struct frcti * frcti, @@ -1653,6 +1816,8 @@ static void ka_snd(struct frcti * frcti) snd_idle = ts_age_ns(now_ns, LOAD_RELAXED(&frcti->snd_cr.act)); if (rcv_idle > timeo_ns) { + log_warn("Peer dead: rcv idle %lld ms on fd %d.", + (long long) rcv_idle / MILLION, frcti->fd); frct_mark_peer_dead(frcti); return; } @@ -1675,7 +1840,7 @@ static void ka_snd(struct frcti * frcti) frct_hcs_set(pci, false); STAT_BUMP(frcti, ka_snd); - frct_tx(frcti, spb); + frct_tx(frcti, spb, false); ka_arm(frcti); } @@ -1807,6 +1972,7 @@ struct frcti * frcti_create(int fd, uint64_t r, uint64_t mpl, time_t rtt_hint, + uint32_t max_rtt, qosspec_t qs, uint32_t mtu) { @@ -1876,8 +2042,8 @@ struct frcti * frcti_create(int fd, } frcti->rto_min = (time_t) MAX(RTO_MIN, 1ULL << RXMQ_RES); - rtt_init(frcti, rtt_hint); - frcti->t_min_rtt = now_ns; + + rtt_init(frcti, rtt_hint, max_rtt, now_ns); frcti->probe_id_next = 1; frcti->t_rcv_rtt = now_ns; frcti->t_snd_probe = now_ns; @@ -1894,6 +2060,7 @@ struct frcti * frcti_create(int fd, frcti->in_recovery = false; frcti->recovery_high = 0; frcti->rack_fired_lwe = 0; + frcti->rxm_budget = RXM_BUDGET_MAX; tw_init_entry(&frcti->ack_tw); tw_init_entry(&frcti->ka_tw); @@ -1954,9 +2121,10 @@ void frcti_destroy(struct frcti * frcti) printf("[FRCT teardown] pid=%d fd=%d " "sdu_snd=%zu sdu_reasm=%zu sdu_sole=%zu " "frag_snd=%zu frag_rcv=%zu frag_drop=%zu " - "rxm_rto=%zu rxm_sack=%zu rxm_rack=%zu rxm_dup=%zu " + "rxm_rto=%zu rxm_sack=%zu rxm_rack=%zu rxm_zreo=%zu " "rxm_due=%zu acked=%zu unowned=%zu aged=%zu defer=%zu " - "fast_skip=%zu fast_stuck=%zu " + "hol_gone=%zu " + "fast_skip=%zu fast_stuck=%zu no_budget=%zu " "cancel=%zu arm_fail=%zu inflight=%u " "nack_snd=%zu nack_rcv=%zu inact_drop=%zu " "tlp_snd=%zu sack_snd=%zu sack_rcv=%zu ack_supp=%zu " @@ -1968,11 +2136,13 @@ void frcti_destroy(struct frcti * frcti) frcti->stat.frag_snd, frcti->stat.frag_rcv, frcti->stat.frag_drop, frcti->stat.rxm_rto, frcti->stat.rxm_sack, - frcti->stat.rxm_rack, frcti->stat.rxm_dupthresh, + frcti->stat.rxm_rack, frcti->stat.rxm_zero_reo, frcti->stat.rxm_due_count, frcti->stat.rxm_due_acked, frcti->stat.rxm_due_unowned, frcti->stat.rxm_due_aged, frcti->stat.rxm_due_defer, + frcti->stat.rxm_hol_gone, frcti->stat.rxm_fast_skip, frcti->stat.rxm_fast_stuck, + frcti->stat.rxm_no_budget, frcti->stat.rxm_cancel, frcti->stat.rxm_arm_fail, frcti->snd_cr.seqno - frcti->snd_cr.lwe, frcti->stat.nack_snd, frcti->stat.nack_rcv, @@ -2052,6 +2222,19 @@ int frcti_set_max_rcv_sdu(struct frcti * frcti, return 0; } +/* Dealloc drain discards SDUs by design; don't count them as drops. */ +static void frcti_set_draining(struct frcti * frcti) +{ + if (frcti == NULL) + return; + + pthread_rwlock_wrlock(&frcti->lock); + + frcti->draining = true; + + pthread_rwlock_unlock(&frcti->lock); +} + size_t frcti_get_rcv_ring_sz(struct frcti * frcti) { size_t ret; @@ -2168,7 +2351,7 @@ static void sack_rxm_snd(struct frcti * frcti, rxm_arm(frcti, seqno, rxm, spb); STAT_BUMP(frcti, rxm_sack); - ret = frct_tx(frcti, spb); + ret = frct_tx(frcti, spb, true); if (ret == -EFLOWDOWN || ret == -ENOTALLOC) STAT_BUMP(frcti, rxm_tx_dead); } @@ -2187,7 +2370,7 @@ static int fast_rxm_send(struct frcti * frcti, if (spb == NULL) return 0; - return frct_tx(frcti, spb); + return frct_tx(frcti, spb, true); } /* PCI bytes survive head_release at receive; just rewind the pointer. */ @@ -2642,14 +2825,16 @@ static ssize_t frcti_consume(struct frcti * frcti, goto unlock; } if (st == FRAG_DROP) { - STAT_ADD(frcti, frag_drop, n); + if (!frcti->draining) + STAT_ADD(frcti, frag_drop, n); frag_drop(frcti, n); continue; } /* FRAG_DELIVER */ total = frag_total_len(frcti, n, &overflow); if (overflow || total > frcti->max_rcv_sdu || total > count) { - STAT_ADD(frcti, frag_drop, n); + if (!frcti->draining) + STAT_ADD(frcti, frag_drop, n); frag_drop(frcti, n); ret = -EMSGSIZE; goto unlock; @@ -2698,6 +2883,49 @@ static bool frcti_pdu_ready(struct frcti * frcti) return ready; } +/* + * Size a ready SDU before consuming it: *len is the total byte + * count, *nfrags the fragment count. 0 on success, -EAGAIN if no + * complete SDU is ready (includes the stream and overflow cases). + */ +static int frcti_pdu_info(struct frcti * frcti, + size_t * len, + size_t * nfrags) +{ + size_t count; + bool overflow; + int ret; + + assert(frcti); + + pthread_rwlock_rdlock(&frcti->lock); + + if (frcti->stream) { + ret = -EAGAIN; + goto unlock; + } + + if (frag_run_inspect(frcti, &count) != FRAG_DELIVER) { + ret = -EAGAIN; + goto unlock; + } + + *len = frag_total_len(frcti, count, &overflow); + + if (overflow) { + ret = -EAGAIN; + goto unlock; + } + + *nfrags = count; + ret = 0; + + unlock: + pthread_rwlock_unlock(&frcti->lock); + + return ret; +} + /* No srtt yet: probe at the cold-probe cadence to seed it. */ #define PROBE_DUE_COLD(frcti, now_ns) \ ((now_ns) - (frcti)->t_snd_probe > (uint64_t) RTTP_COLD_NS) @@ -2967,8 +3195,8 @@ static void tlp_due(void * arg) goto unlock; /* Cap: if HoL RTO is due, let rxm_due fire instead. */ - rto_at = rxm->t0 + ((uint64_t) frcti->rto - << LOAD_RELAXED(&frcti->rto_mul)); + rto_at = rxm->t0 + rxm_backoff(frcti, frcti->rto, + LOAD_RELAXED(&frcti->rto_mul)); if (rto_at <= now_ns) goto unlock; @@ -3126,19 +3354,21 @@ static void fast_rxm_consider(struct frcti * frcti, struct snd_slot * slot; size_t hp; uint64_t R; - bool rack_ok; + uint64_t reo; + int64_t age; hp = RQ_SLOT(frcti->snd_cr.lwe); slot = &frcti->snd_slots[hp]; rxm = LOAD_ACQUIRE(&slot->rxm); R = rack_reorder_window(frcti); + reo = rack_reo_wnd(frcti, R); if (RXM_SLOT_EMPTY(rxm)) return; - /* RFC 8985 §6.2: time-based RACK OR DupThresh count. */ - rack_ok = (int64_t)(frcti->t_latest_ack - slot->time) > (int64_t) R; - if (!rack_ok && frcti->dup_thresh < DUP_THRESH) + /* RFC 8985 §6.2: last transmission older than the latest ack + reo. */ + age = (int64_t)(frcti->t_latest_ack - slot->time); + if (age <= (int64_t) reo) return; /* HoL aged past t_r; let rxm_due tear the flow down. */ @@ -3159,10 +3389,11 @@ static void fast_rxm_consider(struct frcti * frcti, memcpy(pending->fast_rxm.data, rxm->pkt, rxm->len); slot->flags |= SND_RTX | SND_FAST_RXM; frcti->rtt_lwe = frcti->snd_cr.lwe + 1; - if (rack_ok) + + if (age > (int64_t) R) STAT_BUMP(frcti, rxm_rack); else - STAT_BUMP(frcti, rxm_dupthresh); + STAT_BUMP(frcti, rxm_zero_reo); } /* Caller holds wrlock; RACK fast retransmit queued in pending. */ @@ -3175,6 +3406,7 @@ static void frcti_ack_rcv(struct frcti * frcti, { uint32_t ackno; uint32_t lwe; + uint64_t t_ack; size_t p; size_t fresh; @@ -3199,6 +3431,12 @@ static void frcti_ack_rcv(struct frcti * frcti, STORE_RELEASE(&frcti->snd_cr.lwe, ackno); + /* Packet conservation: one repair token per seqno that left. */ + frcti->rxm_budget += ackno - lwe; + + if (frcti->rxm_budget > RXM_BUDGET_MAX) + frcti->rxm_budget = RXM_BUDGET_MAX; + /* §7.3: cum-ACK past the probed seqno resolves the TLP. */ if (frcti->tlp_high_seq != 0 && !before(ackno, frcti->tlp_high_seq)) { @@ -3210,7 +3448,7 @@ static void frcti_ack_rcv(struct frcti * frcti, if (ackno == frcti->snd_cr.seqno) frcti->tlp_count = 0; - /* RFC 8985 §7.2: halve mult per REO_DECAY_PKTS fresh-ACK'd seqnos. */ + /* RFC 8985 §6.2: halve mult per REO_DECAY_PKTS fresh-ACK'd seqnos. */ fresh = ackno - frcti->dsack_lwe_snap; if (frcti->reo_wnd_mult > 1 && fresh >= REO_DECAY_PKTS) { uint8_t half = frcti->reo_wnd_mult >> 1; @@ -3218,8 +3456,15 @@ static void frcti_ack_rcv(struct frcti * frcti, frcti->dsack_lwe_snap = ackno; } - /* RFC 8985: latest cum-ACKed send-time (slot of ackno-1). */ - frcti->t_latest_ack = frcti->snd_slots[RQ_SLOT(ackno - 1)].time; + /* + * RFC 8985 §6.2 RACK_sent_after: RACK.xmit_ts only ever moves + * forward. A cum-ACK covers older seqnos than the SACK blocks + * that raised it, so assigning here would drop it back and + * wedge the loss test for every hole above the cum-ACK. + */ + t_ack = frcti->snd_slots[RQ_SLOT(ackno - 1)].time; + if (t_ack > frcti->t_latest_ack) + frcti->t_latest_ack = t_ack; /* RFC 8985: SACK-above-lwe count is per-recovery-episode. */ frcti->dup_thresh = 0; @@ -3247,10 +3492,12 @@ static void frcti_ack_rcv(struct frcti * frcti, static uint32_t sack_mark_blocks(struct frcti * frcti, const uint8_t * payload, uint16_t n, - uint32_t * newly_marked) + uint32_t * newly_marked, + uint64_t now_ns) { uint32_t hi_sacked = frcti->snd_cr.lwe; uint32_t marked = 0; + uint64_t rtt_t = 0; /* freshest send time worth timing */ uint16_t i; for (i = 0; i < n; ++i) { @@ -3273,10 +3520,14 @@ static uint32_t sack_mark_blocks(struct frcti * frcti, for (k = s; before(k, e); ++k) { size_t kp = RQ_SLOT(k); uint64_t t_k; + uint8_t f_k; if (clamped && k == frcti->snd_cr.lwe) continue; if (LOAD_ACQUIRE(&frcti->snd_slots[kp].rxm) == NULL) continue; + + f_k = frcti->snd_slots[kp].flags; + STORE_RELEASE(&frcti->snd_slots[kp].rxm, NULL); frcti->snd_slots[kp].flags = 0; marked++; @@ -3284,12 +3535,38 @@ static uint32_t sack_mark_blocks(struct frcti * frcti, t_k = frcti->snd_slots[kp].time; if (t_k > frcti->t_latest_ack) frcti->t_latest_ack = t_k; + + /* Karn: a retransmitted seqno times nothing. */ + if (f_k & (SND_RTX | SND_TLP | SND_FAST_RXM)) + continue; + + if (before(k, frcti->rtt_lwe)) + continue; + + if (t_k > rtt_t) + rtt_t = t_k; } if (after(e, hi_sacked)) hi_sacked = e; } + /* + * One sample per SACK, off the freshest packet it confirms. + * A hole keeps every seqno out of the cum-ACK path, so this + * is the only estimator input while one is open. Seeding is + * still left to the probes. + */ + if (rtt_t > 0 && frcti->srtt != 0) { + int64_t mrtt = ts_age_ns(now_ns, rtt_t); + + if (mrtt > 0) { + rtt_update(frcti, (time_t) mrtt, now_ns); + + frcti->t_rcv_rtt = now_ns; + } + } + *newly_marked = marked; return hi_sacked; } @@ -3300,9 +3577,9 @@ static void sack_queue_rxm(struct frcti * frcti, uint64_t now_ns, struct pending * pending) { - uint64_t R = rack_reorder_window(frcti); + uint64_t R = rack_reorder_window(frcti); + uint64_t reo = rack_reo_wnd(frcti, R); uint32_t k; - bool rack_ok; for (k = frcti->snd_cr.lwe; before(k, hi_sacked); ++k) { struct rxm_entry * rxm; @@ -3318,15 +3595,20 @@ static void sack_queue_rxm(struct frcti * frcti, if (rxm == NULL) continue; + /* Repairs are ACK-clocked; RTO/HoL cover a dry bucket. */ + if (frcti->rxm_budget == 0) { + STAT_BUMP(frcti, rxm_no_budget); + break; + } + /* - * A fast-retx still outstanding after its own RTO is - * presumed lost; clear the flag so RACK can repair it - * again instead of stranding it until the HoL timer. + * A fast-retx outstanding past the reorder window is + * presumed lost in turn; clear the flag so RACK can + * repair it again. The rack_ok test below still needs + * an ack for a later packet, so this cannot storm. */ if (frcti->snd_slots[kp].flags & SND_FAST_RXM) { - if (!ts_aged_ns(now_ns, - frcti->snd_slots[kp].time, - LOAD_RELAXED(&frcti->rto))) { + if (!ts_aged_ns(now_ns, frcti->snd_slots[kp].time, R)) { STAT_BUMP(frcti, rxm_fast_skip); continue; } @@ -3339,15 +3621,14 @@ static void sack_queue_rxm(struct frcti * frcti, continue; rack_age = frcti->t_latest_ack - frcti->snd_slots[kp].time; - /* RFC 8985 §6.2: time-based RACK OR DupThresh count. */ - rack_ok = (int64_t) rack_age > (int64_t) R; - if (!rack_ok && frcti->dup_thresh < DUP_THRESH) + /* RFC 8985 §6.2: last transmission older than latest + reo. */ + if ((int64_t) rack_age <= (int64_t) reo) continue; - if (rack_ok) + if ((int64_t) rack_age > (int64_t) R) STAT_BUMP(frcti, rxm_rack); else - STAT_BUMP(frcti, rxm_dupthresh); + STAT_BUMP(frcti, rxm_zero_reo); pending->sack_rxm[cnt].data = malloc(rxm->len); if (pending->sack_rxm[cnt].data == NULL) @@ -3356,6 +3637,7 @@ static void sack_queue_rxm(struct frcti * frcti, pending->sack_rxm[cnt].len = rxm->len; memcpy(pending->sack_rxm[cnt].data, rxm->pkt, rxm->len); pending->sack_rxm_cnt++; + frcti->rxm_budget--; /* NULL slot so the original timer self-cleans. */ STORE_RELEASE(&frcti->snd_slots[kp].rxm, NULL); frcti->snd_slots[kp].time = now_ns; @@ -3409,7 +3691,7 @@ static bool sack_is_dsack(struct frcti * frcti, return false; } -/* RFC 8985 §7.2: grow reo_wnd_mult on DSACK; at most once per RTT. */ +/* RFC 8985 §6.2: grow reo_wnd_mult on DSACK; at most once per RTT. */ static __inline__ void reo_wnd_on_dsack(struct frcti * frcti, uint64_t now_ns) { @@ -3466,9 +3748,15 @@ static void frcti_sack_rcv(struct frcti * frcti, recovery_enter(frcti); marked = 0; - hi_sacked = sack_mark_blocks(frcti, pkt.data, n, &marked); + hi_sacked = sack_mark_blocks(frcti, pkt.data, n, &marked, now_ns); frcti->dup_thresh += marked; + /* Packet conservation: a newly SACKed seqno also left the wire. */ + frcti->rxm_budget += marked; + + if (frcti->rxm_budget > RXM_BUDGET_MAX) + frcti->rxm_budget = RXM_BUDGET_MAX; + if (after(hi_sacked, frcti->snd_cr.lwe)) sack_queue_rxm(frcti, hi_sacked, now_ns, pending); } @@ -3509,7 +3797,7 @@ static void frcti_nack_snd(struct frcti * frcti, frct_hcs_set(pci, false); - frct_tx(frcti, spb); + frct_tx(frcti, spb, false); } enum frct_act { @@ -3881,7 +4169,7 @@ static void frcti_fin_snd(struct frcti * frcti) return; } - if (frct_tx(frcti, spb) < 0) + if (frct_tx(frcti, spb, false) < 0) return; pthread_rwlock_wrlock(&frcti->lock); @@ -4211,6 +4499,9 @@ static void frcti_rcv(struct frcti * frcti, #define FRCTI_PDU_READY(frcti) \ ((frcti) != NULL && frcti_pdu_ready(frcti)) +#define FRCTI_PDU_INFO(frcti, len, nfrags) \ + ((frcti) == NULL ? -EAGAIN : frcti_pdu_info((frcti), (len), (nfrags))) + #define FRCTI_CONSUME(frcti, buf, count) \ ((frcti) == NULL ? (ssize_t) -EAGAIN \ : (frcti)->stream \ diff --git a/src/lib/pb/model.proto b/src/lib/pb/model.proto index e00f3e12..f3e85c33 100644 --- a/src/lib/pb/model.proto +++ b/src/lib/pb/model.proto @@ -42,6 +42,7 @@ message flow_info_msg { required qosspec_msg qos = 6; required uint32 uid = 7; required uint32 mtu = 8; /* Layer MTU (bytes). */ + required uint32 max_rtt = 9; /* Max path RTT (ms). */ } message name_info_msg { diff --git a/src/lib/poa/poa.c b/src/lib/poa/poa.c index b40d9fea..1fd91660 100644 --- a/src/lib/poa/poa.c +++ b/src/lib/poa/poa.c @@ -1501,11 +1501,11 @@ static size_t flows_updown(struct poa * poa, if (up) { flow->oflags &= ~FLOWFDOWN; - ssm_rbuff_clr_bits(flow->rx_rb, RB_FLOWDOWN); + ssm_rbuff_clr_flags(flow->rx_rb, RB_FLOWDOWN); } else { flow->oflags |= FLOWFDOWN; - ssm_rbuff_set_bits(flow->rx_rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(flow->rx_rb, RB_FLOWDOWN); } ssm_flow_set_notify(proc.fqset, pf->flow_id, diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c index aa247755..6eec1159 100644 --- a/src/lib/protobuf.c +++ b/src/lib/protobuf.c @@ -83,6 +83,7 @@ flow_info_msg_t * flow_info_s_to_msg(const struct flow_info * s) msg->state = s->state; msg->uid = s->uid; msg->mtu = s->mtu; + msg->max_rtt = s->max_rtt; msg->qos = qos_spec_s_to_msg(&s->qs); if (msg->qos == NULL) goto fail_msg; @@ -110,6 +111,7 @@ struct flow_info flow_info_msg_to_s(const flow_info_msg_t * msg) s.state = msg->state; s.uid = msg->uid; s.mtu = msg->mtu; + s.max_rtt = msg->max_rtt; s.qs = qos_spec_msg_to_s(msg->qos); return s; diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c index e35a27a9..0480bce1 100644 --- a/src/lib/ssm/rbuff.c +++ b/src/lib/ssm/rbuff.c @@ -27,6 +27,7 @@ #include <ouroboros/ssm_rbuff.h> #include <ouroboros/lockfile.h> +#include <ouroboros/atomics.h> #include <ouroboros/errno.h> #include <ouroboros/fccntl.h> #include <ouroboros/pthread.h> @@ -53,13 +54,6 @@ #define MODB(x) ((x) & (SSM_RBUFF_SIZE - 1)) -#define LOAD_RELAXED(ptr) (__atomic_load_n(ptr, __ATOMIC_RELAXED)) -#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)]) #define HEAD_IDX(rb) (LOAD_ACQUIRE(rb->head)) @@ -69,22 +63,21 @@ #define ADVANCE_TAIL(rb) \ (STORE_RELEASE(rb->tail, MODB(LOAD_RELAXED(rb->tail) + 1))) #define QUEUED(rb) (MODB(HEAD_IDX(rb) - TAIL_IDX(rb))) -#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) +/* Delay-bound the TX queue delay at rate * target. */ +#define TXQ_MIN_SLOTS 4 /* floor: jitter margin */ +#define TXQ_INIT_SLOTS 64 /* ceiling until measured */ +#define TXQ_EWMA_N 4 /* EWMA weight 1/4 */ +#define TXQ_SPW_SHIFT 3 /* aim: 8 samples per window */ +#define TXQ_PERIOD_INIT 16 /* writes between samples */ +#define TXQ_PERIOD_MIN 4 +#define TXQ_PERIOD_MAX 64 +#define TXQ_MIN_DT_NS 1000LL /* shorter windows are noise */ +#define TXQ_MAX_RATE BILLION /* keeps rate * target in s64 */ +#define TXQ_UNLIMITED (SSM_RBUFF_SIZE - 1) +#define TXQ_DATA_MAX (TXQ_UNLIMITED - SSM_RBUFF_TXQ_RESERVE) struct ssm_rbuff { ssize_t * shm_base; /* start of shared memory */ @@ -97,20 +90,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 */ + struct { + uint64_t target; /* target queue delay, ns */ + uint64_t rate; /* EWMA drain rate, slots/s */ + uint64_t ns; /* window start, 0 = unset */ + size_t limit; /* current occupancy limit */ + size_t wr; /* writes this window */ + size_t due; /* sample when wr hits this */ + size_t period; /* writes between samples */ + size_t q0; /* queued at window start */ + bool idle; /* ring ran empty this one */ + bool measured; /* rate holds a measurement */ + } txq; /* tx delay limiter state */ }; -#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, @@ -142,24 +135,29 @@ static struct ssm_rbuff * rbuff_create(pid_t pid, rb->shm_base = shm_base; rb->head = (size_t *) (rb->shm_base + (SSM_RBUFF_SIZE)); rb->tail = (size_t *) (rb->head + 1); - rb->flags = (size_t *) (rb->tail + 1); + rb->flags = (size_t *) (rb->tail + 1); rb->mtx = (pthread_mutex_t *) (rb->flags + 1); rb->add = (pthread_cond_t *) (rb->mtx + 1); rb->del = rb->add + 1; 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; + rb->txq.target = 0; + rb->txq.rate = 0; + rb->txq.ns = 0; + rb->txq.limit = TXQ_INIT_SLOTS; + rb->txq.wr = 0; + rb->txq.due = TXQ_PERIOD_INIT; + rb->txq.period = TXQ_PERIOD_INIT; + rb->txq.q0 = 0; + rb->txq.idle = false; + rb->txq.measured = false; return rb; fail_truncate: close(fd); + if (flags & O_CREAT) shm_unlink(fn); fail_open: @@ -192,27 +190,27 @@ struct ssm_rbuff * ssm_rbuff_create(pid_t pid, if (rb == NULL) goto fail_rb; - if (pthread_mutexattr_init(&mattr)) + if (pthread_mutexattr_init(&mattr) != 0) goto fail_mattr; pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED); #ifdef HAVE_ROBUST_MUTEX pthread_mutexattr_setrobust(&mattr, PTHREAD_MUTEX_ROBUST); #endif - if (pthread_mutex_init(rb->mtx, &mattr)) + if (pthread_mutex_init(rb->mtx, &mattr) != 0) goto fail_mutex; - if (pthread_condattr_init(&cattr)) + if (pthread_condattr_init(&cattr) != 0) goto fail_cattr; pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_SHARED); #ifndef __APPLE__ pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK); #endif - if (pthread_cond_init(rb->add, &cattr)) + if (pthread_cond_init(rb->add, &cattr) != 0) goto fail_add; - if (pthread_cond_init(rb->del, &cattr)) + if (pthread_cond_init(rb->del, &cattr) != 0) goto fail_del; *rb->flags = RB_RDWR; @@ -264,12 +262,9 @@ void ssm_rbuff_close(struct ssm_rbuff * rb) { assert(rb); - /* - * Caller must set RB_FLOWDOWN first; if a user becomes - * cancellable, push a cleanup that decrements n_users. - */ - while (__atomic_load_n(&rb->n_users, __ATOMIC_SEQ_CST) > 0) { - struct timespec tic = { 0, 100000 }; + while (LOAD(&rb->n_users) > 0) { + struct timespec tic = TIMESPEC_INIT_US(100); + nanosleep(&tic, NULL); } @@ -282,103 +277,185 @@ static void __cleanup_rbuff_reader(void * o) struct ssm_rbuff * rb = (struct ssm_rbuff *) o; pthread_mutex_unlock(rb->mtx); - __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_SUB(&rb->n_users, 1); } -/* - * Refresh the drain-rate estimate and derived occupancy limit. - * Called with rb->mtx held, at most once per TXQ_SAMPLE_MASK writes. - */ +static bool txq_is_on(struct ssm_rbuff * rb) +{ + return LOAD_RELAXED(&rb->txq.target) != 0; +} + +/* Occupancy that holds the delay at the target; rate 0 gets the floor. */ +static size_t rbuff_txq_slots(uint64_t rate, + uint64_t target) +{ + uint64_t slots; + + slots = rate * target / BILLION; + if (slots < TXQ_MIN_SLOTS) + return TXQ_MIN_SLOTS; + + return slots > TXQ_UNLIMITED ? TXQ_UNLIMITED : (size_t) slots; +} + +/* Ceiling for one write (taking into account priority). */ +static size_t rbuff_txq_ceiling(struct ssm_rbuff * rb, + bool prio) +{ + size_t lim; + size_t max; + + if (!txq_is_on(rb)) + return TXQ_UNLIMITED; + + if (!rb->txq.measured) + lim = TXQ_INIT_SLOTS; + else + lim = LOAD_RELAXED(&rb->txq.limit); + + max = TXQ_DATA_MAX; + + if (prio) { + lim *= SSM_RBUFF_TXQ_PRIO_MUL; + max = TXQ_UNLIMITED; + } + + return lim > max ? max : lim; +} + +/* Opens a measurement window at now_ns. Caller holds rb->mtx. */ +static void rbuff_txq_anchor(struct ssm_rbuff * rb, + uint64_t now_ns, + size_t queued) +{ + rb->txq.ns = now_ns; + rb->txq.q0 = queued; + rb->txq.wr = 0; + rb->txq.due = rb->txq.period; + rb->txq.idle = false; +} + +/* Enough dequeues to resolve a rate? */ +static bool txq_is_blind(struct ssm_rbuff * rb, + int64_t drained, + int64_t dt_ns) +{ + int64_t target = (int64_t) LOAD_RELAXED(&rb->txq.target); + + if (drained * 2 >= (int64_t) rb->txq.period) + return false; + + return dt_ns < (target >> TXQ_SPW_SHIFT); +} + +/* Leaves the window open and retries a period later. */ +static void rbuff_txq_defer(struct ssm_rbuff * rb) +{ + rb->txq.due = rb->txq.wr + rb->txq.period; +} + +/* Aims the sample period at 1 << TXQ_SPW_SHIFT per target window. */ +static size_t rbuff_txq_retune(size_t period, + int64_t dt_ns, + int64_t target) +{ + if (dt_ns > (target >> TXQ_SPW_SHIFT)) { + period /= 2; + return period < TXQ_PERIOD_MIN ? TXQ_PERIOD_MIN : period; + } + + if (dt_ns < (target >> (TXQ_SPW_SHIFT + 1))) { + period *= 2; + return period > TXQ_PERIOD_MAX ? TXQ_PERIOD_MAX : period; + } + + return period; +} + +/* Only raise the estimate if the window ran empty. Call holding rb->mtx. */ 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 target; int64_t drained; - int64_t sample_rate; + int64_t sample; int64_t rate; - int64_t target; size_t limit; + size_t was; 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); + dt_ns = (int64_t) (now_ns - rb->txq.ns); + if (rb->txq.ns == 0 || dt_ns < 0) { + rbuff_txq_anchor(rb, now_ns, queued); return; } - dt_ns = (int64_t) (now_ns - last_ns); - if (dt_ns < TXQ_MIN_DT_NS) + if (dt_ns < TXQ_MIN_DT_NS) { + rbuff_txq_defer(rb); 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; + target = (int64_t) LOAD_RELAXED(&rb->txq.target); + drained = (int64_t) rb->txq.wr + (int64_t) rb->txq.q0 + - (int64_t) queued; + assert(drained >= 0); - sample_rate = drained * BILLION / dt_ns; + sample = drained * BILLION / dt_ns; + rate = (int64_t) rb->txq.rate; + if (sample > rate && txq_is_blind(rb, drained, dt_ns)) { + rbuff_txq_defer(rb); + return; + } - rate = LOAD_RELAXED(&rb->txq_rate); - rate += (sample_rate - rate) >> TXQ_SHIFT; - if (rate < 0) - rate = 0; + if (!rb->txq.measured) + rate = sample; + else if (rb->txq.idle && queued <= TXQ_MIN_SLOTS) + rate = sample > rate ? sample : rate; + else + rate = (rate * (TXQ_EWMA_N - 1) + sample) / TXQ_EWMA_N; - target = (int64_t) LOAD_RELAXED(&rb->txq_target); + if (rate > TXQ_MAX_RATE) + rate = TXQ_MAX_RATE; - limit = (size_t) (rate * target / BILLION); - if (limit < TXQ_MIN_SLOTS) - limit = TXQ_MIN_SLOTS; + limit = rbuff_txq_slots((uint64_t) rate, (uint64_t) target); + was = rbuff_txq_ceiling(rb, false); - if (limit > TXQ_UNLIMITED) - limit = TXQ_UNLIMITED; + rb->txq.rate = (uint64_t) rate; + rb->txq.period = rbuff_txq_retune(rb->txq.period, dt_ns, target); - 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); -} + STORE_RELAXED(&rb->txq.limit, limit); + STORE_RELAXED(&rb->txq.measured, true); -/* 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 (rbuff_txq_ceiling(rb, false) > was) + pthread_cond_broadcast(rb->del); - if ((wr & TXQ_SAMPLE_MASK) == 0) - rbuff_txq_sample(rb, QUEUED(rb)); + rbuff_txq_anchor(rb, now_ns, queued); } /* - * 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. + * Counts one enqueue. A prio write triggers no sample: it is the only + * traffic left in a stall, and would shrink the ceiling it needs. */ -static size_t rbuff_txq_prio_limit(struct ssm_rbuff * rb) +static void rbuff_txq_touch(struct ssm_rbuff * rb, + bool was_empty, + bool prio) { - size_t lim; + ++rb->txq.wr; - if (!TXQ_ON(rb)) - return TXQ_UNLIMITED; + if (was_empty) + rb->txq.idle = true; - lim = LOAD_RELAXED(&rb->txq_limit) * TXQ_PRIO_MUL; + if (prio) + return; - return lim > TXQ_UNLIMITED ? TXQ_UNLIMITED : lim; + if (rb->txq.wr >= rb->txq.due) + rbuff_txq_sample(rb, QUEUED(rb)); } /* prio outranks new data up to its own, higher, ceiling. */ @@ -392,14 +469,15 @@ static int rbuff_write_nb(struct ssm_rbuff * rb, assert(rb != NULL); - __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_ADD(&rb->n_users, 1); - flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); + flags = LOAD(rb->flags); if (flags != RB_RDWR) { if (flags & RB_FLOWDOWN) { ret = -EFLOWDOWN; goto fail_flags; } + if (!(flags & RB_WR)) { ret = -ENOTALLOC; goto fail_flags; @@ -408,8 +486,7 @@ static int rbuff_write_nb(struct ssm_rbuff * rb, robust_mutex_lock(rb->mtx); - if (QUEUED(rb) >= (prio ? rbuff_txq_prio_limit(rb) - : TXQ_LIMIT(rb))) { + if (QUEUED(rb) >= rbuff_txq_ceiling(rb, prio)) { ret = -EAGAIN; goto fail_mutex; } @@ -417,24 +494,25 @@ static int rbuff_write_nb(struct ssm_rbuff * rb, was_empty = IS_EMPTY(rb); HEAD(rb) = (ssize_t) off; + ADVANCE_HEAD(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); + if (txq_is_on(rb)) + rbuff_txq_touch(rb, was_empty, prio); pthread_mutex_unlock(rb->mtx); - __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_SUB(&rb->n_users, 1); + return 0; fail_mutex: pthread_mutex_unlock(rb->mtx); fail_flags: - __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_SUB(&rb->n_users, 1); return ret; } @@ -457,18 +535,20 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb, { size_t flags; int ret = 0; + int err; bool was_empty; assert(rb != NULL); - __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_ADD(&rb->n_users, 1); - flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); + flags = LOAD(rb->flags); if (flags != RB_RDWR) { if (flags & RB_FLOWDOWN) { ret = -EFLOWDOWN; goto fail_flags; } + if (!(flags & RB_WR)) { ret = -ENOTALLOC; goto fail_flags; @@ -479,32 +559,42 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb, pthread_cleanup_push(__cleanup_rbuff_reader, rb); - while (OVER_LIMIT(rb) && ret != -ETIMEDOUT) { - flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); + while (QUEUED(rb) >= rbuff_txq_ceiling(rb, false)) { + flags = LOAD(rb->flags); if (flags & RB_FLOWDOWN) { ret = -EFLOWDOWN; break; } - ret = -robust_wait(rb->del, rb->mtx, abstime); + + err = robust_wait(rb->del, rb->mtx, abstime); + if (err == EOWNERDEAD) + continue; + + if (err != 0) { + ret = -err; + break; + } } pthread_cleanup_pop(false); - if (ret != -ETIMEDOUT && ret != -EFLOWDOWN) { + if (ret == 0) { was_empty = IS_EMPTY(rb); HEAD(rb) = (ssize_t) off; + ADVANCE_HEAD(rb); + if (was_empty) pthread_cond_broadcast(rb->add); - if (TXQ_ON(rb)) - rbuff_txq_touch(rb); + if (txq_is_on(rb)) + rbuff_txq_touch(rb, was_empty, false); } pthread_mutex_unlock(rb->mtx); fail_flags: - __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_SUB(&rb->n_users, 1); return ret; } @@ -514,8 +604,7 @@ static int check_rb_flags(struct ssm_rbuff * rb) assert(rb != NULL); - flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); - + flags = LOAD(rb->flags); if (flags & RB_FLOWDOWN) return -EFLOWDOWN; @@ -534,7 +623,7 @@ ssize_t ssm_rbuff_read(struct ssm_rbuff * rb) assert(rb != NULL); - __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_ADD(&rb->n_users, 1); if (IS_EMPTY(rb)) { ret = check_rb_flags(rb); @@ -545,11 +634,13 @@ ssize_t ssm_rbuff_read(struct ssm_rbuff * rb) if (IS_EMPTY(rb)) { pthread_mutex_unlock(rb->mtx); + ret = check_rb_flags(rb); goto out; } ret = TAIL(rb); + ADVANCE_TAIL(rb); pthread_cond_broadcast(rb->del); @@ -557,7 +648,8 @@ ssize_t ssm_rbuff_read(struct ssm_rbuff * rb) pthread_mutex_unlock(rb->mtx); out: - __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_SUB(&rb->n_users, 1); + return ret; } @@ -569,9 +661,9 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb, assert(rb != NULL); - __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_ADD(&rb->n_users, 1); - flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); + flags = LOAD(rb->flags); if (IS_EMPTY(rb) && (flags & RB_FLOWDOWN)) { idx = -EFLOWDOWN; goto out; @@ -581,9 +673,13 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb, pthread_cleanup_push(__cleanup_rbuff_reader, rb); - while (IS_EMPTY(rb) && - idx != -ETIMEDOUT && - check_rb_flags(rb) == -EAGAIN) { + while (IS_EMPTY(rb)) { + if (idx == -ETIMEDOUT) + break; + + if (check_rb_flags(rb) != -EAGAIN) + break; + idx = -robust_wait(rb->add, rb->mtx, abstime); } @@ -591,6 +687,7 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb, if (!IS_EMPTY(rb)) { idx = TAIL(rb); + ADVANCE_TAIL(rb); pthread_cond_broadcast(rb->del); } else if (idx != -ETIMEDOUT) { @@ -602,31 +699,35 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb, assert(idx != -EAGAIN); out: - __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_SUB(&rb->n_users, 1); return idx; } -void ssm_rbuff_set_bits(struct ssm_rbuff * rb, - uint32_t bits) +void ssm_rbuff_set_flags(struct ssm_rbuff * rb, + uint32_t flags) { assert(rb != NULL); robust_mutex_lock(rb->mtx); - __atomic_fetch_or(rb->flags, (size_t) bits, __ATOMIC_SEQ_CST); + + FETCH_OR(rb->flags, (size_t) flags); pthread_cond_broadcast(rb->add); pthread_cond_broadcast(rb->del); + pthread_mutex_unlock(rb->mtx); } -void ssm_rbuff_clr_bits(struct ssm_rbuff * rb, - uint32_t bits) +void ssm_rbuff_clr_flags(struct ssm_rbuff * rb, + uint32_t flags) { assert(rb != NULL); robust_mutex_lock(rb->mtx); - __atomic_fetch_and(rb->flags, ~(size_t) bits, __ATOMIC_SEQ_CST); + + FETCH_AND(rb->flags, ~(size_t) flags); pthread_cond_broadcast(rb->add); pthread_cond_broadcast(rb->del); + pthread_mutex_unlock(rb->mtx); } @@ -634,7 +735,7 @@ uint32_t ssm_rbuff_get_flags(struct ssm_rbuff * rb) { assert(rb != NULL); - return (uint32_t) __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST); + return (uint32_t) LOAD(rb->flags); } /* Current occupancy limit; SSM_RBUFF_SIZE - 1 when unlimited. */ @@ -642,23 +743,40 @@ size_t ssm_rbuff_get_limit(struct ssm_rbuff * rb) { assert(rb != NULL); - return TXQ_LIMIT(rb); + return rbuff_txq_ceiling(rb, false); } -/* Target queueing delay; a zero target is unlimited. */ +/* Wakes up writers because target may have changed. */ void ssm_rbuff_set_txq_target(struct ssm_rbuff * rb, const struct timespec * ts) { + uint64_t target; + size_t limit; + assert(rb != NULL); assert(ts != NULL); + assert(ts->tv_sec >= 0); + assert(ts->tv_nsec >= 0); + assert(ts->tv_nsec < BILLION); + + target = TS_TO_UINT64(*ts); + + assert(target <= SSM_RBUFF_TXQ_MAX_DELAY); + + robust_mutex_lock(rb->mtx); - 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); + limit = rbuff_txq_slots(rb->txq.rate, target); - STORE_RELAXED(&rb->txq_target, TS_TO_UINT64(*ts)); + rb->txq.period = TXQ_PERIOD_INIT; + + rbuff_txq_anchor(rb, 0, QUEUED(rb)); + + STORE_RELAXED(&rb->txq.limit, limit); + STORE_RELAXED(&rb->txq.target, target); + + pthread_cond_broadcast(rb->del); + + pthread_mutex_unlock(rb->mtx); } /* Current target queueing delay for the tx occupancy limiter. */ @@ -668,14 +786,14 @@ void ssm_rbuff_get_txq_target(struct ssm_rbuff * rb, assert(rb != NULL); assert(ts != NULL); - UINT64_TO_TS(LOAD_RELAXED(&rb->txq_target), ts); + UINT64_TO_TS(LOAD_RELAXED(&rb->txq.target), ts); } void ssm_rbuff_fini(struct ssm_rbuff * rb) { assert(rb != NULL); - __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_ADD(&rb->n_users, 1); robust_mutex_lock(rb->mtx); @@ -688,7 +806,7 @@ void ssm_rbuff_fini(struct ssm_rbuff * rb) pthread_mutex_unlock(rb->mtx); - __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST); + FETCH_SUB(&rb->n_users, 1); } size_t ssm_rbuff_queued(struct ssm_rbuff * rb) diff --git a/src/lib/ssm/ssm.h.in b/src/lib/ssm/ssm.h.in index 57febae4..a17c8edd 100644 --- a/src/lib/ssm/ssm.h.in +++ b/src/lib/ssm/ssm.h.in @@ -39,6 +39,8 @@ #define SSM_FLOW_SET_PREFIX "@SSM_FLOW_SET_PREFIX@" #define SSM_POOL_NAME "@SSM_POOL_NAME@" #define SSM_RBUFF_SIZE @SSM_RBUFF_SIZE@ +#define SSM_RBUFF_TXQ_PRIO_MUL @SSM_RBUFF_TXQ_PRIO_MUL@ +#define SSM_RBUFF_TXQ_RESERVE @SSM_RBUFF_TXQ_RESERVE@ /* Packet buffer space reservation */ #define SSM_PK_BUFF_HEADSPACE @SSM_PK_BUFF_HEADSPACE@ diff --git a/src/lib/ssm/tests/rbuff_test.c b/src/lib/ssm/tests/rbuff_test.c index 57e6198e..b7ef3dfb 100644 --- a/src/lib/ssm/tests/rbuff_test.c +++ b/src/lib/ssm/tests/rbuff_test.c @@ -36,6 +36,7 @@ /* Mirrors TXQ_MIN_SLOTS in ssm/rbuff.c; keep in sync. */ #define FLOOR_SLOTS 4 +#define CEIL_SLOTS (SSM_RBUFF_SIZE - 1 - SSM_RBUFF_TXQ_RESERVE) #include <errno.h> #include <stdio.h> @@ -57,6 +58,7 @@ static int test_ssm_rbuff_create_destroy(void) ssm_rbuff_destroy(rb); TEST_SUCCESS(); + return TEST_RC_SUCCESS; fail: @@ -103,6 +105,7 @@ static int test_ssm_rbuff_write_read(void) ssm_rbuff_destroy(rb); TEST_SUCCESS(); + return TEST_RC_SUCCESS; fail_rb: @@ -134,6 +137,7 @@ static int test_ssm_rbuff_read_empty(void) ssm_rbuff_destroy(rb); TEST_SUCCESS(); + return TEST_RC_SUCCESS; fail_rb: @@ -163,6 +167,7 @@ static int test_ssm_rbuff_fill_drain(void) i, ssm_rbuff_queued(rb)); goto fail_rb; } + if (ssm_rbuff_write(rb, i) < 0) { printf("Failed to write at index %zu.\n", i); goto fail_rb; @@ -198,11 +203,13 @@ static int test_ssm_rbuff_fill_drain(void) 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(); @@ -228,7 +235,8 @@ static int test_ssm_rbuff_flags(void) goto fail_rb; } - ssm_rbuff_clr_bits(rb, RB_WR); + ssm_rbuff_clr_flags(rb, RB_WR); + flags = ssm_rbuff_get_flags(rb); if (flags != RB_RD) { printf("Expected RB_RD, got %u.\n", flags); @@ -240,7 +248,8 @@ static int test_ssm_rbuff_flags(void) goto fail_rb; } - ssm_rbuff_set_bits(rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(rb, RB_FLOWDOWN); + if (ssm_rbuff_write(rb, 1) != -EFLOWDOWN) { printf("Expected -EFLOWDOWN on FLOWDOWN.\n"); goto fail_rb; @@ -254,6 +263,7 @@ static int test_ssm_rbuff_flags(void) ssm_rbuff_destroy(rb); TEST_SUCCESS(); + return TEST_RC_SUCCESS; fail_rb: @@ -305,6 +315,7 @@ static int test_ssm_rbuff_open_close(void) ssm_rbuff_destroy(rb1); TEST_SUCCESS(); + return TEST_RC_SUCCESS; fail_rb2: @@ -351,8 +362,10 @@ static void * reader_thread(void * arg) val = ssm_rbuff_read(args->rb); while (val < 0) { nanosleep(&delay, NULL); + val = ssm_rbuff_read(args->rb); } + if (val != i) { printf("Expected %d, got %zd.\n", i, val); return (void *) -1; @@ -362,7 +375,7 @@ static void * reader_thread(void * arg) return NULL; } -static void * blocking_writer_thread(void * arg) +static void * blocking_wr_thread(void * arg) { struct thread_args * args = (struct thread_args *) arg; int i; @@ -375,7 +388,7 @@ static void * blocking_writer_thread(void * arg) return NULL; } -static void * blocking_reader_thread(void * arg) +static void * blocking_rd_thread(void * arg) { struct thread_args * args = (struct thread_args *) arg; int i; @@ -394,13 +407,13 @@ static void * blocking_reader_thread(void * arg) static int test_ssm_rbuff_blocking(void) { - struct ssm_rbuff * rb; - pthread_t wthread; - pthread_t rthread; - struct thread_args args; - struct timespec delay = {0, 10 * MILLION}; - void * ret_w; - void * ret_r; + struct ssm_rbuff * rb; + pthread_t wthread; + pthread_t rthread; + struct thread_args args; + struct timespec delay = {0, 10 * MILLION}; + void * ret_w; + void * ret_r; TEST_START(); @@ -413,15 +426,14 @@ static int test_ssm_rbuff_blocking(void) args.rb = rb; args.iterations = 50; args.delay_us = 0; - - if (pthread_create(&rthread, NULL, blocking_reader_thread, &args)) { + if (pthread_create(&rthread, NULL, blocking_rd_thread, &args) != 0) { printf("Failed to create reader thread.\n"); goto fail_rthread; } nanosleep(&delay, NULL); - if (pthread_create(&wthread, NULL, blocking_writer_thread, &args)) { + if (pthread_create(&wthread, NULL, blocking_wr_thread, &args) != 0) { printf("Failed to create writer thread.\n"); pthread_cancel(rthread); goto fail_wthread; @@ -438,6 +450,7 @@ static int test_ssm_rbuff_blocking(void) ssm_rbuff_destroy(rb); TEST_SUCCESS(); + return TEST_RC_SUCCESS; fail_ret: @@ -485,8 +498,7 @@ static int test_ssm_rbuff_blocking_timeout(void) (end.tv_nsec - start.tv_nsec) / 1000000L; if (elapsed_ms < 90 || elapsed_ms > 200) { - printf("Timeout took %ld ms, expected ~100 ms.\n", - elapsed_ms); + printf("Timeout took %ld ms, expected ~100 ms.\n", elapsed_ms); goto fail_rb; } @@ -505,8 +517,7 @@ static int test_ssm_rbuff_blocking_timeout(void) clock_gettime(PTHREAD_COND_CLOCK, &end); if (ret != -ETIMEDOUT) { - printf("Expected -ETIMEDOUT on full buffer, got %zd.\n", - ret); + printf("Expected -ETIMEDOUT on full buffer, got %zd.\n", ret); goto fail_rb; } @@ -525,11 +536,13 @@ static int test_ssm_rbuff_blocking_timeout(void) 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(); @@ -556,7 +569,7 @@ static int test_ssm_rbuff_blocking_flowdown(void) clock_gettime(PTHREAD_COND_CLOCK, &now); ts_add(&now, &interval, &abs_timeout); - ssm_rbuff_set_bits(rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(rb, RB_FLOWDOWN); ret = ssm_rbuff_read_b(rb, &abs_timeout); if (ret != -EFLOWDOWN) { @@ -564,7 +577,7 @@ static int test_ssm_rbuff_blocking_flowdown(void) goto fail_rb; } - ssm_rbuff_clr_bits(rb, RB_FLOWDOWN); + ssm_rbuff_clr_flags(rb, RB_FLOWDOWN); for (i = 0; i < SSM_RBUFF_SIZE - 1; ++i) { if (ssm_rbuff_write(rb, i) < 0) { @@ -576,7 +589,7 @@ static int test_ssm_rbuff_blocking_flowdown(void) clock_gettime(PTHREAD_COND_CLOCK, &now); ts_add(&now, &interval, &abs_timeout); - ssm_rbuff_set_bits(rb, RB_FLOWDOWN); + ssm_rbuff_set_flags(rb, RB_FLOWDOWN); ret = ssm_rbuff_write_b(rb, 999, &abs_timeout); if (ret != -EFLOWDOWN) { @@ -584,18 +597,21 @@ static int test_ssm_rbuff_blocking_flowdown(void) goto fail_rb; } - ssm_rbuff_clr_bits(rb, RB_FLOWDOWN); + ssm_rbuff_clr_flags(rb, RB_FLOWDOWN); + 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(); @@ -604,12 +620,12 @@ static int test_ssm_rbuff_blocking_flowdown(void) static int test_ssm_rbuff_threaded(void) { - struct ssm_rbuff * rb; - pthread_t wthread; - pthread_t rthread; - struct thread_args args; - void * ret_w; - void * ret_r; + struct ssm_rbuff * rb; + pthread_t wthread; + pthread_t rthread; + struct thread_args args; + void * ret_w; + void * ret_r; TEST_START(); @@ -622,13 +638,12 @@ static int test_ssm_rbuff_threaded(void) args.rb = rb; args.iterations = 100; args.delay_us = 100; - - if (pthread_create(&wthread, NULL, writer_thread, &args)) { + if (pthread_create(&wthread, NULL, writer_thread, &args) != 0) { printf("Failed to create writer thread.\n"); goto fail_rb; } - if (pthread_create(&rthread, NULL, reader_thread, &args)) { + if (pthread_create(&rthread, NULL, reader_thread, &args) != 0) { printf("Failed to create reader thread.\n"); pthread_cancel(wthread); pthread_join(wthread, NULL); @@ -646,6 +661,7 @@ static int test_ssm_rbuff_threaded(void) ssm_rbuff_destroy(rb); TEST_SUCCESS(); + return TEST_RC_SUCCESS; fail_rb: @@ -708,7 +724,7 @@ static int test_ssm_rbuff_limit_off(void) static int test_ssm_rbuff_limit_slow(void) { struct ssm_rbuff * rb; - struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec dfl = TIMESPEC_INIT_MS(SSM_RBUFF_TXQ_DELAY); struct timespec delay = {0, 10 * MILLION}; size_t limit; size_t i; @@ -761,7 +777,7 @@ static int test_ssm_rbuff_limit_slow(void) static int test_ssm_rbuff_limit_fast(void) { struct ssm_rbuff * rb; - struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec dfl = TIMESPEC_INIT_MS(SSM_RBUFF_TXQ_DELAY); size_t limit; size_t i; @@ -788,9 +804,8 @@ static int test_ssm_rbuff_limit_fast(void) } limit = ssm_rbuff_get_limit(rb); - if (limit != SSM_RBUFF_SIZE - 1) { - printf("Expected limit %d, got %zu.\n", - SSM_RBUFF_SIZE - 1, limit); + if (limit != CEIL_SLOTS) { + printf("Expected limit %d, got %zu.\n", CEIL_SLOTS, limit); goto fail_rb; } @@ -813,7 +828,7 @@ static int test_ssm_rbuff_limit_fast(void) static int test_ssm_rbuff_limit_floor(void) { struct ssm_rbuff * rb; - struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec dfl = TIMESPEC_INIT_MS(SSM_RBUFF_TXQ_DELAY); struct timespec interval = {0, 50 * MILLION}; struct timespec now; struct timespec abs_timeout; @@ -875,10 +890,11 @@ static int test_ssm_rbuff_limit_floor(void) return TEST_RC_FAIL; } +/* A fresh ring is unlimited; rx rings must not inherit a bound. */ static int test_ssm_rbuff_txq_target(void) { struct ssm_rbuff * rb; - struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec dfl = TIMESPEC_INIT_MS(SSM_RBUFF_TXQ_DELAY); struct timespec delay = {0, 5 * MILLION}; struct timespec small = {0, 2 * MILLION}; struct timespec big = {0, 200 * MILLION}; @@ -896,8 +912,8 @@ static int test_ssm_rbuff_txq_target(void) 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; @@ -969,10 +985,11 @@ static int test_ssm_rbuff_txq_target(void) return TEST_RC_FAIL; } +/* Ages the seed sample past the estimator's dt floor at write 16. */ static int test_ssm_rbuff_write_over_limit(void) { struct ssm_rbuff * rb; - struct timespec dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION}; + struct timespec dfl = TIMESPEC_INIT_MS(SSM_RBUFF_TXQ_DELAY); struct timespec age = {0, 20 * 1000}; size_t count; int ret = 0; @@ -997,7 +1014,6 @@ static int test_ssm_rbuff_write_over_limit(void) goto fail_rb; } - /* Age the seed sample past the estimator's dt floor. */ if (count == 16) nanosleep(&age, NULL); } |