From f921e50952334d99b6c25ee8df09e7fb1523e92f Mon Sep 17 00:00:00 2001 From: Dimitri Staessens Date: Sun, 16 Aug 2026 19:46:34 +0000 Subject: lib: Update FRCT loss recovery Some more stability fixes in FRCT. Signed-off-by: Dimitri Staessens Signed-off-by: Sander Vrijders --- src/lib/dev.c | 219 ++++++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 190 insertions(+), 29 deletions(-) (limited to 'src/lib/dev.c') 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, -- cgit v1.2.3