summaryrefslogtreecommitdiff
path: root/src/lib/dev.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/lib/dev.c')
-rw-r--r--src/lib/dev.c867
1 files changed, 746 insertions, 121 deletions
diff --git a/src/lib/dev.c b/src/lib/dev.c
index ae0401b7..3b0363da 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -27,7 +27,11 @@
#endif
#include "config.h"
+#include "cap.h"
#include "ssm.h"
+#include "poa/poa.h"
+
+#define OUROBOROS_PREFIX "libouroboros"
#include <ouroboros/atomics.h>
#include <ouroboros/bitmap.h>
@@ -45,6 +49,7 @@
#include <ouroboros/ipcp-dev.h>
#include <ouroboros/list.h>
#include <ouroboros/local-dev.h>
+#include <ouroboros/logs.h>
#include <ouroboros/np1_flow.h>
#include <ouroboros/pthread.h>
#include <ouroboros/random.h>
@@ -56,6 +61,7 @@
#include <ouroboros/ssm_flow_set.h>
#include <ouroboros/ssm_pool.h>
#include <ouroboros/ssm_rbuff.h>
+#include <ouroboros/time.h>
#include <ouroboros/tw.h>
#include <ouroboros/utils.h>
@@ -63,8 +69,10 @@
#ifdef HAVE_LIBGCRYPT
#include <gcrypt.h>
#endif
+#include <arpa/inet.h>
#include <stdarg.h>
#include <stdbool.h>
+#include <stddef.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
@@ -79,6 +87,7 @@
#define DONE_PART -2
#define CRCLEN (sizeof(uint32_t))
+#define FLOW_AVG_SHIFT 3
#define SECMEMSZ 16384
#define MSGBUFSZ 2048
@@ -98,8 +107,14 @@ struct flow {
ssize_t part_idx;
struct crypt_ctx * crypt;
- int headsz; /* IV */
- int tailsz; /* Tag + CRC */
+ int headsz; /* Selector */
+ int tailsz; /* Tag + CRC */
+
+ struct timespec rk_grace; /* TX-promote deadline */
+ struct timespec rk_attempt; /* Last re-key attempt */
+ bool rk_wm_inflight; /* Re-key trigger in flight */
+ uint32_t rk_wm_ctr; /* Throttles the consult */
+ bool rk_initiator; /* OAP initiator this re-key */
struct timespec snd_act;
struct timespec rcv_act;
@@ -110,6 +125,14 @@ struct flow {
struct timespec rcv_timeo;
struct frcti * frcti;
+
+ /* Mean written packet size (bytes), EWMA over the send path. */
+ size_t mean_len;
+
+ struct poa_flow * poa; /* NULL for shared memory flows */
+
+ /* Egress capacity estimator; armed by the IPCP, else NULL. */
+ struct cap_est * cap;
};
struct flow_set {
@@ -133,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;
@@ -261,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)
@@ -278,7 +303,7 @@ static int spb_encrypt(struct flow * flow,
return 0;
fail_alloc:
freebuf(out);
- fail_encrypt:
+
return -ECRYPT;
}
@@ -296,8 +321,7 @@ static int spb_decrypt(struct flow * flow,
in.len = ssm_pk_buff_len(spb);
if (crypt_decrypt(flow->crypt, in, &out) < 0)
- return -ENOMEM;
-
+ return -ECRYPT;
head = ssm_pk_buff_pop(spb, flow->headsz) + flow->headsz;
ssm_pk_buff_pop_tail(spb, flow->tailsz);
@@ -342,8 +366,14 @@ static int crc_check(struct ssm_pk_buff * spb,
size_t head_skip)
{
uint32_t crc;
- uint8_t * head = ssm_pk_buff_head(spb) + head_skip;
- uint8_t * tail = ssm_pk_buff_pop_tail(spb, CRCLEN);
+ uint8_t * head;
+ uint8_t * tail;
+
+ if (ssm_pk_buff_len(spb) < head_skip + CRCLEN)
+ return 1;
+
+ head = ssm_pk_buff_head(spb) + head_skip;
+ tail = ssm_pk_buff_pop_tail(spb, CRCLEN);
mem_hash(HASH_CRC32, &crc, head, tail - head);
@@ -353,10 +383,7 @@ static int crc_check(struct ssm_pk_buff * spb,
/* FRCT included here so it can use proc and dev.c statics directly. */
#include "frct.c"
-/*
- * SACK / DATA carry trailer CRC32; HCS protects the headers on every
- * FRCT packet. Decrypt before any check so plaintext is authoritative.
- */
+/* Decrypt before any check so the plaintext is authoritative. */
static bool invalid_pkt(struct flow * flow,
struct ssm_pk_buff * spb)
{
@@ -438,10 +465,6 @@ static void compute_wait_deadline(const struct timespec * dl,
*out = *dl;
}
-/*
- * proc.lock rdlock held across each iteration so flow_fini's wrlock
- * waits for us to finish; FLOWDOWN already set means we exit promptly.
- */
static void flow_drain_rx_nb(struct flow * flow)
{
ssize_t idx;
@@ -509,10 +532,80 @@ static void flow_drain_rx_nb(struct flow * flow)
}
}
-/*
- * Wait clamped by caller deadline, next tw expiry, and TICTIME;
- * a clamp-timeout means tw work is due, not caller-deadline.
- */
+/* TX-promotion grace when the peer's install latency is unknown (raw). */
+#define REKEY_GRACE_MS 1000
+
+/* Last-resort promote within N node-keys of exhaustion (< watermark). */
+#define REKEY_PROMOTE_FLOOR 1
+
+/* Throttle re-key retries so a failed attempt can't storm the IRMd. */
+#define REKEY_BACKOFF_NS (250 * MILLION)
+
+/* proc.lock (rd) only guards teardown; crypt_rekey self-synchronises. */
+static void flow_rekey(struct flow * flow)
+{
+ struct flow_info info;
+ struct crypt_sk sk;
+ struct timespec now;
+ struct timespec intv;
+ time_t ms;
+ uint8_t key[SYMMKEYSZ];
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {SOCK_BUF_SIZE, buf};
+ bool has_key;
+ bool initiator = false;
+
+ pthread_rwlock_rdlock(&proc.lock);
+ if (flow->info.id < 0 || flow->crypt == NULL) {
+ pthread_rwlock_unlock(&proc.lock);
+ return;
+ }
+
+ /* Back off so a failed attempt can't storm the IRMd per syscall. */
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+ if (ts_diff_ns(&now, &flow->rk_attempt) < REKEY_BACKOFF_NS) {
+ pthread_rwlock_unlock(&proc.lock);
+ return;
+ }
+
+ flow->rk_attempt = now;
+ info = flow->info;
+ pthread_rwlock_unlock(&proc.lock);
+
+ if (flow_update__irm_req_ser(&msg, &info, false) < 0)
+ return;
+
+ if (send_recv_msg(&msg) < 0)
+ return;
+
+ sk.key = key;
+ if (flow_rekey__irm_result_des(&msg, &sk, &has_key, &initiator) < 0)
+ return;
+
+ if (!has_key)
+ return;
+
+ pthread_rwlock_rdlock(&proc.lock);
+ if (flow->info.id == info.id && flow->crypt != NULL) {
+ if (crypt_rekey(flow->crypt, &sk) == 0) {
+ flow->rk_initiator = initiator;
+ /* Hold TX on the old epoch until the peer installs. */
+ ms = flow->info.mpl > 0 ? flow->info.mpl * 3
+ : REKEY_GRACE_MS;
+ intv.tv_sec = ms / 1000;
+ intv.tv_nsec = (ms % 1000) * MILLION;
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+ ts_add(&now, &intv, &flow->rk_grace);
+ }
+ /* Re-arm the watermark even if the install was a no-op. */
+ STORE_RELAXED(&flow->rk_wm_inflight, false);
+ }
+ pthread_rwlock_unlock(&proc.lock);
+
+ crypt_secure_clear(key, SYMMKEYSZ);
+}
+
+/* A clamp-timeout means tw work is due, not the caller deadline. */
static int flow_rx_one(struct flow * flow,
struct timespec * abs)
{
@@ -533,7 +626,20 @@ static int flow_rx_one(struct flow * flow,
return -EFLOWDOWN;
}
+ /* Pull a parked re-key before re-blocking (idle reader). */
+ if (flow->crypt != NULL
+ && (ssm_rbuff_get_flags(rx_rb) & RB_REKEY)) {
+ pthread_rwlock_unlock(&proc.lock);
+ flow_rekey(flow);
+ continue;
+ }
+
+ pthread_cleanup_push(__cleanup_rwlock_unlock, &proc.lock);
+
idx = ssm_rbuff_read_b(rx_rb, &wait_abs);
+
+ pthread_cleanup_pop(false);
+
if (idx == -ETIMEDOUT) {
pthread_rwlock_unlock(&proc.lock);
if (deadline_passed(abs))
@@ -592,27 +698,26 @@ static void flow_clear(int fd)
proc.flows[fd].info.id = -1;
}
-/*
- * Set ACL_FLOWDOWN on rx/tx so any in-flight blocking reads or writes
- * wake up and drop their proc.lock rdlock. Must run BEFORE flow_fini's
- * wrlock, else the wrlock blocks on those rdlock holders and the
- * in-flight calls never see the FLOWDOWN signal.
- */
+/* Order before flow_fini's wrlock, which blocks on rdlock holders. */
static void flow_quiesce(int fd)
{
struct ssm_rbuff * rx_rb = proc.flows[fd].rx_rb;
struct ssm_rbuff * tx_rb = proc.flows[fd].tx_rb;
if (rx_rb != NULL)
- ssm_rbuff_set_acl(rx_rb, ACL_FLOWDOWN);
+ ssm_rbuff_set_flags(rx_rb, RB_FLOWDOWN);
+
if (tx_rb != NULL)
- ssm_rbuff_set_acl(tx_rb, ACL_FLOWDOWN);
+ ssm_rbuff_set_flags(tx_rb, RB_FLOWDOWN);
}
static void do_flow_fini(int fd)
{
assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+ if (proc.flows[fd].poa != NULL)
+ poa_flow_detach(proc.flows[fd].poa);
+
if (proc.flows[fd].frcti != NULL)
frcti_destroy(proc.flows[fd].frcti);
@@ -636,6 +741,8 @@ static void do_flow_fini(int fd)
crypt_destroy_ctx(proc.flows[fd].crypt);
+ free(proc.flows[fd].cap);
+
flow_clear(fd);
}
@@ -667,14 +774,21 @@ static __inline__ size_t flow_user_mtu(const struct flow * flow,
return raw > hdr ? raw - hdr : 0;
}
+/* A PoA flow transmits on its own socket; it has no tx ring. */
static int flow_init(struct flow_info * info,
struct crypt_sk * sk,
- time_t rtt_hint)
+ time_t rtt_hint,
+ struct poa_flow * pf)
{
- struct timespec now;
- 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);
@@ -694,13 +808,17 @@ static int flow_init(struct flow_info * info,
if (flow->rx_rb == NULL)
goto fail_rx_rb;
- flow->tx_rb = ssm_rbuff_open(info->n_1_pid, info->id);
- if (flow->tx_rb == NULL)
- goto fail_tx_rb;
+ if (pf == NULL) {
+ flow->tx_rb = ssm_rbuff_open(info->n_1_pid, info->id);
+ if (flow->tx_rb == NULL)
+ goto fail_tx_rb;
- flow->set = ssm_flow_set_open(info->n_1_pid);
- if (flow->set == NULL)
- goto fail_set;
+ tx_rb = flow->tx_rb;
+
+ flow->set = ssm_flow_set_open(info->n_1_pid);
+ if (flow->set == NULL)
+ goto fail_set;
+ }
flow->oflags = FLOWFDEFAULT;
flow->part_idx = NO_PART;
@@ -709,13 +827,15 @@ static int flow_init(struct flow_info * info,
flow->crypt = NULL;
flow->headsz = 0;
flow->tailsz = 0;
+ flow->poa = pf;
if (IS_ENCRYPTED(sk)) {
- sk->rot_bit = KEY_ROTATION_BIT;
flow->crypt = crypt_create_ctx(sk);
- if (flow->crypt == NULL)
+ if (flow->crypt == NULL) {
+ err = -ECRYPT;
goto fail_crypt;
- flow->headsz = crypt_get_ivsz(flow->crypt);
+ }
+ flow->headsz = crypt_get_headsz(flow->crypt);
flow->tailsz = crypt_get_tagsz(flow->crypt);
}
@@ -725,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;
@@ -733,18 +853,30 @@ static int flow_init(struct flow_info * info,
proc.id_to_fd[info->id].fd = fd;
+ if (pf != NULL)
+ poa_flow_attach(pf, info->id, flow->rx_rb);
+
flow_set_state(&proc.id_to_fd[info->id], FLOW_ALLOCATED);
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:
crypt_destroy_ctx(flow->crypt);
fail_crypt:
- ssm_flow_set_close(flow->set);
+ if (flow->set != NULL)
+ ssm_flow_set_close(flow->set);
fail_set:
- ssm_rbuff_close(flow->tx_rb);
+ if (flow->tx_rb != NULL)
+ ssm_rbuff_close(flow->tx_rb);
fail_tx_rb:
ssm_rbuff_close(flow->rx_rb);
fail_rx_rb:
@@ -768,9 +900,10 @@ static void init(int argc,
char ** argv,
char ** envp)
{
- struct proc_info info;
- char * prog = argv[0];
- int i;
+ struct proc_info info;
+ char * prog = argv[0];
+ int i;
+ pthread_rwlockattr_t attr;
#ifdef PROC_FLOW_STATS
char procstr[32];
#endif
@@ -855,7 +988,16 @@ static void init(int argc,
goto fail_cond;
}
- if (pthread_rwlock_init(&proc.lock, NULL) < 0) {
+ /* Writer-preferred: FRCT readers must not starve flow accept. */
+ if (pthread_rwlockattr_init(&attr) != 0) {
+ fprintf(stderr, "FATAL: Could not init rwlock attributes.\n");
+ goto fail_rwlock_attr;
+ }
+#if defined(__GLIBC__)
+ pthread_rwlockattr_setkind_np(
+ &attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
+#endif
+ if (pthread_rwlock_init(&proc.lock, &attr) != 0) {
fprintf(stderr, "FATAL: Could not initialize flow lock.\n");
goto fail_flow_lock;
}
@@ -885,6 +1027,8 @@ static void init(int argc,
}
}
#endif
+ pthread_rwlockattr_destroy(&attr);
+
return;
#if defined PROC_FLOW_STATS
@@ -898,6 +1042,8 @@ static void init(int argc,
fail_fqset:
pthread_rwlock_destroy(&proc.lock);
fail_flow_lock:
+ pthread_rwlockattr_destroy(&attr);
+ fail_rwlock_attr:
pthread_cond_destroy(&proc.cond);
fail_cond:
pthread_mutex_destroy(&proc.mtx);
@@ -981,14 +1127,20 @@ static void fini(void)
__attribute__((section(INIT_SECTION))) __typeof__(init) * __init = init;
__attribute__((section(FINI_SECTION))) __typeof__(fini) * __fini = fini;
+/*
+ * A PoA flow is announced by its peer before the accept; from the
+ * reply on, the peer may transmit, so the flow must be able to
+ * receive.
+ */
int flow_accept(qosspec_t * qs,
const struct timespec * timeo)
{
struct flow_info flow;
- struct crypt_sk crypt;
+ struct crypt_sk crypt;
uint8_t buf[SOCK_BUF_SIZE];
buffer_t msg = {SOCK_BUF_SIZE, buf};
uint8_t key[SYMMKEYSZ];
+ struct poa_flow * pf;
int fd;
int err;
@@ -1012,14 +1164,21 @@ int flow_accept(qosspec_t * qs,
if (err < 0)
return err;
- crypt.key = key;
+ crypt.key = key;
+ crypt.epoch = 0;
+ crypt.role = CRYPT_ROLE_RESP;
err = flow__irm_result_des(&msg, &flow, &crypt);
if (err < 0)
return err;
- /* No RTT in accept; rtt_hint=0 bootstraps from first ACK. */
- fd = flow_init(&flow, &crypt, 0);
+ pf = poa_flow_take_pending(flow.id);
+
+ fd = flow_init(&flow, &crypt, 0, pf);
+ if (fd >= 0)
+ poa_flow_ready(pf);
+ else if (pf != NULL)
+ poa_flow_detach(pf);
crypt_secure_clear(key, SYMMKEYSZ);
@@ -1067,13 +1226,15 @@ int flow_alloc(const char * dst,
clock_gettime(PTHREAD_COND_CLOCK, &t1);
- crypt.key = key;
+ crypt.key = key;
+ crypt.epoch = 0;
+ crypt.role = CRYPT_ROLE_INIT;
err = flow__irm_result_des(&msg, &flow, &crypt);
if (err < 0)
return err;
- fd = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0));
+ fd = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0), NULL);
crypt_secure_clear(key, SYMMKEYSZ);
@@ -1106,13 +1267,15 @@ int flow_join(const char * dst,
if (err < 0)
return err;
- crypt.key = key;
+ crypt.key = key;
+ crypt.epoch = 0;
+ crypt.role = CRYPT_ROLE_INIT;
err = flow__irm_result_des(&msg, &flow, &crypt);
if (err < 0)
return err;
- fd = flow_init(&flow, &crypt, 0);
+ fd = flow_init(&flow, &crypt, 0, NULL);
crypt_secure_clear(key, SYMMKEYSZ);
@@ -1152,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);
@@ -1173,7 +1338,8 @@ int flow_dealloc(int fd)
pthread_cleanup_push(__cleanup_rwlock_unlock, &proc.lock);
- ssm_rbuff_fini(flow->tx_rb);
+ if (flow->tx_rb != NULL)
+ ssm_rbuff_fini(flow->tx_rb);
pthread_cleanup_pop(true);
@@ -1241,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,
...)
@@ -1251,8 +1429,6 @@ int fccntl(int fd,
va_list l;
struct timespec * timeo;
qosspec_t * qs;
- uint32_t rx_acl;
- uint32_t tx_acl;
size_t * qlen;
struct flow * flow;
uint16_t old_acc;
@@ -1265,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;
@@ -1328,7 +1505,11 @@ int fccntl(int fd,
break;
case FLOWGTXQLEN:
qlen = va_arg(l, size_t *);
- *qlen = ssm_rbuff_queued(flow->tx_rb);
+
+ if (flow->poa != NULL)
+ *qlen = poa_flow_qpkts(flow->poa);
+ else
+ *qlen = ssm_rbuff_queued(flow->tx_rb);
break;
case FLOWGMTU:
maxp = va_arg(l, size_t *);
@@ -1336,6 +1517,28 @@ int fccntl(int fd,
goto einval;
*maxp = flow_user_mtu(flow, flow->info.mtu);
break;
+ case FLOWSTXQDLY:
+ timeo = va_arg(l, struct timespec *);
+ if (timeo == NULL)
+ goto einval;
+
+ if (flow->tx_rb == NULL)
+ goto eperm;
+
+ if (!delay_is_valid(timeo))
+ goto einval;
+
+ set_txq = true;
+ break;
+ case FLOWGTXQDLY:
+ timeo = va_arg(l, struct timespec *);
+ if (timeo == NULL)
+ goto einval;
+
+ if (flow->tx_rb == NULL)
+ goto eperm;
+ ssm_rbuff_get_txq_target(flow->tx_rb, timeo);
+ break;
case FLOWSFLAGS:
old_acc = flow->oflags & FLOWFACCMODE;
flow->oflags = va_arg(l, uint32_t);
@@ -1348,31 +1551,28 @@ int fccntl(int fd,
&& flow->frcti != NULL)
emit_eos = true;
- rx_acl = ssm_rbuff_get_acl(flow->rx_rb);
- tx_acl = ssm_rbuff_get_acl(flow->tx_rb);
- /* Our flow write-only -> peer's read-only. */
+ /* Our flow write-only -> peer's read-only; restore on RDWR. */
if (flow->oflags & FLOWFWRONLY)
- rx_acl |= ACL_RDONLY;
- if (flow->oflags & FLOWFRDWR)
- rx_acl |= ACL_RDWR;
+ ssm_rbuff_clr_flags(flow->rx_rb, RB_WR);
+ else
+ ssm_rbuff_set_flags(flow->rx_rb, RB_WR);
if (flow->oflags & FLOWFDOWN) {
- rx_acl |= ACL_FLOWDOWN;
- tx_acl |= ACL_FLOWDOWN;
- ssm_flow_set_notify(flow->set,
- flow->info.id,
- FLOW_DOWN);
+ ssm_rbuff_set_flags(flow->rx_rb, RB_FLOWDOWN);
+ if (flow->tx_rb != NULL)
+ 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 {
- rx_acl &= ~ACL_FLOWDOWN;
- tx_acl &= ~ACL_FLOWDOWN;
- ssm_flow_set_notify(flow->set,
- flow->info.id,
- FLOW_UP);
+ ssm_rbuff_clr_flags(flow->rx_rb, RB_FLOWDOWN);
+ if (flow->tx_rb != NULL)
+ ssm_rbuff_clr_flags(flow->tx_rb, RB_FLOWDOWN);
+ if (flow->set != NULL)
+ ssm_flow_set_notify(flow->set, flow->info.id,
+ FLOW_UP);
}
- ssm_rbuff_set_acl(flow->rx_rb, rx_acl);
- ssm_rbuff_set_acl(flow->tx_rb, tx_acl);
-
break;
case FLOWGFLAGS:
fflags = va_arg(l, uint32_t *);
@@ -1459,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;
@@ -1473,6 +1676,25 @@ int fccntl(int fd,
return -EPERM;
}
+/*
+ * The ring counts slots, so the queue is only bytes if we know what a
+ * packet weighs. Ordered so the unsigned arithmetic cannot wrap.
+ */
+static void flow_mean_len_update(struct flow * flow,
+ size_t len)
+{
+ size_t avg = LOAD_RELAXED(&flow->mean_len);
+
+ if (avg == 0) {
+ STORE_RELAXED(&flow->mean_len, len);
+ return;
+ }
+
+ avg = avg + (len >> FLOW_AVG_SHIFT) - (avg >> FLOW_AVG_SHIFT);
+
+ STORE_RELAXED(&flow->mean_len, avg == 0 ? 1 : avg);
+}
+
static int flow_tx_spb(struct flow * flow,
struct ssm_pk_buff * spb,
uint16_t flags,
@@ -1504,21 +1726,23 @@ static int flow_tx_spb(struct flow * flow,
goto enomem;
}
+ if (flow->poa != NULL)
+ return poa_flow_tx(flow->poa, spb, block, abstime);
+
+ flow_mean_len_update(flow, ssm_pk_buff_len(spb));
+
if (!block)
ret = ssm_rbuff_write(flow->tx_rb, idx);
else
ret = ssm_rbuff_write_b(flow->tx_rb, idx, abstime);
- if (ret < 0) {
- ssm_pool_remove(proc.pool, idx);
+ if (ret < 0)
return ret;
- }
ssm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT);
return 0;
enomem:
- ssm_pool_remove(proc.pool, idx);
return -ENOMEM;
}
@@ -1527,20 +1751,16 @@ static __inline__ uint16_t flow_frag_role(size_t i, size_t n)
{
if (n == 1)
return FRCT_FR_SOLE;
+
if (i == 0)
return FRCT_FR_FIRST;
+
if (i + 1 == n)
return FRCT_FR_LAST;
return FRCT_FR_MID;
}
-/*
- * Stream-mode write: split buf into chunks of
- * (frag_mtu - PCI - PCI_STREAM) bytes; each chunk goes through the
- * normal tx path. frcti_snd injects the [start,end) extension and
- * advances snd_byte_next under its wrlock. No FFGM/LFGM role bits.
- */
static ssize_t flow_write_stream(struct flow * flow,
const void * buf,
size_t count,
@@ -1581,8 +1801,10 @@ static ssize_t flow_write_stream(struct flow * flow,
memcpy(ptr, src + off, clen);
ret = flow_tx_spb(flow, spb, 0, block, dl);
- if (ret < 0)
+ if (ret < 0) {
+ ssm_pool_remove(proc.pool, idx);
return off > 0 ? (ssize_t) off : (ssize_t) ret;
+ }
off += clen;
}
@@ -1614,6 +1836,7 @@ static ssize_t flow_write_frag(struct flow * flow,
/* Guard the ceil-divide against size_t overflow. */
if (count > SIZE_MAX - frag_payload + 1)
return -EMSGSIZE;
+
n = (count + frag_payload - 1) / frag_payload;
/* SDU larger than the FC window can ever offer would deadlock. */
@@ -1648,9 +1871,9 @@ static ssize_t flow_write_frag(struct flow * flow,
memcpy(ptr, src + off, clen);
- ret = flow_tx_spb(flow, spb, flow_frag_role(i, n),
- block, dl);
+ ret = flow_tx_spb(flow, spb, flow_frag_role(i, n), block, dl);
if (ret < 0) {
+ ssm_pool_remove(proc.pool, idx);
if (off > 0)
STAT_BUMP(flow->frcti, sdu_snd_tx);
return off > 0 ? (ssize_t) off : (ssize_t) ret;
@@ -1662,6 +1885,91 @@ static ssize_t flow_write_frag(struct flow * flow,
return (ssize_t) count;
}
+/*
+ * Initiator promotes on the install grace (it holds the key-confirm
+ * 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)
+{
+ struct timespec now;
+ int nodes_left;
+ bool promote;
+
+ if (flow->crypt == NULL)
+ return;
+
+ if (flow->rk_grace.tv_sec == 0 && flow->rk_grace.tv_nsec == 0)
+ return;
+
+ promote = crypt_peer_synced(flow->crypt);
+
+ if (!promote && flow->rk_initiator) {
+ clock_gettime(PTHREAD_COND_CLOCK, &now);
+ promote = ts_diff_ns(&now, &flow->rk_grace) >= 0;
+ }
+
+ if (!promote) {
+ nodes_left = crypt_nodes_left(flow->crypt);
+ promote = nodes_left >= 0 && nodes_left <= REKEY_PROMOTE_FLOOR;
+ }
+
+ if (!promote)
+ return;
+
+ crypt_tx_promote(flow->crypt);
+ flow->rk_grace.tv_sec = 0;
+ flow->rk_grace.tv_nsec = 0;
+}
+
+/* The reply carries no key; the seed arrives later over RB_REKEY. */
+static int flow_rekey_trigger(struct flow * flow)
+{
+ struct flow_info info;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {SOCK_BUF_SIZE, buf};
+
+ pthread_rwlock_rdlock(&proc.lock);
+ if (flow->info.id < 0 || flow->crypt == NULL) {
+ pthread_rwlock_unlock(&proc.lock);
+ return -1;
+ }
+ info = flow->info;
+ pthread_rwlock_unlock(&proc.lock);
+
+ if (flow_update__irm_req_ser(&msg, &info, true) < 0)
+ return -1;
+
+ if (send_recv_msg(&msg) < 0)
+ return -1;
+
+ return 0;
+}
+
+static bool flow_wm_due(struct flow * flow)
+{
+ uint32_t tick;
+
+ if (KEY_REKEY_WATERMARK == 0)
+ return false;
+
+ if (flow->crypt == NULL)
+ return false;
+
+ if (LOAD_RELAXED(&flow->rk_wm_inflight))
+ return false;
+
+ tick = FETCH_ADD_RELAXED(&flow->rk_wm_ctr, 1);
+ if ((tick & (FLOW_WM_CHECK - 1)) != 0)
+ return false;
+
+ if (ssm_rbuff_get_flags(flow->rx_rb) & RB_REKEY)
+ return false;
+
+ return crypt_nodes_left(flow->crypt) <= KEY_REKEY_WATERMARK;
+}
+
ssize_t flow_write(int fd,
const void * buf,
size_t count)
@@ -1705,6 +2013,19 @@ ssize_t flow_write(int fd,
if ((flags & FLOWFACCMODE) == FLOWFRDONLY)
return -EPERM;
+ 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();
if (flow->frcti != NULL) {
@@ -1736,8 +2057,12 @@ ssize_t flow_write(int fd,
ret = flow_tx_spb(flow, spb, FRCT_FR_SOLE,
!(flags & FLOWFWNOBLOCK), dl);
+ if (ret < 0) {
+ ssm_pool_remove(proc.pool, idx);
+ return (ssize_t) ret;
+ }
- return ret < 0 ? (ssize_t) ret : (ssize_t) count;
+ return (ssize_t) count;
}
static ssize_t flow_rx_spb(struct flow * flow,
@@ -1775,6 +2100,10 @@ static ssize_t raw_flow_read_pkt(struct flow * flow,
ssize_t idx;
while (true) {
+ if (flow->crypt != NULL
+ && (ssm_rbuff_get_flags(flow->rx_rb) & RB_REKEY))
+ flow_rekey(flow);
+
if (!block) {
idx = ssm_rbuff_read(flow->rx_rb);
if (idx < 0)
@@ -1908,6 +2237,13 @@ ssize_t flow_read(int fd,
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 (ACK-only) flows. */
+ flow_tx_promote(flow);
+
tw_move_safe();
idx = flow->part_idx;
@@ -2092,6 +2428,18 @@ static int fqueue_filter(struct fqueue * fq)
pthread_rwlock_rdlock(&proc.lock);
while (fq->next < fq->fqsize) {
+ if (fq->fqueue[fq->next].event == FLOW_UPD) {
+ /* Re-key doorbell: pull internally, never surface. */
+ fd = proc.id_to_fd[fq->fqueue[fq->next].flow_id].fd;
+ ++fq->next;
+ if (fd >= 0) {
+ pthread_rwlock_unlock(&proc.lock);
+ flow_rekey(&proc.flows[fd]);
+ pthread_rwlock_rdlock(&proc.lock);
+ }
+ continue;
+ }
+
if (fq->fqueue[fq->next].event != FLOW_PKT) {
ret = 1;
goto out;
@@ -2224,7 +2572,8 @@ int np1_flow_alloc(pid_t n_pid,
int flow_id)
{
struct flow_info flow;
- struct crypt_sk crypt = { .nid = NID_undef, .key = NULL };
+ struct crypt_sk crypt = { .nid = NID_undef, .key = NULL,
+ .epoch = 0, .role = CRYPT_ROLE_INIT };
memset(&flow, 0, sizeof(flow));
@@ -2235,7 +2584,7 @@ int np1_flow_alloc(pid_t n_pid,
/* np1 flow: n_1_pid is the upper. */
flow.n_1_pid = n_pid;
- return flow_init(&flow, &crypt, 0);
+ return flow_init(&flow, &crypt, 0, NULL);
}
int np1_flow_dealloc(int flow_id,
@@ -2272,6 +2621,38 @@ int np1_flow_resp(int flow_id,
return fd;
}
+int np1_flow_fd(int flow_id)
+{
+ int fd;
+
+ if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+ return -1;
+
+ pthread_rwlock_rdlock(&proc.lock);
+
+ fd = proc.id_to_fd[flow_id].fd;
+
+ pthread_rwlock_unlock(&proc.lock);
+
+ return fd;
+}
+
+int np1_flow_id(int fd)
+{
+ int flow_id;
+
+ if (fd < 0 || fd >= PROC_MAX_FLOWS)
+ return -1;
+
+ pthread_rwlock_rdlock(&proc.lock);
+
+ flow_id = proc.flows[fd].info.id;
+
+ pthread_rwlock_unlock(&proc.lock);
+
+ return flow_id;
+}
+
int ipcp_create_r(const struct ipcp_info * info)
{
uint8_t buf[SOCK_BUF_SIZE];
@@ -2288,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,
@@ -2309,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;
@@ -2317,7 +2705,9 @@ int ipcp_flow_req_arr(const buffer_t * dst,
if (err < 0)
return err;
- crypt.key = key;
+ crypt.key = key;
+ crypt.epoch = 0;
+ crypt.role = CRYPT_ROLE_INIT;
err = flow__irm_result_des(&msg, &flow, &crypt);
if (err < 0)
@@ -2335,7 +2725,30 @@ int ipcp_flow_req_arr(const buffer_t * dst,
crypt.nid = NID_undef;
- return flow_init(&flow, &crypt, 0);
+ return flow_init(&flow, &crypt, 0, NULL);
+}
+
+int ipcp_flow_update_arr(int flow_id,
+ const buffer_t * data)
+{
+ struct flow_info flow;
+ uint8_t buf[SOCK_BUF_SIZE];
+ buffer_t msg = {SOCK_BUF_SIZE, buf};
+ int err;
+
+ memset(&flow, 0, sizeof(flow));
+
+ flow.id = flow_id;
+ flow.n_1_pid = getpid();
+
+ if (ipcp_flow_update_arr__irm_req_ser(&msg, &flow, data) < 0)
+ return -ENOMEM;
+
+ err = send_recv_msg(&msg);
+ if (err < 0)
+ return err;
+
+ return irm__irm_result_des(&msg);
}
int ipcp_flow_alloc_reply(int fd,
@@ -2359,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;
@@ -2373,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);
@@ -2383,37 +2803,101 @@ int ipcp_flow_read(int fd,
pthread_rwlock_rdlock(&proc.lock);
- assert(flow->info.id >= 0);
+ if (flow->info.id < 0) {
+ pthread_rwlock_unlock(&proc.lock);
+ 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) {
- pthread_rwlock_unlock(&proc.lock);
idx = flow_rx_spb(flow, spb, false, NULL);
+ pthread_rwlock_unlock(&proc.lock);
return idx < 0 ? (int) idx : 0;
}
while (!FRCTI_PDU_READY(flow->frcti)) {
- pthread_rwlock_unlock(&proc.lock);
-
idx = flow_rx_spb(flow, spb, false, NULL);
- if (idx < 0)
+ if (idx < 0) {
+ pthread_rwlock_unlock(&proc.lock);
return idx;
-
- pthread_rwlock_rdlock(&proc.lock);
+ }
FRCTI_RCV(flow->frcti, *spb);
}
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);
@@ -2432,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;
@@ -2472,10 +3002,13 @@ int np1_flow_read(int fd,
flow = &proc.flows[fd];
- assert(flow->info.id >= 0);
-
pthread_rwlock_rdlock(&proc.lock);
+ if (flow->info.id < 0) {
+ pthread_rwlock_unlock(&proc.lock);
+ return -ENOTALLOC;
+ }
+
off = ssm_rbuff_read(flow->rx_rb);
if (off < 0) {
pthread_rwlock_unlock(&proc.lock);
@@ -2498,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)
@@ -2576,12 +3114,14 @@ int ipcp_flow_fini(int fd)
return -1;
}
- ssm_rbuff_set_acl(proc.flows[fd].rx_rb, ACL_FLOWDOWN);
- ssm_rbuff_set_acl(proc.flows[fd].tx_rb, ACL_FLOWDOWN);
+ ssm_rbuff_set_flags(proc.flows[fd].rx_rb, RB_FLOWDOWN);
+
+ if (proc.flows[fd].tx_rb != NULL)
+ ssm_rbuff_set_flags(proc.flows[fd].tx_rb, RB_FLOWDOWN);
- ssm_flow_set_notify(proc.flows[fd].set,
- proc.flows[fd].info.id,
- FLOW_DEALLOC);
+ if (proc.flows[fd].set != NULL)
+ ssm_flow_set_notify(proc.flows[fd].set, proc.flows[fd].info.id,
+ FLOW_DEALLOC);
rx_rb = proc.flows[fd].rx_rb;
@@ -2610,19 +3150,101 @@ int ipcp_flow_get_qoscube(int fd,
return 0;
}
+/* Not a snapshot: two atomic loads; caller keeps the fd live. */
size_t ipcp_flow_queued(int fd)
{
- size_t q;
+ assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+ assert(proc.flows[fd].info.id >= 0);
- pthread_rwlock_rdlock(&proc.lock);
+ if (proc.flows[fd].poa != NULL)
+ return poa_flow_qlen(proc.flows[fd].poa);
+
+ return ssm_rbuff_queued(proc.flows[fd].tx_rb)
+ * LOAD_RELAXED(&proc.flows[fd].mean_len);
+}
+size_t ipcp_flow_mean_len(int fd)
+{
+ assert(fd >= 0 && fd < PROC_MAX_FLOWS);
assert(proc.flows[fd].info.id >= 0);
- q = ssm_rbuff_queued(proc.flows[fd].tx_rb);
+ if (proc.flows[fd].poa != NULL)
+ return poa_flow_mean_len(proc.flows[fd].poa);
- pthread_rwlock_unlock(&proc.lock);
+ return LOAD_RELAXED(&proc.flows[fd].mean_len);
+}
+
+/* An update racing the arm seeds one bogus window; the filter absorbs. */
+int ipcp_flow_cap_arm(int fd)
+{
+ struct flow * flow;
+ struct cap_est * e;
- return q;
+ assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+ assert(proc.flows[fd].info.id >= 0);
+
+ flow = &proc.flows[fd];
+ if (flow->poa != NULL) {
+ cap_clear(poa_flow_cap_est(flow->poa));
+ return 0;
+ }
+
+ e = flow->cap;
+ if (e != NULL) {
+ cap_clear(e);
+ return 0;
+ }
+
+ if (posix_memalign((void **) &e, CAP_ALIGN, sizeof(*e)) != 0)
+ return -ENOMEM;
+
+ cap_clear(e);
+
+ STORE_RELEASE(&flow->cap, e);
+
+ return 0;
+}
+
+void ipcp_flow_cap_update(int fd,
+ size_t qlen,
+ size_t len)
+{
+ struct flow * flow;
+ struct cap_est * e;
+
+ assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+ assert(proc.flows[fd].info.id >= 0);
+
+ flow = &proc.flows[fd];
+ if (flow->poa != NULL) {
+ cap_update(poa_flow_cap_est(flow->poa), qlen, len);
+ return;
+ }
+
+ e = LOAD_ACQUIRE(&flow->cap);
+ if (e == NULL)
+ return;
+
+ cap_update(e, qlen, len);
+}
+
+uint64_t ipcp_flow_cap(int fd)
+{
+ struct flow * flow;
+ struct cap_est * e;
+
+ assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+ assert(proc.flows[fd].info.id >= 0);
+
+ flow = &proc.flows[fd];
+ if (flow->poa != NULL)
+ return cap_rate(poa_flow_cap_est(flow->poa));
+
+ e = LOAD_ACQUIRE(&flow->cap);
+ if (e == NULL)
+ return 0;
+
+ return cap_rate(e);
}
int local_flow_transfer(int src_fd,
@@ -2690,3 +3312,6 @@ int local_flow_transfer(int src_fd,
return ret;
}
+
+#include "cap.c"
+#include "poa/poa.c"
generated by cgit v1.2.3 (git 2.46.0) at 2026-09-07 17:16:12 +0000