fix: a full reorder ring keeps the arriving packet and counts the overflow, instead of dropping it silently

A packet that needed a slot and found none was neither held nor delivered: no
counter recorded it and the consumer reserved a hole nothing could ever backfill.
It now commits the gap to free a slot and takes the ordering decision once more
against the sequence that commit left behind, bounded to one retry; past that it
is delivered in order with the counters it skipped as its `lost`, never flagged
late, because a packet ahead of the sequence has no hole to be looked up by. A
duplicate can now replace its slot at a full ring, as hold_packet's own comment
already promised, and the 20 ms default window is a named constant.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino FerrariandClaude Opus 4.6 committed 2026-10-09 06:50:43 +02:00
1 parent cb1bbd2b7b
commit 176ef1df9f
6 files changed
+113 -38

No files matched your search

+79 -33
View File
@@ -37,6 +37,8 @@
#define UDPS_REORDER_SLOTS 64 #define UDPS_REORDER_SLOTS 64
/** Total bytes the reorder ring may hold, so a fast stream cannot grow it without bound. */ /** Total bytes the reorder ring may hold, so a fast stream cannot grow it without bound. */
#define UDPS_REORDER_BYTES (4u * 1024u * 1024u) #define UDPS_REORDER_BYTES (4u * 1024u * 1024u)
/** How long a gap is given to resolve before it is committed, by default. */
#define UDPS_REORDER_WINDOW_S_DEFAULT 0.020
/** Concurrent in-flight reassemblies. The server interleaves at most a couple. */ /** Concurrent in-flight reassemblies. The server interleaves at most a couple. */
#define UDPS_MAX_SLOTS 4 #define UDPS_MAX_SLOTS 4
@@ -576,20 +578,16 @@ static int deliver_data(udps_client_t *c, const uint8_t *pl, size_t len,
* *
* A duplicate counter (same datagram arriving twice on the wire) replaces the * A duplicate counter (same datagram arriving twice on the wire) replaces the
* existing slot so one counter never consumes two slots and both get delivered. * existing slot so one counter never consumes two slots and both get delivered.
* The search for that slot therefore runs *before* the capacity checks, and the
* checks are applied only on the path that needs a new slot: a duplicate at a
* full ring consumes nothing and must still be accepted.
* Returns non-zero and holds nothing when there is no room (ring full or byte * Returns non-zero and holds nothing when there is no room (ring full or byte
* budget exhausted). */ * budget exhausted); `order_data` below says what the caller then does. */
static int hold_packet(udps_client_t *c, const uint8_t *pl, size_t len, static int hold_packet(udps_client_t *c, const uint8_t *pl, size_t len,
uint32_t counter, double recv_time) { uint32_t counter, double recv_time) {
int i; int i;
int slot = -1; int slot = -1;
if (c->held_count >= UDPS_REORDER_SLOTS) {
return -1;
}
if (c->held_bytes + len > UDPS_REORDER_BYTES) {
return -1;
}
/* Duplicate counter: reuse the existing slot to avoid double delivery. */ /* Duplicate counter: reuse the existing slot to avoid double delivery. */
for (i = 0; i < UDPS_REORDER_SLOTS; i++) { for (i = 0; i < UDPS_REORDER_SLOTS; i++) {
if (c->held[i].used && c->held[i].counter == counter) { if (c->held[i].used && c->held[i].counter == counter) {
@@ -597,16 +595,28 @@ static int hold_packet(udps_client_t *c, const uint8_t *pl, size_t len,
break; break;
} }
} }
if (slot < 0) { if (slot >= 0) {
/* The replacement releases the length already held, so the budget is
* weighed against what the slot will cost rather than what it adds. */
if (c->held_bytes - c->held[slot].len + len > UDPS_REORDER_BYTES) {
return -1;
}
} else {
if (c->held_count >= UDPS_REORDER_SLOTS) {
return -1;
}
if (c->held_bytes + len > UDPS_REORDER_BYTES) {
return -1;
}
for (i = 0; i < UDPS_REORDER_SLOTS; i++) { for (i = 0; i < UDPS_REORDER_SLOTS; i++) {
if (!c->held[i].used) { if (!c->held[i].used) {
slot = i; slot = i;
break; break;
} }
} }
} if (slot < 0) {
if (slot < 0) { return -1;
return -1; }
} }
if (ensure_cap((void **)&c->held[slot].buf, &c->held[slot].cap, len) != 0) { if (ensure_cap((void **)&c->held[slot].buf, &c->held[slot].cap, len) != 0) {
@@ -717,10 +727,27 @@ static int commit_gap(udps_client_t *c) {
* milliseconds, and holding is what turns it back into order. A packet behind the * milliseconds, and holding is what turns it back into order. A packet behind the
* sequence lost that race: the consumer already has a hole where it belongs, so it * sequence lost that race: the consumer already has a hole where it belongs, so it
* is delivered flagged rather than discarded, which is the whole point -- a dropped * is delivered flagged rather than discarded, which is the whole point -- a dropped
* packet cannot fill the hole it left. */ * packet cannot fill the hole it left.
*
* The full ring is specified here, in one place. A packet that needs a slot and
* finds none gives up on the current gap -- `commit_gap()` delivers the lowest held
* run and so frees at least one slot -- counts the event in
* `stats.reorder_overflows`, and the decision is then taken again, once, against
* the sequence that commit left behind. That re-decision is why this is a bounded
* loop and not a straight line: the commit advances `next_counter` to the lowest
* held counter, which may put it past the arriving packet, and the packet is then
* behind the sequence rather than ahead of it. Whatever happens, the arriving
* packet is held or delivered; it is never dropped, because nothing else records it
* and the consumer would reserve a hole for it that nothing could ever backfill.
* Past the cap it is delivered *in order* and not flagged late: a packet ahead of
* the sequence has no reserved hole, the consumer looks a late packet's hole up by
* counter and ignores one it cannot place, so advancing `next_counter` to it and
* reporting the counters it skipped as `lost` is what reserves their hole. */
static int order_data(udps_client_t *c, const uint8_t *pl, size_t len, static int order_data(udps_client_t *c, const uint8_t *pl, size_t len,
uint32_t counter, double recv_time) { uint32_t counter, double recv_time) {
int32_t delta; int32_t delta;
int attempt;
int rc;
if (c->cfg.reorder_window_s <= 0.0) { if (c->cfg.reorder_window_s <= 0.0) {
/* Resequencing off: byte-for-byte the behaviour before this task. `delta` /* Resequencing off: byte-for-byte the behaviour before this task. `delta`
@@ -744,27 +771,46 @@ static int order_data(udps_client_t *c, const uint8_t *pl, size_t len,
c->next_counter = counter + 1u; c->next_counter = counter + 1u;
return deliver_data(c, pl, len, counter, recv_time, 0u, 0); return deliver_data(c, pl, len, counter, recv_time, 0u, 0);
} }
for (attempt = 0; attempt < 2; attempt++) {
delta = (int32_t)(counter - c->next_counter);
if (delta == 0) {
rc = deliver_data(c, pl, len, counter, recv_time, 0u, 0);
c->next_counter = counter + 1u;
if (rc == 0) { rc = drain_held(c, 0u); }
return rc;
}
if (delta < 0) {
/* Behind the sequence. */
c->stats.late_delivered++;
return deliver_data(c, pl, len, counter, recv_time, 0u, 1);
}
if (hold_packet(c, pl, len, counter, recv_time) == 0) {
if (c->gap_deadline == 0.0) {
c->gap_deadline = recv_time + c->cfg.reorder_window_s;
}
return 0;
}
/* No room: give up on the gap rather than growing the ring, and say so
* in a counter, so an undersized window is a reading and not a mystery. */
c->stats.reorder_overflows++;
rc = commit_gap(c);
if (rc != 0) {
/* The consumer refused a delivery; nothing is freed and nothing more
* may be delivered, so the arriving packet is its caller's problem. */
return rc;
}
}
/* The attempt cap is spent and the ring still has no room (the byte budget
* cannot fit this one even with a slot free). Deliver it rather than drop it. */
delta = (int32_t)(counter - c->next_counter); delta = (int32_t)(counter - c->next_counter);
if (delta == 0) { if (delta < 0) {
int rc = deliver_data(c, pl, len, counter, recv_time, 0u, 0); c->stats.late_delivered++;
c->next_counter = counter + 1u; return deliver_data(c, pl, len, counter, recv_time, 0u, 1);
if (rc == 0) { rc = drain_held(c, 0u); }
return rc;
} }
if (delta > 0) { rc = deliver_data(c, pl, len, counter, recv_time, (uint32_t)delta, 0);
if (hold_packet(c, pl, len, counter, recv_time) != 0) { c->next_counter = counter + 1u;
/* No room: commit the gap now rather than growing. The counter is if (rc == 0) { rc = drain_held(c, 0u); }
* visible, so an undersized window is a reading and not a mystery. */ return rc;
return commit_gap(c);
}
if (c->gap_deadline == 0.0) {
c->gap_deadline = recv_time + c->cfg.reorder_window_s;
}
return 0;
}
/* delta < 0: behind the sequence. */
c->stats.late_delivered++;
return deliver_data(c, pl, len, counter, recv_time, 0u, 1);
} }
/*---------------------------------------------------------------------------*/ /*---------------------------------------------------------------------------*/
@@ -1187,7 +1233,7 @@ void udps_client_config_init(udps_client_config_t *cfg) {
cfg->keepalive_interval_s = 15.0; /* server evicts silent clients at 30 s */ cfg->keepalive_interval_s = 15.0; /* server evicts silent clients at 30 s */
cfg->recv_buffer_bytes = 4u * 1024u * 1024u; cfg->recv_buffer_bytes = 4u * 1024u * 1024u;
cfg->max_packet_bytes = 1u * 1024u * 1024u; cfg->max_packet_bytes = 1u * 1024u * 1024u;
cfg->reorder_window_s = 0.020; cfg->reorder_window_s = UDPS_REORDER_WINDOW_S_DEFAULT;
} }
static void copy_str(char *dst, size_t cap, const char *src) { static void copy_str(char *dst, size_t cap, const char *src) {
+9
View File
@@ -202,6 +202,15 @@ typedef struct {
* before the window existed they were counted in @c stale_packets and dropped. * before the window existed they were counted in @c stale_packets and dropped.
*/ */
uint64_t late_delivered; uint64_t late_delivered;
/**
* Times a DATA packet needed a slot in the reorder ring and found none, so the
* current gap was committed early to make room. Every such packet is still held
* or delivered -- none is dropped -- but the window it was given was shorter
* than @c reorder_window_s asked for. A non-zero reading says the ring is
* undersized for the stream's reordering depth, which is a reading and not a
* mystery; it is the only place that shows up.
*/
uint64_t reorder_overflows;
uint64_t reconnects; uint64_t reconnects;
} udps_stats_t; } udps_stats_t;
+13 -2
View File
@@ -63,7 +63,7 @@ handler clears either pointer without knowing which dialog it is. Presenting eit
app's plain `GtkApplicationWindow` is supported and warning-free; it was checked. app's plain `GtkApplicationWindow` is supported and warning-free; it was checked.
`deps/udpsclient` is a vendored C99 source pair and its diff from the upstream source `deps/udpsclient` is a vendored C99 source pair and its diff from the upstream source
is intentional. Four things were changed: is intentional. Five things were changed:
1. **A reorder ring of reassembled payloads.** When a packet arrives with a higher counter 1. **A reorder ring of reassembled payloads.** When a packet arrives with a higher counter
than expected, the raw datagram is held in a fixed-size ring rather than decoded immediately. than expected, the raw datagram is held in a fixed-size ring rather than decoded immediately.
@@ -78,11 +78,22 @@ is intentional. Four things were changed:
timeout; the reorder deadline is a third call to the same helper, which already understood timeout; the reorder deadline is a third call to the same helper, which already understood
the pattern. the pattern.
4. **`reorder_window_s = 0` as the switch.** `udps_client_config_t::reorder_window_s`, 4. **`reorder_window_s = 0` as the switch.** `udps_client_config_t::reorder_window_s`,
defaulted to `0.020` in `udps_client_default_config()` (`deps/udpsclient/udps_client.c:1190`), defaulted to `UDPS_REORDER_WINDOW_S_DEFAULT` (0.020) in `udps_client_config_init()`,
is the only knob. Setting it to zero disables resequencing and restores the previous is the only knob. Setting it to zero disables resequencing and restores the previous
discard-on-late behaviour exactly — so a reported regression is bisectable without reverting discard-on-late behaviour exactly — so a reported regression is bisectable without reverting
the source. **There is no command-line flag for the reorder window.** It is not surfaced on the source. **There is no command-line flag for the reorder window.** It is not surfaced on
udpscope's command line; do not document a `--reorder-window-ms`. udpscope's command line; do not document a `--reorder-window-ms`.
5. **The full ring never drops the arriving packet, and says so in a counter.** A packet that
needs a slot and finds none commits the current gap (which frees at least one, since
`commit_gap()` always delivers) and the ordering decision is taken once more against the
sequence that commit left behind — a bounded retry, because the commit can advance
`next_counter` past the arriving counter and turn an ahead-of-sequence packet into a behind-
of-sequence one. Past the retry it is delivered *in order*, with the counters it skipped as
its `lost`, and never flagged `late`: a packet ahead of the sequence has no reserved hole and
`Acquisition::backfill_from()` silently ignores a late packet it cannot place, so flagging it
would be the same silent loss by another route. `stats.reorder_overflows` counts the event,
so an undersized ring is a reading rather than a mystery; it is surfaced in the exit dump and
the `[reseq-probe]` line as `overflows=`.
`deps/plotter` is upstream's own repository and this app's figure support was added `deps/plotter` is upstream's own repository and this app's figure support was added
there rather than worked around here: `Figure::set_min_y_gutter()` (a floor under the there rather than worked around here: `Figure::set_min_y_gutter()` (a floor under the
+1
View File
@@ -94,6 +94,7 @@ void Acquisition::run() {
out.resequenced.store(s.late_resequenced, std::memory_order_relaxed); out.resequenced.store(s.late_resequenced, std::memory_order_relaxed);
out.late.store(s.late_delivered, std::memory_order_relaxed); out.late.store(s.late_delivered, std::memory_order_relaxed);
out.stale.store(s.stale_packets, std::memory_order_relaxed); out.stale.store(s.stale_packets, std::memory_order_relaxed);
out.overflows.store(s.reorder_overflows, std::memory_order_relaxed);
out.reconnects.store(s.reconnects, std::memory_order_relaxed); out.reconnects.store(s.reconnects, std::memory_order_relaxed);
out.connected.store(udps_client_is_connected(client_) != 0, std::memory_order_relaxed); out.connected.store(udps_client_is_connected(client_) != 0, std::memory_order_relaxed);
} }
+5
View File
@@ -216,6 +216,10 @@ struct StreamStats {
std::atomic<std::uint64_t> resequenced{0}; std::atomic<std::uint64_t> resequenced{0};
/// Out-of-order packets delivered behind the sequence, past the window. /// Out-of-order packets delivered behind the sequence, past the window.
std::atomic<std::uint64_t> late{0}; std::atomic<std::uint64_t> late{0};
/// Times the client's reorder ring was full when a packet needed a slot, so the gap was
/// committed early. Nothing is dropped; the window was just shorter than asked for, and a
/// non-zero reading is what says the ring is undersized for the stream.
std::atomic<std::uint64_t> overflows{0};
/// Packets the client still refused outright (a duplicate, or any late packet while /// Packets the client still refused outright (a duplicate, or any late packet while
/// resequencing is disabled). Counted since the beginning and displayed nowhere until /// resequencing is disabled). Counted since the beginning and displayed nowhere until
/// now, which is why an inversion had to be inferred from `packets` and `frames` /// now, which is why an inversion had to be inferred from `packets` and `frames`
@@ -547,6 +551,7 @@ class ScopeModel {
/// Rebuilds buffers from the current channel list. Caller holds `mutex()`. /// Rebuilds buffers from the current channel list. Caller holds `mutex()`.
void rebuild_locked(); void rebuild_locked();
private: private:
void rebuild_channels(); void rebuild_channels();
/// Re-appends one synthetic signal per math channel, after the stream's own. Called under the /// Re-appends one synthetic signal per math channel, after the stream's own. Called under the
+6 -3
View File
@@ -10559,12 +10559,14 @@ gboolean reseq_probe(gpointer user) {
} }
} }
std::fprintf(stderr, std::fprintf(stderr,
"[reseq-probe] reseq=%llu late=%llu stale=%llu holes=%llu hole_slots=%llu " "[reseq-probe] reseq=%llu late=%llu stale=%llu overflows=%llu holes=%llu "
"hole_slots=%llu "
"backfilled=%llu open_holes=%zu bands=%zu bands_repaired=%zu nan_slots=%llu " "backfilled=%llu open_holes=%zu bands=%zu bands_repaired=%zu nan_slots=%llu "
"scanned=%llu trigger_in_hole=%d\n", "scanned=%llu trigger_in_hole=%d\n",
static_cast<unsigned long long>(m.stats().resequenced.load()), static_cast<unsigned long long>(m.stats().resequenced.load()),
static_cast<unsigned long long>(m.stats().late.load()), static_cast<unsigned long long>(m.stats().late.load()),
static_cast<unsigned long long>(m.stats().stale.load()), static_cast<unsigned long long>(m.stats().stale.load()),
static_cast<unsigned long long>(m.stats().overflows.load()),
static_cast<unsigned long long>(m.stats().holes.load()), static_cast<unsigned long long>(m.stats().holes.load()),
static_cast<unsigned long long>(m.stats().hole_slots.load()), static_cast<unsigned long long>(m.stats().hole_slots.load()),
static_cast<unsigned long long>(m.stats().backfilled.load()), open_holes, bands, static_cast<unsigned long long>(m.stats().backfilled.load()), open_holes, bands,
@@ -10809,8 +10811,8 @@ int run_ui(int argc, char** argv, const Config& cfg, ScopeModel& model, Acquisit
const Partition& part = model.partition(); const Partition& part = model.partition();
std::printf("[udpscope] frames=%llu packets=%llu gaps=%llu channels=%zu span=%.3fs " std::printf("[udpscope] frames=%llu packets=%llu gaps=%llu channels=%zu span=%.3fs "
"stride=%llu" "stride=%llu"
" reseq=%llu late=%llu stale=%llu holes=%llu hole_slots=%llu backfilled=%llu" " reseq=%llu late=%llu stale=%llu overflows=%llu holes=%llu hole_slots=%llu"
"\n", " backfilled=%llu\n",
static_cast<unsigned long long>(model.stats().frames.load()), static_cast<unsigned long long>(model.stats().frames.load()),
static_cast<unsigned long long>(model.stats().packets.load()), static_cast<unsigned long long>(model.stats().packets.load()),
static_cast<unsigned long long>(model.stats().gaps.load()), static_cast<unsigned long long>(model.stats().gaps.load()),
@@ -10819,6 +10821,7 @@ int run_ui(int argc, char** argv, const Config& cfg, ScopeModel& model, Acquisit
static_cast<unsigned long long>(model.stats().resequenced.load()), static_cast<unsigned long long>(model.stats().resequenced.load()),
static_cast<unsigned long long>(model.stats().late.load()), static_cast<unsigned long long>(model.stats().late.load()),
static_cast<unsigned long long>(model.stats().stale.load()), static_cast<unsigned long long>(model.stats().stale.load()),
static_cast<unsigned long long>(model.stats().overflows.load()),
static_cast<unsigned long long>(model.stats().holes.load()), static_cast<unsigned long long>(model.stats().holes.load()),
static_cast<unsigned long long>(model.stats().hole_slots.load()), static_cast<unsigned long long>(model.stats().hole_slots.load()),
static_cast<unsigned long long>(model.stats().backfilled.load())); static_cast<unsigned long long>(model.stats().backfilled.load()));