summaryrefslogtreecommitdiff
path: root/src/lib
diff options
context:
space:
mode:
Diffstat (limited to 'src/lib')
-rw-r--r--src/lib/dev.c29
-rw-r--r--src/lib/frct.c68
-rw-r--r--src/lib/pb/ipcp_config.proto1
-rw-r--r--src/lib/protobuf.c3
-rw-r--r--src/lib/ssm/rbuff.c2
-rw-r--r--src/lib/tests/crypt_test.c8
6 files changed, 74 insertions, 37 deletions
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 166aba5c..3fb8d831 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -98,14 +98,14 @@ struct flow {
ssize_t part_idx;
struct crypt_ctx * crypt;
- int headsz; /* Selector */
- int tailsz; /* Tag + CRC */
+ int headsz; /* Selector */
+ int tailsz; /* Tag + CRC */
- struct timespec rk_grace; /* TX-promote deadline (0 = none) */
- struct timespec rk_attempt; /* Last re-key attempt (backoff) */
- 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 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;
@@ -1739,8 +1739,7 @@ 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)
@@ -2887,19 +2886,13 @@ 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;
-
- pthread_rwlock_rdlock(&proc.lock);
-
+ assert(fd >= 0 && fd < PROC_MAX_FLOWS);
assert(proc.flows[fd].info.id >= 0);
- q = ssm_rbuff_queued(proc.flows[fd].tx_rb);
-
- pthread_rwlock_unlock(&proc.lock);
-
- return q;
+ return ssm_rbuff_queued(proc.flows[fd].tx_rb);
}
int local_flow_transfer(int src_fd,
diff --git a/src/lib/frct.c b/src/lib/frct.c
index c055433d..efd50b9a 100644
--- a/src/lib/frct.c
+++ b/src/lib/frct.c
@@ -193,6 +193,8 @@ struct frcti_stat {
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_fast_skip; /* SACK skips: slot has FAST_RXM */
+ size_t rxm_fast_stuck; /* those skips with age > rto */
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 */
@@ -549,6 +551,8 @@ static int frct_rib_read(const char * path,
" bail (unowned): %20zu\n"
" bail (aged): %20zu\n"
" bail (defer): %20zu\n"
+ " skip (fast-rxm set): %20zu\n"
+ " skip (stuck past rto): %20zu\n"
"RXM-arm malloc failures: %20zu\n"
"RXM cancels (teardown): %20zu\n"
"RXM tx into dead flow: %20zu\n"
@@ -597,6 +601,7 @@ 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_fast_skip, s.stat.rxm_fast_stuck,
s.stat.rxm_arm_fail,
s.stat.rxm_cancel,
s.stat.rxm_tx_dead, s.stat.tx_drop,
@@ -1857,6 +1862,7 @@ struct frcti * frcti_create(int fd,
/ SACK_BLOCK_SIZE;
if (bb > SACK_MAX_BLOCKS)
bb = SACK_MAX_BLOCKS;
+
frcti->sack_n_max = (uint16_t) bb;
frcti->max_rcv_sdu = FRCT_MAX_SDU;
@@ -1948,10 +1954,13 @@ 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_dup=%zu "
+ "rxm_rto=%zu rxm_sack=%zu rxm_rack=%zu rxm_dup=%zu "
"rxm_due=%zu acked=%zu unowned=%zu aged=%zu defer=%zu "
+ "fast_skip=%zu fast_stuck=%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 "
+ "out_rcv=%zu rqo_rcv=%zu dup_rcv=%zu rxm_dup_rcv=%zu "
"drf_rebase=%zu rq_released=%zu\n",
(int) getpid(), frcti->fd,
frcti->stat.sdu_snd_frag, frcti->stat.sdu_reasm,
@@ -1959,14 +1968,19 @@ 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_dupthresh,
+ frcti->stat.rxm_rack, frcti->stat.rxm_dupthresh,
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_fast_skip, frcti->stat.rxm_fast_stuck,
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,
frcti->stat.inact_drop,
+ frcti->stat.tlp_snd, frcti->stat.sack_snd,
+ frcti->stat.sack_rcv, frcti->stat.ack_supp_seqno,
+ frcti->stat.out_rcv, frcti->stat.rqo_rcv,
+ frcti->stat.dup_rcv, frcti->stat.rxm_dup_rcv,
frcti->stat.drf_rebase, frcti->stat.rq_released);
#endif
@@ -2062,6 +2076,7 @@ int frcti_set_rcv_ring_sz(struct frcti * frcti,
if (!frcti->stream)
return -ENOTSUP;
+
if (!stream_ring_sz_ok(frcti, n))
return -EINVAL;
@@ -2930,9 +2945,6 @@ static void tlp_due(void * arg)
if (frcti->snd_cr.seqno == frcti->snd_cr.lwe)
goto unlock;
- if (!before(frcti->snd_cr.seqno, frcti->snd_cr.rwe))
- goto unlock; /* FC-blocked: RDV handles it. */
-
/* RFC 8985 §7.3: one outstanding probe, MAX_TLP_PER_EP per ep. */
if (frcti->tlp_high_seq != 0)
goto unlock;
@@ -2965,10 +2977,10 @@ static void tlp_due(void * arg)
memcpy(pkt_copy, rxm->pkt, rxm->len);
pkt_len = rxm->len;
frcti->snd_slots[hp].time = now_ns;
- frcti->snd_slots[hp].flags |= SND_TLP | SND_FAST_RXM;
+ frcti->snd_slots[hp].flags |= SND_TLP;
frcti->rtt_lwe = frcti->snd_cr.lwe + 1;
- /* §7.3 outstanding-probe marker; ack_rcv/rxm_snd clear. */
- frcti->tlp_high_seq = frcti->snd_cr.seqno;
+ /* Probe is the HoL: any cum-ACK resolves the episode. */
+ frcti->tlp_high_seq = frcti->snd_cr.lwe + 1;
frcti->tlp_count++;
STAT_BUMP(frcti, tlp_snd);
}
@@ -2998,8 +3010,10 @@ static int tlp_arm(struct frcti * frcti)
/* §7.3: one outstanding probe, MAX_TLP_PER_EP per recovery ep. */
if (LOAD_RELAXED(&frcti->tlp_high_seq) != 0)
return 0;
+
if (LOAD_RELAXED(&frcti->tlp_count) >= MAX_TLP_PER_EP)
return 0;
+
if (__atomic_test_and_set(&frcti->tlp_pending, __ATOMIC_RELAXED))
return 0;
@@ -3082,15 +3096,20 @@ static bool rtt_sample_eligible(struct frcti * frcti,
{
if (flags & FRCT_RXM)
return false;
+
if (frcti->snd_slots[p].flags & (SND_RTX | SND_TLP))
return false;
+
if (LOAD_ACQUIRE(&frcti->snd_slots[p].rxm) == NULL)
return false;
+
if (before(lwe, frcti->rtt_lwe))
return false;
+
/* Don't seed srtt from a cum-ACK; let probes seed. */
if (frcti->srtt == 0)
return false;
+
return true;
}
@@ -3182,8 +3201,10 @@ static void frcti_ack_rcv(struct frcti * frcti,
/* §7.3: cum-ACK past the probed seqno resolves the TLP. */
if (frcti->tlp_high_seq != 0
- && !before(ackno, frcti->tlp_high_seq))
+ && !before(ackno, frcti->tlp_high_seq)) {
frcti->tlp_high_seq = 0;
+ frcti->tlp_count = 0;
+ }
/* §7.3: end the probe episode once inflight drains. */
if (ackno == frcti->snd_cr.seqno)
@@ -3297,8 +3318,22 @@ static void sack_queue_rxm(struct frcti * frcti,
if (rxm == NULL)
continue;
- if (frcti->snd_slots[kp].flags & SND_FAST_RXM)
- continue;
+ /*
+ * 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.
+ */
+ if (frcti->snd_slots[kp].flags & SND_FAST_RXM) {
+ if (!ts_aged_ns(now_ns,
+ frcti->snd_slots[kp].time,
+ LOAD_RELAXED(&frcti->rto))) {
+ STAT_BUMP(frcti, rxm_fast_skip);
+ continue;
+ }
+
+ STAT_BUMP(frcti, rxm_fast_stuck);
+ frcti->snd_slots[kp].flags &= ~SND_FAST_RXM;
+ }
if (RXM_AGED_OUT(rxm->t0, now_ns, frcti->t_r))
continue;
@@ -3584,13 +3619,10 @@ static bool sack_check(struct frcti * frcti,
n = dsack_consume(frcti, out->blocks);
if (n == 1)
out->dsack = true;
+
n += sack_blocks_build(frcti, out->blocks + n,
frcti->sack_n_max - n);
- if (!out->dsack
- && rcv_cr->lwe == frcti->sack_lwe && n == frcti->sack_n)
- return false;
-
out->n = n;
out->ack = rcv_cr->lwe;
out->rwe = frcti_advert_rwe(frcti);
@@ -3646,6 +3678,7 @@ static void seqno_rotate(struct frcti * frcti,
if (!ts_aged_ns(now_ns, snd_cr->act, snd_cr->inact))
return;
+
/* Idle-on-wire ≠ idle e2e: don't orphan in-flight rxm. */
if (snd_cr->seqno != snd_cr->lwe)
return;
@@ -3709,9 +3742,6 @@ static int frcti_snd(struct frcti * frcti,
memset(pci, 0, FRCT_PCILEN);
- if (frcti->stream)
- spci = FRCT_SPCI(pci);
-
clock_gettime(PTHREAD_COND_CLOCK, &now);
now_ns = TS_TO_UINT64(now);
@@ -3727,6 +3757,8 @@ static int frcti_snd(struct frcti * frcti,
STAT_BUMP(frcti, frag_snd);
if (frcti->stream) {
+ spci = FRCT_SPCI(pci);
+
if (flags & FRCT_FIN)
pci_flags |= FRCT_FIN;
diff --git a/src/lib/pb/ipcp_config.proto b/src/lib/pb/ipcp_config.proto
index eac4da37..a111b6d8 100644
--- a/src/lib/pb/ipcp_config.proto
+++ b/src/lib/pb/ipcp_config.proto
@@ -42,6 +42,7 @@ message dt_config_msg {
required uint32 eid_size = 2;
required uint32 max_ttl = 3;
required routing_config_msg routing = 4;
+ optional uint32 max_rtt = 5;
}
message dir_dht_config_msg {
diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c
index a824d357..6beae000 100644
--- a/src/lib/protobuf.c
+++ b/src/lib/protobuf.c
@@ -362,6 +362,8 @@ dt_config_msg_t * dt_config_s_to_msg(const struct dt_config * s)
msg->addr_size = s->addr_size;
msg->eid_size = s->eid_size;
msg->max_ttl = s->max_ttl;
+ msg->has_max_rtt = true;
+ msg->max_rtt = s->max_rtt;
msg->routing = routing_config_s_to_msg(&s->routing);
if (msg->routing == NULL)
goto fail_routing;
@@ -381,6 +383,7 @@ struct dt_config dt_config_msg_to_s(const dt_config_msg_t * msg)
s.addr_size = msg->addr_size;
s.eid_size = msg->eid_size;
s.max_ttl = msg->max_ttl;
+ s.max_rtt = msg->has_max_rtt ? msg->max_rtt : 200; /* ms */
s.routing = routing_config_msg_to_s(msg->routing);
return s;
diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c
index 7886b5c2..04978d82 100644
--- a/src/lib/ssm/rbuff.c
+++ b/src/lib/ssm/rbuff.c
@@ -74,7 +74,7 @@ struct ssm_rbuff {
ssize_t * shm_base; /* start of shared memory */
size_t * head; /* start of ringbuffer */
size_t * tail;
- size_t * flags; /* out-of-band flags (RB_*) */
+ size_t * flags; /* out-of-band flags (RB_*) */
pthread_mutex_t * mtx; /* lock for cond vars only */
pthread_cond_t * add; /* signal when new data */
pthread_cond_t * del; /* signal when data removed */
diff --git a/src/lib/tests/crypt_test.c b/src/lib/tests/crypt_test.c
index f00618d8..50b7268a 100644
--- a/src/lib/tests/crypt_test.c
+++ b/src/lib/tests/crypt_test.c
@@ -51,12 +51,20 @@ static int test_crypt_create_destroy(void)
memset(key, 0, sizeof(key));
ctx = crypt_create_ctx(&sk);
+#ifdef HAVE_OPENSSL
if (ctx == NULL) {
printf("Failed to initialize cryptography.\n");
goto fail;
}
crypt_destroy_ctx(ctx);
+#else
+ if (ctx != NULL) {
+ printf("Created cipher context without a backend.\n");
+ crypt_destroy_ctx(ctx);
+ goto fail;
+ }
+#endif
TEST_SUCCESS();
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-26 14:04:26 +0000