Twelve tasks in five phases, so relative timing ships before any of the client-side reordering work lands. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1552 lines
68 KiB
Markdown
1552 lines
68 KiB
Markdown
<!-- SPDX-License-Identifier: MIT -->
|
||
<!-- SPDX-FileCopyrightText: 2026 Martino Ferrari -->
|
||
|
||
# Resequencing and Relative Timing Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** Recover out-of-order UDP packets instead of discarding them — resequence inside a 20 ms window, and reserve an invalidated hole that a later arrival backfills — and add a relative-time mode whose x comes from the packet counter and the declared rate rather than from a jittery stream clock.
|
||
|
||
**Architecture:** The client gains a bounded reorder ring of *reassembled raw payloads* released in counter order; a packet past the window is delivered flagged late instead of dropped. `Acquisition` reserves a NaN hole of `lost × batch` samples by running them through its existing per-sample loop, so decimation, stamping and the LOD pyramid stay consistent for free, and backfills values through a new amend path. Relative timing replaces `build_stamps()`'s inferred clock with `(slot_base_ + k) / rate`.
|
||
|
||
**Tech Stack:** C++20 (udpscope, `deps/plotter`), C99 (`deps/udpsclient`), GTK4, CMake + Ninja, Python 3 (`tools/mock_udpstreamer.py`).
|
||
|
||
**Spec:** `docs/superpowers/specs/2026-10-08-resequencing-and-relative-timing-design.md`
|
||
|
||
## Global Constraints
|
||
|
||
- **Every new source file carries the SPDX MIT header** — `// SPDX-License-Identifier: MIT` then `// SPDX-FileCopyrightText: 2026 Martino Ferrari` (`/** */` or `#` comment form in C and Python).
|
||
- **Two build-warning baselines, read on every build:** exactly **four** warnings on a clean build, **two** on an incremental build. More than that is a regression to fix, not to note.
|
||
- **Zero `Gtk-CRITICAL`** in any headless run. One is a defect.
|
||
- **No unit test suite in udpscope.** udpscope changes are verified by the application's own headless self-check (env-var probes under `xvfb-run` against `tools/mock_udpstreamer.py`). `deps/plotter` *does* have a test suite and every library change gets a real test there.
|
||
- **Verification runs are short and serial.** Batching several `xvfb` runs back to back starves the mock past its 2 s `silence_timeout_s` and reads as a reconnect — a measurement artefact, not a defect. Kill leftover mocks between runs, and put `pkill -f mock_udpstreamer.py` inside a script file rather than typing it in the invoking shell (the pattern matches the invoking shell's own argv and the shell kills itself, exit code 144).
|
||
- **The latency budget is 20 ms**, as `UDPS_REORDER_WINDOW_S_DEFAULT = 0.020`. A contiguous arrival is released with **zero** added latency.
|
||
- **The reservation cap is 256 missing packets**, reusing the existing `kGlitchPackets` constant and its reasoning.
|
||
- **`MinMaxPyramid::amend()`'s precondition is non-finite → finite only.** It is additive, not a rewrite.
|
||
|
||
### Build and run commands
|
||
|
||
```bash
|
||
# udpscope
|
||
cmake -S . -B build -G Ninja -DCMAKE_BUILD_TYPE=RelWithDebInfo
|
||
cmake --build build
|
||
|
||
# deps/plotter test suite
|
||
cmake --build deps/plotter/build/debug
|
||
xvfb-run -a deps/plotter/build/debug/tests/plot_tests # expect "0 failed"
|
||
PLOT_TEST_FILTER=amend deps/plotter/build/debug/tests/plot_tests # one test, no EGL needed
|
||
```
|
||
|
||
### File structure
|
||
|
||
| File | Responsibility | Phase |
|
||
| --- | --- | --- |
|
||
| `src/config.hpp`, `src/config.cpp` | `--relative-time` flag | 1 |
|
||
| `src/acquisition.hpp`, `src/acquisition.cpp` | relative stamps, hole reservation, backfill dispatch | 1, 3, 4 |
|
||
| `src/model.hpp`, `src/model.cpp` | absolute-time ring, `StreamStats` counters, invalid-span table, amend queue drain | 1, 3, 4 |
|
||
| `src/channel.hpp`, `src/channel.cpp` | `amend_published()` — applies a queued pyramid amend inside `acquire()` | 4 |
|
||
| `deps/udpsclient/udps_client.h`, `.c` | `reorder_window_s`, the reorder ring, late delivery, two new stats | 2 |
|
||
| `deps/plotter/include/plot/data/ring_buffer.hpp` | `amend()` | 4 |
|
||
| `deps/plotter/include/plot/data/minmax_pyramid.hpp`, `src/data/minmax_pyramid.cpp` | `amend()` | 4 |
|
||
| `deps/plotter/include/plot/plot.hpp`, `src/render/renderer.cpp` | `set_invalid_spans()` and the band | 5 |
|
||
| `src/status_text.cpp` | the four new counters in the detail tier | 3 |
|
||
| `src/ui.cpp` | exit dump, `UDPSCOPE_PROBE_RESEQ_MS` | 3 |
|
||
| `tools/mock_udpstreamer.py` | `--reorder-delay-ms` | 2 |
|
||
|
||
### Correction to the spec, carried into this plan
|
||
|
||
The spec says the backfill queue is an SPSC queue. It need not be: `ScopeModel::acquire_all()` is called from `src/ui.cpp:6265` **inside `std::lock_guard<std::mutex> lock(model.mutex())`**, and the producer takes that same mutex for the whole of `Acquisition::on_data` (`src/acquisition.cpp:386`). A plain `std::vector` guarded by that mutex is correct and simpler. Task 11 fixes the spec line.
|
||
|
||
---
|
||
|
||
## Phase 1 — Relative timing (independently shippable)
|
||
|
||
### Task 1: The `--relative-time` flag
|
||
|
||
**Files:**
|
||
- Modify: `src/config.hpp:37` (add the field), `src/config.cpp:55-57` (usage), `src/config.cpp:94-100` (parse)
|
||
- Test: headless run of `--help`
|
||
|
||
**Interfaces:**
|
||
- Produces: `Config::relative_time` (`bool`, default `false`), read by Task 2.
|
||
|
||
- [ ] **Step 1: Add the field to `Config`**
|
||
|
||
In `src/config.hpp`, after the `plots` member:
|
||
|
||
```cpp
|
||
/// Take the plot's time from the packet counter and the declared sample rate rather than
|
||
/// from the stream's own time signal, so arrival and producer-clock jitter cannot move a
|
||
/// sample's x. Absolute stream time is still recorded, for export only (`todo.md` item 2).
|
||
bool relative_time = false;
|
||
```
|
||
|
||
- [ ] **Step 2: Add the usage line**
|
||
|
||
In `src/config.cpp:print_usage`, between the `--plots` block and the `-h, --help` line:
|
||
|
||
```c
|
||
" --relative-time take the plot's time from the packet counter and the\n"
|
||
" declared rate, not the stream's time signal; absolute\n"
|
||
" time is then kept for export only\n"
|
||
```
|
||
|
||
- [ ] **Step 3: Add the parse branch**
|
||
|
||
In `src/config.cpp:parse_args`, after the `--plots` branch and before the `else { fail(...) }`:
|
||
|
||
```cpp
|
||
} else if (a == "--relative-time") {
|
||
c.relative_time = true;
|
||
```
|
||
|
||
- [ ] **Step 4: Build and read the help**
|
||
|
||
```bash
|
||
cmake --build build && ./build/udpscope --help | grep -A2 relative-time
|
||
```
|
||
|
||
Expected: the three usage lines above. Warning count unchanged (two on this incremental build).
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add src/config.hpp src/config.cpp
|
||
git commit -m "feat: --relative-time selects the counter-derived clock"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 2: Counter-derived stamps
|
||
|
||
**Files:**
|
||
- Modify: `src/acquisition.hpp:92-93` (state), `src/acquisition.cpp:128-382` (`build_stamps`)
|
||
- Test: `UDPSCOPE_PROBE_RESEQ_MS` is Task 8's; this task is verified by the existing exit dump plus a one-off stderr line.
|
||
|
||
**Interfaces:**
|
||
- Consumes: `Config::relative_time` (Task 1).
|
||
- Produces: `Acquisition::slot_base_` (`std::uint64_t`), `Acquisition::batch_` (`std::uint32_t`, 0 until the first frame), `Acquisition::have_counter_` (`bool`), `Acquisition::last_counter_` (`std::uint32_t`). Task 6 reads `batch_` to size a hole and `slot_base_` is advanced there.
|
||
|
||
- [ ] **Step 1: Add the state**
|
||
|
||
In `src/acquisition.hpp`, after `int time_source_ = 0;`:
|
||
|
||
```cpp
|
||
/// Relative-time mode's position counter: samples seen since the stream began, which is
|
||
/// what the counter names. Advanced by `batch_` for every packet the counter reports
|
||
/// missing (before the frame's own samples) and then by the frame's own `n`, so it is
|
||
/// exactly `counter * batch + sample_index` for a constant batch and degrades gracefully
|
||
/// when a frame's length disagrees.
|
||
std::uint64_t slot_base_ = 0;
|
||
/// Samples per DATA packet, fixed by the first frame. Only ever needed to size what is
|
||
/// *absent*: a delivered frame carries its own length. 0 until the first frame.
|
||
std::uint32_t batch_ = 0;
|
||
/// The previous frame's counter, for the signed-delta ordering the client also uses.
|
||
std::uint32_t last_counter_ = 0;
|
||
bool have_counter_ = false;
|
||
```
|
||
|
||
- [ ] **Step 2: Branch `build_stamps()` on the mode**
|
||
|
||
In `src/acquisition.cpp`, immediately after `stamps_.assign(n, 0.0);` (line 137) and before the `double period = ...` block, insert:
|
||
|
||
```cpp
|
||
// Relative mode: the stamp is a pure function of position and the *declared* rate. Nothing
|
||
// below this block runs -- not the arrival-cadence override, not the degeneracy guard, not
|
||
// the glitch bridge, not the monotonic clamp's reason for existing -- because a position
|
||
// cannot jitter, cannot fail to advance and cannot jump.
|
||
if (cfg_.relative_time) {
|
||
if (batch_ == 0) {
|
||
batch_ = n;
|
||
std::fprintf(stderr,
|
||
"[udpscope] relative time: batch=%u samples/packet, rate=%.6g S/s declared\n",
|
||
batch_, model_.partition().base_dt > 0.0 ? 1.0 / model_.partition().base_dt
|
||
: 0.0);
|
||
std::fflush(stderr);
|
||
} else if (n != batch_) {
|
||
// A frame whose length disagrees with the first one is an inconsistency in the
|
||
// producer, not a loss: the counter still names one packet. Warn once and keep the
|
||
// established batch, which is only used to size holes -- the timeline itself is built
|
||
// from the lengths actually seen, so it stays exact for this frame either way.
|
||
static bool warned_batch = false;
|
||
if (!warned_batch) {
|
||
warned_batch = true;
|
||
std::fprintf(stderr,
|
||
"[udpscope] warning: packet carries %u samples but the stream's first "
|
||
"carried %u; keeping %u as the packet length a lost packet is worth.\n",
|
||
n, batch_, batch_);
|
||
std::fflush(stderr);
|
||
}
|
||
}
|
||
double dt = model_.partition().base_dt;
|
||
if (!(dt > 0.0)) dt = 1.0e-3;
|
||
for (std::uint32_t k = 0; k < n; ++k) {
|
||
stamps_[k] = double(slot_base_ + k) * dt;
|
||
}
|
||
slot_base_ += n;
|
||
return n;
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 3: Build and run against the mock, clean stream**
|
||
|
||
```bash
|
||
cat > /tmp/udps_rel.sh <<'EOF'
|
||
#!/bin/bash
|
||
cd /home/martino/Projects/tools/udpscope
|
||
pkill -f mock_udpstreamer.py; sleep 0.5
|
||
nohup python3 tools/mock_udpstreamer.py --port 8130 --rate 100000 --batch 100 --freq 5 \
|
||
> /tmp/mock_rel.log 2>&1 &
|
||
m=$!; sleep 1
|
||
xvfb-run -a env UDPSCOPE_EXIT_AFTER_MS=8000 UDPSCOPE_ENABLE_ALL_MS=600 UDPSCOPE_EXIT_DUMP=1 \
|
||
./build/udpscope --server 127.0.0.1 --port 8130 --relative-time > /tmp/fault_rel.log 2>&1
|
||
kill $m 2>/dev/null
|
||
grep -E "^\[udpscope\] (frames=|relative time:)" /tmp/fault_rel.log
|
||
echo "criticals=$(grep -c Gtk-CRITICAL /tmp/fault_rel.log) warnings=$(grep -c 'warning:' /tmp/fault_rel.log)"
|
||
EOF
|
||
chmod +x /tmp/udps_rel.sh && /tmp/udps_rel.sh
|
||
```
|
||
|
||
Expected: a `relative time: batch=100 samples/packet` line, `frames=` near 1029 with `gaps=0`, `criticals=0 warnings=0`. A `does not advance across packets` or `jumped implausibly far ahead` warning means the branch was placed after the guard instead of before it.
|
||
|
||
- [ ] **Step 4: Confirm the mode is actually off by default**
|
||
|
||
```bash
|
||
sed -i 's/--relative-time //' /tmp/udps_rel.sh && /tmp/udps_rel.sh
|
||
```
|
||
|
||
Expected: no `relative time:` line at all, `frames=` and `gaps=0` as before. Then restore the flag in the script.
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add src/acquisition.hpp src/acquisition.cpp
|
||
git commit -m "feat: relative mode stamps a sample from its position and the declared rate"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 3: An absolute-time ring for export
|
||
|
||
**Files:**
|
||
- Modify: `src/model.hpp:118-152` (`BufferStore`), `src/model.cpp:140-158`, `src/acquisition.cpp:562-568` (the time push)
|
||
- Test: an export in relative mode carries absolute stamps
|
||
|
||
**Interfaces:**
|
||
- Produces: `BufferStore::abs_time()` → `plot::data::RingBuffer<f64>&`, same capacity and same group-mean cadence as `time()`. Written only in relative mode; read only by the export path.
|
||
|
||
- [ ] **Step 1: Add the ring to `BufferStore`**
|
||
|
||
In `src/model.hpp`, after the `time()` accessors:
|
||
|
||
```cpp
|
||
/// The stream's **absolute** clock, parallel to `time()` and on the same slots.
|
||
///
|
||
/// Relative mode takes `time()` for the counter-derived timeline, and there is exactly one
|
||
/// time ring per store shared by every channel -- so without this ring, selecting relative
|
||
/// mode would silently drop absolute time from an export, which is the one output
|
||
/// `todo.md` item 2 says must keep it. Written only in relative mode (in stream mode
|
||
/// `time()` already *is* the absolute clock) and read only by the export.
|
||
plot::data::RingBuffer<f64>& abs_time() noexcept { return *abs_time_; }
|
||
const plot::data::RingBuffer<f64>& abs_time() const noexcept { return *abs_time_; }
|
||
```
|
||
|
||
and the member, after `time_`:
|
||
|
||
```cpp
|
||
std::shared_ptr<plot::data::RingBuffer<f64>> abs_time_;
|
||
```
|
||
|
||
- [ ] **Step 2: Allocate it alongside the time ring**
|
||
|
||
In `src/model.cpp`, in the `BufferStore` constructor, immediately after the line that constructs `time_`:
|
||
|
||
```cpp
|
||
// Same capacity as the clock it parallels: the two are written on the same slots, so a
|
||
// smaller ring would make an export's stamps run out before its values do.
|
||
abs_time_ = std::make_shared<plot::data::RingBuffer<f64>>(part.time_capacity);
|
||
```
|
||
|
||
- [ ] **Step 3: Push the absolute stamp on the same slot**
|
||
|
||
In `src/acquisition.cpp`, replace the time-push block at lines 562-568 with:
|
||
|
||
```cpp
|
||
time_acc_ += rel;
|
||
abs_time_acc_ += abs_stamps_[k] - epoch;
|
||
++time_acc_n_;
|
||
if (tick % g == g - 1) {
|
||
store.time().push(time_acc_ / static_cast<double>(time_acc_n_));
|
||
// The absolute clock on the same slot and under the same group mean, so an export can
|
||
// pair value, relative stamp and absolute stamp by index with no interpolation.
|
||
store.abs_time().push(abs_time_acc_ / static_cast<double>(time_acc_n_));
|
||
time_acc_ = 0.0;
|
||
abs_time_acc_ = 0.0;
|
||
time_acc_n_ = 0;
|
||
}
|
||
```
|
||
|
||
Add `double abs_time_acc_ = 0.0;` next to `time_acc_` in `src/acquisition.hpp`, reset it in the re-plan block at `src/acquisition.cpp:443`, and add `std::vector<double> abs_stamps_;` next to `stamps_`.
|
||
|
||
- [ ] **Step 4: Fill `abs_stamps_`**
|
||
|
||
In `build_stamps()`, `abs_stamps_` must always carry the stream's own clock regardless of mode. Change `stamps_.assign(n, 0.0);` (line 137) to:
|
||
|
||
```cpp
|
||
stamps_.assign(n, 0.0);
|
||
abs_stamps_.assign(n, 0.0);
|
||
```
|
||
|
||
In the relative branch added by Task 2, before `return n;`, fill it from the declared time signal when there is one and from the relative stamp otherwise:
|
||
|
||
```cpp
|
||
// The absolute clock is still read, for export only. When the stream carries no usable
|
||
// time signal there is no absolute clock to record and the relative stamp is the honest
|
||
// answer -- an export then says what the counter says, which is all anyone had.
|
||
if (time_signal >= 0 && time_signal < static_cast<int>(frame->num_signals)) {
|
||
const udps_signal_t& ts = frame->signals[time_signal];
|
||
const bool scalar_time = udps_signal_num_elements(&ts) == 1;
|
||
for (std::uint32_t k = 0; k < n; ++k) {
|
||
abs_stamps_[k] =
|
||
scalar_time
|
||
? udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), k, 0)
|
||
: udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), 0, k);
|
||
}
|
||
} else {
|
||
for (std::uint32_t k = 0; k < n; ++k) abs_stamps_[k] = stamps_[k];
|
||
}
|
||
```
|
||
|
||
At the end of the stream-mode path, just before `return n;` at line 381, add:
|
||
|
||
```cpp
|
||
// Stream mode: `time()` already holds the absolute clock, so the parallel ring repeats it
|
||
// rather than carrying a second timeline. One rule for the export either way.
|
||
for (std::uint32_t k = 0; k < n; ++k) abs_stamps_[k] = stamps_[k];
|
||
```
|
||
|
||
- [ ] **Step 5: Make the export emit it**
|
||
|
||
Find the export's column assembly and add an `abs_time` column, read from `store.abs_time()` on the same index as the relative stamp:
|
||
|
||
```bash
|
||
grep -n "npy\|csv\|export" src/export.cpp | head -20
|
||
```
|
||
|
||
The column is only added when it differs from the relative stamp — i.e. in relative mode — so a stream-mode export is byte-identical to today's.
|
||
|
||
- [ ] **Step 6: Build, export in both modes, compare**
|
||
|
||
```bash
|
||
cmake --build build
|
||
cat > /tmp/udps_exp.sh <<'EOF'
|
||
#!/bin/bash
|
||
cd /home/martino/Projects/tools/udpscope
|
||
run() {
|
||
pkill -f mock_udpstreamer.py; sleep 0.5
|
||
nohup python3 tools/mock_udpstreamer.py --port 8131 --rate 100000 --batch 100 --freq 5 \
|
||
> /tmp/mock_exp.log 2>&1 &
|
||
local m=$!; sleep 1
|
||
xvfb-run -a env UDPSCOPE_EXIT_AFTER_MS=6000 UDPSCOPE_ENABLE_ALL_MS=600 \
|
||
UDPSCOPE_EXPORT_PATH="/tmp/exp_$1.csv" \
|
||
./build/udpscope --server 127.0.0.1 --port 8131 $2 > "/tmp/exp_$1.log" 2>&1
|
||
kill $m 2>/dev/null
|
||
echo "=== $1"; head -2 "/tmp/exp_$1.csv"
|
||
echo "criticals=$(grep -c Gtk-CRITICAL /tmp/exp_$1.log)"
|
||
}
|
||
run stream ""
|
||
run rel "--relative-time"
|
||
EOF
|
||
chmod +x /tmp/udps_exp.sh && /tmp/udps_exp.sh
|
||
```
|
||
|
||
Expected: `exp_stream.csv`'s header is unchanged from before this task; `exp_rel.csv` carries an extra absolute-time column whose first value is a stream timestamp, not a small multiple of the sample period. `criticals=0` for both.
|
||
|
||
- [ ] **Step 7: Commit**
|
||
|
||
```bash
|
||
git add src/model.hpp src/model.cpp src/acquisition.hpp src/acquisition.cpp src/export.cpp
|
||
git commit -m "feat: relative mode keeps the stream's absolute clock for the export"
|
||
```
|
||
|
||
---
|
||
|
||
## Phase 2 — Client resequencing
|
||
|
||
### Task 4: A reorder ring of reassembled payloads
|
||
|
||
**Files:**
|
||
- Modify: `deps/udpsclient/udps_client.h:170-184` (stats), `:218-229` (config)
|
||
- Modify: `deps/udpsclient/udps_client.c:982` (defaults), `:504-514` (the ordering), `:595-599` (the hook), `:1145-1152` (the deadline clamp), `:1222` (the flush)
|
||
- Test: the mock's `--reorder-every`, measured through udpscope's `gaps`
|
||
|
||
**Interfaces:**
|
||
- Produces: `udps_client_config_t::reorder_window_s` (`double`, default 0.020; 0 disables), `udps_stats_t::late_resequenced` and `udps_stats_t::late_delivered` (`uint64_t`), and `udps_frame_t::late` (`uint8_t`, 1 when the frame's counter is behind the delivered sequence). Tasks 6 and 7 read all three.
|
||
|
||
- [ ] **Step 1: Extend the public structs**
|
||
|
||
In `deps/udpsclient/udps_client.h`, in `udps_stats_t` after `stale_packets`:
|
||
|
||
```c
|
||
/**
|
||
* DATA packets that arrived out of order and were put *back* in order inside
|
||
* the reorder window, so the consumer never saw a gap for them.
|
||
*/
|
||
uint64_t late_resequenced;
|
||
/**
|
||
* DATA packets delivered behind the sequence because the reorder window had
|
||
* already closed on them. The consumer has a hole for these and may fill it;
|
||
* before the window existed they were counted in @c stale_packets and dropped.
|
||
*/
|
||
uint64_t late_delivered;
|
||
```
|
||
|
||
In `udps_client_config_t` after `max_packet_bytes`:
|
||
|
||
```c
|
||
/**
|
||
* How long a packet may be held waiting for an earlier counter, in seconds.
|
||
*
|
||
* A *contiguous* arrival is never held, so this is the latency a gap costs and
|
||
* not a latency the stream pays. 0 disables resequencing entirely and restores
|
||
* the drop-on-late behaviour. Default 0.020.
|
||
*/
|
||
double reorder_window_s;
|
||
```
|
||
|
||
In `udps_frame_t` after `lost`:
|
||
|
||
```c
|
||
/**
|
||
* 1 when this frame's counter is behind the sequence already delivered: it lost
|
||
* the race past the reorder window and the consumer has a hole where it belongs.
|
||
* @c lost is 0 for such a frame -- it is not missing anything, it *is* what was
|
||
* missing -- so a consumer sizes its backfill from @c counter, not from @c lost.
|
||
*/
|
||
uint8_t late;
|
||
```
|
||
|
||
- [ ] **Step 2: Default the window**
|
||
|
||
In `deps/udpsclient/udps_client.c:udps_client_config_init`, beside `cfg->recv_buffer_bytes`:
|
||
|
||
```c
|
||
cfg->reorder_window_s = 0.020;
|
||
```
|
||
|
||
- [ ] **Step 3: Add the ring to the client struct**
|
||
|
||
Near the client's other buffers:
|
||
|
||
```c
|
||
/** Datagrams held waiting for an earlier counter. */
|
||
#define UDPS_REORDER_SLOTS 64
|
||
/** Total bytes the reorder ring may hold, so a fast stream cannot grow it without bound. */
|
||
#define UDPS_REORDER_BYTES (4u * 1024u * 1024u)
|
||
|
||
typedef struct {
|
||
uint8_t *buf; /**< owned; grown on demand to the payload's length */
|
||
size_t cap;
|
||
size_t len;
|
||
uint32_t counter;
|
||
double recv_time;
|
||
int used;
|
||
} udps_held_t;
|
||
```
|
||
|
||
and in `struct udps_client`:
|
||
|
||
```c
|
||
udps_held_t held[UDPS_REORDER_SLOTS];
|
||
size_t held_bytes; /**< payload bytes currently held */
|
||
int held_count;
|
||
uint32_t next_counter; /**< the counter the sequence is waiting for */
|
||
double gap_deadline; /**< wall time the current gap gives up at; 0 = no gap */
|
||
```
|
||
|
||
Free every `held[i].buf` in `udps_client_destroy`, and reset `held_count`, `held_bytes`, `gap_deadline` and `next_counter` wherever `slot_reset_all()` is called (a reconnect starts a new sequence).
|
||
|
||
- [ ] **Step 4: Replace the ordering test with the reorder policy**
|
||
|
||
`deps/udpsclient/udps_client.c:504-514` becomes a *delivery* decision taken before decode, in the wrapper at `:595-599`. Rename the existing `decode_data()` body's ordering block away and put this in the wrapper:
|
||
|
||
```c
|
||
/* Orders the sequence before spending anything on the payload.
|
||
*
|
||
* The counter is a wrapping uint32, so it is ordered by the signed difference;
|
||
* comparing the values directly would call the first packet after the wrap stale
|
||
* and reject the stream from then on.
|
||
*
|
||
* Three cases. A packet that *continues* the sequence is decoded now and costs no
|
||
* latency at all -- that is the common case and it must stay free. A packet that
|
||
* runs ahead of the sequence means an earlier one is missing, so it is held and the
|
||
* gap is given `reorder_window_s` to resolve; a reorder on the wire resolves in
|
||
* 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
|
||
* is delivered flagged rather than discarded, which is the whole point -- a dropped
|
||
* packet cannot fill the hole it left. */
|
||
static int order_data(udps_client_t *c, const uint8_t *pl, size_t len,
|
||
uint32_t counter, double recv_time) {
|
||
int32_t delta;
|
||
|
||
if (c->cfg.reorder_window_s <= 0.0) {
|
||
return deliver_data(c, pl, len, counter, recv_time, 0);
|
||
}
|
||
if (!c->have_counter) {
|
||
c->have_counter = 1;
|
||
c->next_counter = counter + 1u;
|
||
return deliver_data(c, pl, len, counter, recv_time, 0);
|
||
}
|
||
delta = (int32_t)(counter - c->next_counter);
|
||
if (delta == 0) {
|
||
int rc = deliver_data(c, pl, len, counter, recv_time, 0);
|
||
c->next_counter = counter + 1u;
|
||
if (rc == 0) rc = drain_held(c);
|
||
return rc;
|
||
}
|
||
if (delta > 0) {
|
||
if (hold_packet(c, pl, len, counter, recv_time) != 0) {
|
||
/* No room: commit the gap now rather than growing. The counter is
|
||
* visible, so an undersized window is a reading and not a mystery. */
|
||
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, 1);
|
||
}
|
||
```
|
||
|
||
`deliver_data()` is the old `decode_data()` with its ordering block removed, a `late` parameter, and `frame.late = late; frame.lost = late ? 0u : lost;`. `lost` is computed in `drain_held`/`commit_gap` as the number of counters skipped, and is 0 for a contiguous delivery.
|
||
|
||
`hold_packet()` finds a free slot, grows its buffer to `len`, copies the payload, and refuses when `held_count == UDPS_REORDER_SLOTS` or `held_bytes + len > UDPS_REORDER_BYTES`.
|
||
|
||
`drain_held()` repeatedly delivers the held entry whose counter equals `next_counter`, incrementing `stats.late_resequenced` for each, until none matches; it clears `gap_deadline` when `held_count` reaches 0.
|
||
|
||
`commit_gap()` finds the lowest held counter, sets `lost = (uint32_t)(lowest - next_counter)`, advances `next_counter = lowest`, and calls `drain_held()`. `gap_deadline` is cleared by that drain.
|
||
|
||
- [ ] **Step 5: Serve the deadline**
|
||
|
||
In `udps_client_poll`, beside the existing clamps at `:1145-1152`:
|
||
|
||
```c
|
||
/* A held packet must flush on time even when nothing further arrives: at 10 pkt/s
|
||
* the next arrival is 100 ms away, which is five times the whole budget. */
|
||
if (c->gap_deadline > 0.0) {
|
||
double left = c->gap_deadline - now_wall();
|
||
if (left < 0.0) left = 0.0;
|
||
if (left < budget) budget = left;
|
||
}
|
||
```
|
||
|
||
and after `gc_slots(c, now_wall());` at `:1222`:
|
||
|
||
```c
|
||
if (c->gap_deadline > 0.0 && now_wall() >= c->gap_deadline) {
|
||
(void)commit_gap(c);
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 6: Build and verify a reorder no longer costs a gap**
|
||
|
||
```bash
|
||
cmake --build build
|
||
cat > /tmp/udps_reseq.sh <<'EOF'
|
||
#!/bin/bash
|
||
cd /home/martino/Projects/tools/udpscope
|
||
run() {
|
||
local tag="$1"; shift
|
||
pkill -f mock_udpstreamer.py; sleep 0.5
|
||
nohup python3 tools/mock_udpstreamer.py --port 8132 --rate 100000 --batch 100 --freq 5 \
|
||
"$@" > "/tmp/mock_$tag.log" 2>&1 &
|
||
local m=$!; sleep 1
|
||
xvfb-run -a env UDPSCOPE_EXIT_AFTER_MS=8000 UDPSCOPE_ENABLE_ALL_MS=600 UDPSCOPE_EXIT_DUMP=1 \
|
||
./build/udpscope --server 127.0.0.1 --port 8132 > "/tmp/fault_$tag.log" 2>&1
|
||
kill $m 2>/dev/null
|
||
echo "=== $tag : $*"
|
||
grep -E "^\[udpscope\] frames=" "/tmp/fault_$tag.log"
|
||
echo "criticals=$(grep -c Gtk-CRITICAL /tmp/fault_$tag.log)"
|
||
}
|
||
run clean
|
||
EOF
|
||
chmod +x /tmp/udps_reseq.sh && /tmp/udps_reseq.sh
|
||
```
|
||
|
||
Expected for `clean`: `frames=` near 1029, `gaps=0`, `criticals=0` — unchanged, proving the zero-latency path is still free. Then run the reorder case alone:
|
||
|
||
```bash
|
||
sed -i 's/^run clean$/run reorder --reorder-every 10/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh
|
||
```
|
||
|
||
Expected: `gaps=0` where it previously measured **102**, and `frames` near 1029 rather than 925. That delta is the whole feature: 102 gaps became 0.
|
||
|
||
- [ ] **Step 7: Verify the kill switch**
|
||
|
||
Set `c.reorder_window_s = 0.0;` temporarily in `src/acquisition.cpp:start()`, rebuild, re-run the reorder case. Expected: `gaps=102`, `frames≈925` — exactly today's numbers. Revert the line.
|
||
|
||
- [ ] **Step 8: Commit**
|
||
|
||
```bash
|
||
git add deps/udpsclient/udps_client.h deps/udpsclient/udps_client.c
|
||
git commit -m "feat: hold a packet that runs ahead of the sequence, and deliver a late one flagged"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 5: `--reorder-delay-ms` in the mock
|
||
|
||
**Files:**
|
||
- Modify: `tools/mock_udpstreamer.py` (argparse, the send policy, the `done` line)
|
||
- Test: the option's own effect, read through udpscope's counters
|
||
|
||
**Interfaces:**
|
||
- Produces: `--reorder-delay-ms MS` — a withheld packet is released `MS` milliseconds later instead of immediately after the packet that overtook it. Task 8's probe needs a value past 20 ms to reach the backfill path.
|
||
|
||
- [ ] **Step 1: Add the option**
|
||
|
||
Beside `--reorder-every` in the argparse block:
|
||
|
||
```python
|
||
p.add_argument(
|
||
"--reorder-delay-ms",
|
||
type=float,
|
||
default=0.0,
|
||
metavar="MS",
|
||
help="hold a withheld packet this many milliseconds instead of releasing it "
|
||
"immediately behind the packet that overtook it. Under the receiver's reorder "
|
||
"window the packet is resequenced and no hole is ever seen; past it the hole is "
|
||
"committed first and the packet has to backfill it -- which is the only way to "
|
||
"reach the backfill path at all, since a 0 ms delay always wins the race.",
|
||
)
|
||
```
|
||
|
||
- [ ] **Step 2: Release on a deadline instead of on the next send**
|
||
|
||
`held` becomes `(counter, pkt, release_at)`. Replace the `if held is not None:` release block in the `else:` branch with a time test evaluated every iteration, before the send:
|
||
|
||
```python
|
||
if held is not None and now >= held[2]:
|
||
# Out of order on the wire now. Within the receiver's reorder window this is put
|
||
# back in order and costs nothing; past it the receiver has already committed a
|
||
# hole and this packet is what fills it.
|
||
sock.sendto(held[1], peer)
|
||
inverted += 1
|
||
if not noted_reorder:
|
||
noted_reorder = True
|
||
print(
|
||
"mock udpstreamer: sent packet %d before %d, releasing it %.1f ms late"
|
||
% (counter, held[0], args.reorder_delay_ms),
|
||
flush=True,
|
||
)
|
||
held = None
|
||
```
|
||
|
||
and where the packet is withheld:
|
||
|
||
```python
|
||
held = (counter, pkt, now + args.reorder_delay_ms / 1000.0)
|
||
```
|
||
|
||
- [ ] **Step 3: Check it compiles and sends**
|
||
|
||
```bash
|
||
python3 -m py_compile tools/mock_udpstreamer.py && echo OK
|
||
cat > /tmp/udps_mock.sh <<'EOF'
|
||
#!/bin/bash
|
||
cd /home/martino/Projects/tools/udpscope
|
||
pkill -f mock_udpstreamer.py; sleep 0.5
|
||
timeout 4 python3 tools/mock_udpstreamer.py --port 8133 --rate 100000 --batch 100 \
|
||
--reorder-every 10 --reorder-delay-ms 50 2>&1 | grep -E "before|done"
|
||
EOF
|
||
chmod +x /tmp/udps_mock.sh && /tmp/udps_mock.sh
|
||
```
|
||
|
||
Expected: a `sent packet N before M, releasing it 50.0 ms late` line and a `done (... inverted)` line with a non-zero inverted count. (`timeout` makes the mock exit without a CONNECT, so a `done` line may be absent; the `before` line is the assertion.)
|
||
|
||
- [ ] **Step 4: Commit**
|
||
|
||
```bash
|
||
git add tools/mock_udpstreamer.py
|
||
git commit -m "test: the mock can release a withheld packet past the reorder window"
|
||
```
|
||
|
||
---
|
||
|
||
## Phase 3 — Hole reservation and surfacing
|
||
|
||
### Task 6: Reserve an invalidated hole
|
||
|
||
**Files:**
|
||
- Modify: `src/acquisition.cpp:384-416` (before the per-sample loop), `src/acquisition.cpp:553` (the loop)
|
||
- Modify: `src/model.hpp` (the invalid-span table)
|
||
- Test: the mock's `--drop-every`, read through the exit dump
|
||
|
||
**Interfaces:**
|
||
- Consumes: `udps_frame_t::lost`, `udps_frame_t::late`, `Acquisition::batch_` (Task 2), `kGlitchPackets`.
|
||
- Produces: `ScopeModel::note_hole(f64 x_begin, f64 x_end, u64 slot_begin, u64 slot_count, std::uint32_t counter_first)` and `ScopeModel::holes()` → `const std::vector<Hole>&` (under `mutex()`). Task 7 counts them, Task 10 addresses them, Task 11 draws them.
|
||
|
||
- [ ] **Step 1: Add the hole table**
|
||
|
||
In `src/model.hpp`, before `StreamStats`:
|
||
|
||
```cpp
|
||
/// One reserved, invalidated span: samples the counter says exist and the stream has not
|
||
/// delivered. Its stored slots hold NaN, so a reader needs no special case -- the table is
|
||
/// what lets a *late* packet find the slots it belongs to, and what the band draws.
|
||
struct Hole {
|
||
f64 x_begin = 0.0; ///< stream seconds of the hole's first slot
|
||
f64 x_end = 0.0; ///< stream seconds of the hole's last slot
|
||
u64 slot_begin = 0; ///< absolute stored-slot index of the first slot
|
||
u64 slot_count = 0; ///< stored slots the hole spans
|
||
u64 tick_begin = 0; ///< absolute base tick of the hole's first raw sample
|
||
std::uint32_t counter_first = 0; ///< counter of the first missing packet
|
||
std::uint32_t packets = 0; ///< packets the hole is worth
|
||
};
|
||
```
|
||
|
||
and on `ScopeModel`, with the table capped so a pathological stream cannot grow it:
|
||
|
||
```cpp
|
||
/// Records a reserved hole. Caller holds `mutex()` (the producer already does). The table
|
||
/// is capped at `kMaxHoles`; past that the oldest is dropped, because an unbounded list of
|
||
/// gaps is a leak and the oldest gap is the one furthest from being backfilled.
|
||
static constexpr std::size_t kMaxHoles = 64;
|
||
void note_hole(const Hole& h);
|
||
const std::vector<Hole>& holes() const noexcept { return holes_; }
|
||
/// Drops holes whose slots the ring has already discarded.
|
||
void forget_holes_before(u64 slot);
|
||
```
|
||
|
||
- [ ] **Step 2: Reserve before the frame's own samples**
|
||
|
||
In `src/acquisition.cpp`, after `const u64 g = part.time_stride;` (line 433) and before the accumulator re-plan block, insert:
|
||
|
||
```cpp
|
||
// The gap the counter reports, reserved as NaN before this frame's own samples are
|
||
// stamped: the hole belongs *ahead* of the data that revealed it. Running the missing
|
||
// samples through the same per-sample loop below is the whole trick -- `tick_` advances
|
||
// so the decimation phase does not shift, the group mean lands on the mean stamp, and
|
||
// `MinMaxPyramid::append` sees NaN, which its non-finite contract already stores as an
|
||
// empty aggregate every reader treats as absent.
|
||
//
|
||
// Capped at `kGlitchPackets`: 256 packet spans in one go is an outage, not loss, and
|
||
// reserving one would push the entire retained history out of the ring to make room for
|
||
// samples that are not coming.
|
||
const std::uint32_t missing = frame->late ? 0u : frame->lost;
|
||
if (missing > 0 && batch_ > 0 && double(missing) <= kGlitchPackets) {
|
||
reserve_hole(missing, epoch, part, channels, store, g);
|
||
} else if (missing > 0) {
|
||
static bool warned_outage = false;
|
||
if (!warned_outage) {
|
||
warned_outage = true;
|
||
std::fprintf(stderr,
|
||
"[udpscope] warning: %u packets missing in one go, past the %g-packet "
|
||
"reservation cap; treating it as an outage and carrying on rather than "
|
||
"reserving %llu samples of hole.\n",
|
||
missing, kGlitchPackets,
|
||
static_cast<unsigned long long>(std::uint64_t(missing) * batch_));
|
||
std::fflush(stderr);
|
||
}
|
||
}
|
||
```
|
||
|
||
`kGlitchPackets` moves from a function-local `constexpr` in `build_stamps()` to a file-scope one in the anonymous namespace so both call sites read the same constant.
|
||
|
||
- [ ] **Step 3: Write `reserve_hole()`**
|
||
|
||
A new private member of `Acquisition`, factored out so `on_data` stays readable. It is a reduced copy of the per-sample loop that pushes NaN instead of a sample and skips the trigger and the math channels:
|
||
|
||
```cpp
|
||
void Acquisition::reserve_hole(std::uint32_t missing, double epoch, const Partition& part,
|
||
const std::vector<ChannelInfo>& channels, BufferStore& store,
|
||
u64 g) {
|
||
const u64 raw = u64(missing) * u64(batch_);
|
||
Hole h;
|
||
h.tick_begin = tick_;
|
||
h.slot_begin = store.values(0).produced();
|
||
h.counter_first = last_counter_ + 1u;
|
||
h.packets = missing;
|
||
|
||
double dt = part.base_dt;
|
||
if (!(dt > 0.0)) dt = 1.0e-3;
|
||
for (u64 j = 0; j < raw; ++j) {
|
||
const u64 tick = tick_++;
|
||
// The hole is stamped at the declared cadence from the last good stamp -- exact in
|
||
// relative mode, and in stream mode the same continuation the glitch bridge makes.
|
||
// It must be *stamped* rather than left short: the time ring is binary-searched by
|
||
// `ChannelBuffer::neighbours()`, so it has to stay monotone and dense, and backfill
|
||
// then changes values only and never a stamp.
|
||
const double rel = last_stamp_ + dt;
|
||
last_stamp_ = rel;
|
||
if (j == 0) h.x_begin = rel;
|
||
h.x_end = rel;
|
||
|
||
time_acc_ += rel;
|
||
abs_time_acc_ += rel;
|
||
++time_acc_n_;
|
||
if (tick % g == g - 1) {
|
||
store.time().push(time_acc_ / double(time_acc_n_));
|
||
store.abs_time().push(abs_time_acc_ / double(time_acc_n_));
|
||
time_acc_ = 0.0;
|
||
abs_time_acc_ = 0.0;
|
||
time_acc_n_ = 0;
|
||
}
|
||
for (std::size_t i = 0; i < channels.size(); ++i) {
|
||
if (channels[i].math) continue;
|
||
// A NaN in the accumulator makes the whole group NaN, which is why a slot straddling
|
||
// a hole edge is NaN too: its real members were averaged into a sum that is now
|
||
// non-finite and were never retained, so the backfill cannot recover it and must not
|
||
// pretend to. One slot per edge, against a hole thousands of slots wide.
|
||
value_acc_[i] += std::numeric_limits<double>::quiet_NaN();
|
||
++value_acc_n_[i];
|
||
const u64 period = g * part.stride[i];
|
||
if (tick % period != period - 1) continue;
|
||
store.values(i).push(static_cast<f32>(value_acc_[i] / double(value_acc_n_[i])));
|
||
value_acc_[i] = 0.0;
|
||
value_acc_n_[i] = 0;
|
||
}
|
||
// The math channels are fed NaN like everything else, so a derived trace keeps the gap
|
||
// its source had. The trigger is *not* fed: a NaN comparison is false, so the engine
|
||
// could not latch here anyway, and feeding it would advance its hold-off clock over
|
||
// time nothing was sampled in.
|
||
if (tick % g == g - 1) {
|
||
std::fill(math_sample_.begin(), math_sample_.end(), std::nanf(""));
|
||
model_.push_math_sample(math_sample_);
|
||
}
|
||
}
|
||
h.slot_count = store.values(0).produced() - h.slot_begin;
|
||
model_.note_hole(h);
|
||
model_.stats().holes.fetch_add(1, std::memory_order_relaxed);
|
||
model_.stats().hole_slots.fetch_add(h.slot_count, std::memory_order_relaxed);
|
||
}
|
||
```
|
||
|
||
Declare it in `src/acquisition.hpp` and add `#include <limits>`.
|
||
|
||
- [ ] **Step 4: Track the counter**
|
||
|
||
At the top of `on_data`, after the `running_` check:
|
||
|
||
```cpp
|
||
// The counter is this frame's identity and the hole's address. It is tracked here rather
|
||
// than read from `frame->lost` alone because a *late* frame's counter is behind the
|
||
// sequence and must not move the mark forward.
|
||
if (!have_counter_ || (std::int32_t)(frame->counter - last_counter_) > 0) {
|
||
last_counter_ = frame->counter;
|
||
have_counter_ = true;
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 5: Build and measure a loss**
|
||
|
||
```bash
|
||
cmake --build build
|
||
sed -i 's/^run reorder.*$/run drop10 --drop-every 10/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh
|
||
```
|
||
|
||
Expected: `gaps=102` (unchanged — the packets really are lost), but `frames` now near **1029** rather than 925, because the hole's samples occupy the slots the lost packets would have. `criticals=0`. The channel lines of the exit dump show `stored=` grown by `102 × 100 / stride`.
|
||
|
||
- [ ] **Step 6: Verify the cap**
|
||
|
||
```bash
|
||
sed -i 's/^run drop10.*$/run burst300 --drop-every 1000 --drop-burst 300/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh
|
||
grep "warning:" /tmp/fault_burst300.log | head -2
|
||
```
|
||
|
||
Expected: the `past the 256-packet reservation cap` warning, and **no** `jumped implausibly far ahead` / `timing samples by arrival instead` pair. Then the same case with `--relative-time` added: the cap warning is still there and the stream stays on its own clock.
|
||
|
||
- [ ] **Step 7: Commit**
|
||
|
||
```bash
|
||
git add src/acquisition.hpp src/acquisition.cpp src/model.hpp src/model.cpp
|
||
git commit -m "fix: a lost packet reserves the samples it owed instead of vanishing from the timeline"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 7: Surface the counters
|
||
|
||
**Files:**
|
||
- Modify: `src/model.hpp:155-162` (`StreamStats`), `src/acquisition.cpp:84-89` (the poll copy)
|
||
- Modify: `src/status_text.cpp:68-80` (the detail tier), `src/ui.cpp:10718-10724` (the exit dump)
|
||
- Test: the counters read non-zero in a reorder run and zero in a clean one
|
||
|
||
**Interfaces:**
|
||
- Consumes: `udps_stats_t::late_resequenced`, `late_delivered`, `stale_packets` (Task 4); `ScopeModel::stats().holes`, `hole_slots` (Task 6).
|
||
- Produces: `StreamStats::resequenced`, `late`, `stale`, `holes`, `hole_slots`, `backfilled` (all `std::atomic<std::uint64_t>`). Task 10 increments `backfilled`.
|
||
|
||
- [ ] **Step 1: Extend `StreamStats`**
|
||
|
||
```cpp
|
||
/// Out-of-order packets put back in order inside the reorder window: no hole was ever
|
||
/// seen for these.
|
||
std::atomic<std::uint64_t> resequenced{0};
|
||
/// Out-of-order packets delivered behind the sequence, past the window.
|
||
std::atomic<std::uint64_t> late{0};
|
||
/// Packets the client still refused outright (a duplicate, or any late packet while
|
||
/// 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`
|
||
/// diverging.
|
||
std::atomic<std::uint64_t> stale{0};
|
||
/// Reserved holes, and the stored slots they cost.
|
||
std::atomic<std::uint64_t> holes{0};
|
||
std::atomic<std::uint64_t> hole_slots{0};
|
||
/// Stored slots a late packet repaired.
|
||
std::atomic<std::uint64_t> backfilled{0};
|
||
```
|
||
|
||
- [ ] **Step 2: Copy them on the poll**
|
||
|
||
In `src/acquisition.cpp:run()`, beside the existing stores:
|
||
|
||
```cpp
|
||
out.resequenced.store(s.late_resequenced, std::memory_order_relaxed);
|
||
out.late.store(s.late_delivered, std::memory_order_relaxed);
|
||
out.stale.store(s.stale_packets, std::memory_order_relaxed);
|
||
```
|
||
|
||
- [ ] **Step 3: Add them to the detail tier**
|
||
|
||
`StatusSnapshot` gains the six fields, filled where `packets`/`frames`/`gaps` are, and `format_detail`'s format string grows one segment after `gaps %llu`:
|
||
|
||
```c
|
||
" | reseq %llu late %llu stale %llu holes %llu (%llu slots, %llu filled)"
|
||
```
|
||
|
||
with the six arguments in that order. The segment is omitted entirely when all six are zero, so a clean stream's status line is unchanged.
|
||
|
||
- [ ] **Step 4: Add them to the exit dump**
|
||
|
||
In `src/ui.cpp`, extend the `frames=... packets=... gaps=...` printf with:
|
||
|
||
```c
|
||
" reseq=%llu late=%llu stale=%llu holes=%llu hole_slots=%llu backfilled=%llu"
|
||
```
|
||
|
||
- [ ] **Step 5: Build and read both cases**
|
||
|
||
```bash
|
||
cmake --build build
|
||
sed -i 's/^run burst300.*$/run clean/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh
|
||
```
|
||
|
||
Expected: `reseq=0 late=0 stale=0 holes=0 hole_slots=0 backfilled=0`.
|
||
|
||
```bash
|
||
sed -i 's/^run clean$/run reorder --reorder-every 10/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh
|
||
```
|
||
|
||
Expected: `reseq` near **102**, `late=0`, `holes=0`, `gaps=0` — every inversion resequenced, nothing lost.
|
||
|
||
```bash
|
||
sed -i 's/^run reorder.*$/run late --reorder-every 10 --reorder-delay-ms 50/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh
|
||
```
|
||
|
||
Expected: `reseq=0`, `late` near **102**, `holes` near 102, `backfilled=0` (Task 10 is what makes it non-zero). `criticals=0`.
|
||
|
||
- [ ] **Step 6: Commit**
|
||
|
||
```bash
|
||
git add src/model.hpp src/acquisition.cpp src/status_text.cpp src/status_text.hpp src/ui.cpp
|
||
git commit -m "feat: the status bar and the exit dump report resequencing, holes and stale packets"
|
||
```
|
||
|
||
---
|
||
|
||
## Phase 4 — Backfill
|
||
|
||
### Task 8: `MinMaxPyramid::amend()`
|
||
|
||
**Files:**
|
||
- Modify: `deps/plotter/include/plot/data/minmax_pyramid.hpp:78` (declare), `deps/plotter/src/data/minmax_pyramid.cpp:180` (define)
|
||
- Test: `deps/plotter/tests/test_minmax_pyramid.cpp`
|
||
|
||
**Interfaces:**
|
||
- Produces: `void MinMaxPyramid::amend(u64 absolute_sample, f32 v)`. Precondition: the sample at that index was appended **non-finite**. A no-op for an index outside `[window_base(), window_end())`. Task 10 calls it.
|
||
|
||
- [ ] **Step 1: Write the failing test**
|
||
|
||
Append to `deps/plotter/tests/test_minmax_pyramid.cpp`:
|
||
|
||
```cpp
|
||
PLOT_TEST(amend_folds_a_backfilled_sample_into_every_covering_level) {
|
||
// A hole is appended as NaN -- which the pyramid already stores as an empty aggregate --
|
||
// and amended later when the packet that owned it finally arrives. The amendment has to
|
||
// reach every level that covers the index, because the renderer reads whichever level the
|
||
// zoom picked: repairing level 0 alone would make a trace whole when zoomed in and still
|
||
// broken when zoomed out.
|
||
MinMaxPyramid p;
|
||
p.reserve(8 * kBucket);
|
||
for (u64 i = 0; i < 8 * kBucket; ++i) {
|
||
// One NaN, at an index comfortably inside the first bucket of every level.
|
||
p.append(i == 5 ? std::numeric_limits<f32>::quiet_NaN() : 1.0f);
|
||
}
|
||
// Before: every covering bucket aggregates kBucket-1 ones and knows nothing of index 5.
|
||
PLOT_CHECK(p.bucket_count(0, 0) == kBucket - 1);
|
||
PLOT_CHECK_NEAR(p.bucket(0, 0).max, 1.0);
|
||
|
||
p.amend(5, 7.0f);
|
||
|
||
// Level 0 gained the sample.
|
||
PLOT_CHECK(p.bucket_count(0, 0) == kBucket);
|
||
PLOT_CHECK_NEAR(p.bucket(0, 0).max, 7.0);
|
||
PLOT_CHECK_NEAR(p.bucket_sum(0, 0), f64(kBucket - 1) + 7.0);
|
||
PLOT_CHECK_NEAR(p.bucket_sumsq(0, 0), f64(kBucket - 1) + 49.0);
|
||
// And so did every coarser level that covers it.
|
||
for (u32 l = 1; l < p.level_count(); ++l) {
|
||
if (p.level_size(l) == 0) continue;
|
||
const u64 span = u64{kBucket} << l;
|
||
const u64 b = 5 / span;
|
||
if (b < p.level_base(l) || b - p.level_base(l) >= p.level_size(l)) continue;
|
||
PLOT_CHECK_NEAR(p.bucket(l, b).max, 7.0);
|
||
PLOT_CHECK(p.bucket_count(l, b) == span);
|
||
}
|
||
}
|
||
|
||
PLOT_TEST(amend_makes_an_all_nan_bucket_present_again) {
|
||
// A bucket whose every sample was non-finite is stored with count 0, which every reader
|
||
// treats as absent. One amendment has to make it present -- min and max become the value
|
||
// rather than widening from a meaningless zero, which is the case a plain fold gets wrong.
|
||
MinMaxPyramid p;
|
||
p.reserve(4 * kBucket);
|
||
const f32 nan = std::numeric_limits<f32>::quiet_NaN();
|
||
for (u64 i = 0; i < 2 * kBucket; ++i) p.append(i < kBucket ? nan : 1.0f);
|
||
PLOT_CHECK(p.bucket_count(0, 0) == 0);
|
||
|
||
p.amend(3, -2.0f);
|
||
|
||
PLOT_CHECK(p.bucket_count(0, 0) == 1);
|
||
PLOT_CHECK_NEAR(p.bucket(0, 0).min, -2.0);
|
||
PLOT_CHECK_NEAR(p.bucket(0, 0).max, -2.0);
|
||
}
|
||
|
||
PLOT_TEST(amend_outside_the_window_is_a_no_op) {
|
||
// An index the window has already dropped, or one not yet appended, has nowhere to go.
|
||
// Silently ignoring it is what lets the caller amend without first proving the ring still
|
||
// holds the slot -- a race it cannot win, because the window slides on the other thread.
|
||
MinMaxPyramid p;
|
||
p.reserve(4 * kBucket);
|
||
for (u64 i = 0; i < 4 * kBucket; ++i) p.append(1.0f);
|
||
p.drop_before(2 * kBucket);
|
||
const u64 before = p.bucket_count(0, 2);
|
||
p.amend(5, 99.0f); // dropped
|
||
p.amend(100 * kBucket, 99.0f); // never appended
|
||
PLOT_CHECK(p.bucket_count(0, 2) == before);
|
||
PLOT_CHECK_NEAR(p.bucket(0, 2).max, 1.0);
|
||
}
|
||
```
|
||
|
||
Add `#include <limits>` to the test file if it is not already there.
|
||
|
||
- [ ] **Step 2: Run the tests to verify they fail**
|
||
|
||
```bash
|
||
cmake --build deps/plotter/build/debug 2>&1 | tail -5
|
||
```
|
||
|
||
Expected: a compile error, `'amend' is not a member of 'plot::data::MinMaxPyramid'`.
|
||
|
||
- [ ] **Step 3: Declare it**
|
||
|
||
In `deps/plotter/include/plot/data/minmax_pyramid.hpp`, after `append`:
|
||
|
||
```cpp
|
||
/// Folds a backfilled sample into every aggregate that covers absolute index
|
||
/// `absolute_sample`.
|
||
///
|
||
/// **Precondition: the sample at that index was appended non-finite.** That is what makes
|
||
/// this purely additive -- min/max widen, sum/sumsq accumulate and count rises by one, at
|
||
/// every covering level -- so no bucket is ever recomputed and the cost is O(levels)
|
||
/// regardless of how wide the repaired span is. Amending a *finite* sample would need the
|
||
/// old value removed from sum/sumsq and could not restore min/max at all, which is why
|
||
/// this is a one-way NaN-to-value operation and not a general write.
|
||
///
|
||
/// An index outside `[window_base(), window_end())` is a no-op: the window slides on the
|
||
/// consumer thread, so a caller cannot prove the slot is still held before asking.
|
||
void amend(u64 absolute_sample, f32 v);
|
||
```
|
||
|
||
- [ ] **Step 4: Define it**
|
||
|
||
In `deps/plotter/src/data/minmax_pyramid.cpp`, after `append()`:
|
||
|
||
```cpp
|
||
void MinMaxPyramid::amend(u64 absolute_sample, f32 v) {
|
||
if (!std::isfinite(v)) return;
|
||
if (absolute_sample < base_ || absolute_sample >= count_) return;
|
||
|
||
const f64 vd = f64(v);
|
||
const auto fold = [&](Entry& e) {
|
||
// An empty bucket's `mm` is meaningless (see the Entry note), so the first finite
|
||
// sample *sets* the extremes instead of widening from a zero that was never a sample.
|
||
if (e.count == 0) {
|
||
e.mm.min = v;
|
||
e.mm.max = v;
|
||
} else {
|
||
e.mm.min = std::min(e.mm.min, v);
|
||
e.mm.max = std::max(e.mm.max, v);
|
||
}
|
||
e.sum += vd;
|
||
e.sumsq += vd * vd;
|
||
++e.count;
|
||
};
|
||
|
||
// The in-progress level-0 bucket, when the index lies inside it. It is not an entry yet,
|
||
// so the loop below cannot reach it; leaving it out would lose the amendment for a hole
|
||
// whose tail is still filling, which is exactly the hole a prompt backfill arrives into.
|
||
if (count_ % kBucket != 0 && absolute_sample >= (count_ / kBucket) * kBucket) {
|
||
fold(pending_);
|
||
}
|
||
|
||
// Every level independently: min/max/sum/sumsq/count are each associative, so folding the
|
||
// value into a parent gives the same aggregate as re-merging its children would.
|
||
for (u32 l = 0; l < static_cast<u32>(levels_.size()); ++l) {
|
||
const u64 i = absolute_sample / level_span(l);
|
||
if (i < levels_[l].base) continue;
|
||
if (i - levels_[l].base >= static_cast<u64>(levels_[l].entries.size())) continue;
|
||
fold(levels_[l].entries[static_cast<usize>(i - levels_[l].base)]);
|
||
}
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 5: Run the tests to verify they pass**
|
||
|
||
```bash
|
||
cmake --build deps/plotter/build/debug && \
|
||
PLOT_TEST_FILTER=amend deps/plotter/build/debug/tests/plot_tests
|
||
```
|
||
|
||
Expected: three tests run, `0 failed`.
|
||
|
||
- [ ] **Step 6: Run the whole suite once**
|
||
|
||
```bash
|
||
xvfb-run -a deps/plotter/build/debug/tests/plot_tests 2>&1 | tail -3
|
||
```
|
||
|
||
Expected: `0 failed`. Run this **once**, in the background if it is slow; do not re-run it to chase a cosmetic difference.
|
||
|
||
- [ ] **Step 7: Commit**
|
||
|
||
```bash
|
||
git add deps/plotter/include/plot/data/minmax_pyramid.hpp \
|
||
deps/plotter/src/data/minmax_pyramid.cpp \
|
||
deps/plotter/tests/test_minmax_pyramid.cpp
|
||
git commit -m "feat(plotter): a non-finite aggregate can be amended additively at every level"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 9: `RingBuffer::amend()`
|
||
|
||
**Files:**
|
||
- Modify: `deps/plotter/include/plot/data/ring_buffer.hpp:65` (after the batch `push`)
|
||
- Test: `deps/plotter/tests/test_ring_buffer.cpp`
|
||
|
||
**Interfaces:**
|
||
- Produces: `bool RingBuffer<T>::amend(u64 absolute, T v)` — `true` when the slot was still retained and written, `false` when the index had aged out or was never produced. Task 10 uses the return to decide whether to queue the pyramid amend.
|
||
|
||
- [ ] **Step 1: Write the failing test**
|
||
|
||
Append to `deps/plotter/tests/test_ring_buffer.cpp`:
|
||
|
||
```cpp
|
||
PLOT_TEST(ring_amend_rewrites_a_retained_slot) {
|
||
// Backfill: a slot pushed as a placeholder is rewritten once the data that belongs there
|
||
// arrives. The index is absolute, because that is the only coordinate a late packet can
|
||
// compute -- a retained-window index would have moved by the time it got here.
|
||
plot::data::RingBuffer<f32> r(8);
|
||
for (int i = 0; i < 6; ++i) r.push(f32(i));
|
||
PLOT_CHECK(r.amend(3, 42.0f));
|
||
PLOT_CHECK_NEAR(r[3], 42.0);
|
||
PLOT_CHECK(r.produced() == 6); // an amend is not a push
|
||
}
|
||
|
||
PLOT_TEST(ring_amend_refuses_an_index_the_ring_no_longer_holds) {
|
||
// The window slides on the producer, so a late amend may arrive after its slot was
|
||
// overwritten. Writing it anyway would corrupt whatever sample now owns that slot, so the
|
||
// bounds test is the whole safety of the operation and the caller reads the result.
|
||
plot::data::RingBuffer<f32> r(8);
|
||
for (int i = 0; i < 20; ++i) r.push(f32(i));
|
||
PLOT_CHECK(r.base() == 12);
|
||
PLOT_CHECK(!r.amend(3, 42.0f)); // aged out
|
||
PLOT_CHECK(!r.amend(20, 42.0f)); // never produced
|
||
PLOT_CHECK_NEAR(r[0], 12.0); // and nothing moved
|
||
PLOT_CHECK_NEAR(r[7], 19.0);
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 2: Run the tests to verify they fail**
|
||
|
||
```bash
|
||
cmake --build deps/plotter/build/debug 2>&1 | tail -5
|
||
```
|
||
|
||
Expected: a compile error, `'amend' is not a member`.
|
||
|
||
- [ ] **Step 3: Implement it**
|
||
|
||
In `deps/plotter/include/plot/data/ring_buffer.hpp`, after the batch `push`:
|
||
|
||
```cpp
|
||
/// Producer thread only: rewrites the slot at absolute index `absolute`, without pushing.
|
||
/// Returns false when that index is not retained -- already overwritten, or not yet
|
||
/// produced -- in which case nothing is written.
|
||
///
|
||
/// This is for *backfill*: a slot pushed as a placeholder and corrected when the data that
|
||
/// belongs there arrives late. It is the same race class as `push()` -- one relaxed store
|
||
/// into a slot a consumer may be reading, and T is lock-free by the class's static_assert,
|
||
/// so a reader sees the old value or the new one and never a torn one. It deliberately does
|
||
/// not touch `produced_`: the count fixes which indices a snapshot reads, and an amend
|
||
/// changes a value inside that window rather than extending it.
|
||
bool amend(u64 absolute, T v) {
|
||
const u64 p = produced_.load(std::memory_order_relaxed);
|
||
if (absolute >= p) return false;
|
||
if (p > cap_ && absolute < p - cap_) return false;
|
||
buf_[static_cast<usize>(absolute & mask_)].store(v, std::memory_order_relaxed);
|
||
return true;
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 4: Run the tests to verify they pass**
|
||
|
||
```bash
|
||
cmake --build deps/plotter/build/debug && \
|
||
PLOT_TEST_FILTER=ring_amend deps/plotter/build/debug/tests/plot_tests
|
||
```
|
||
|
||
Expected: two tests, `0 failed`.
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add deps/plotter/include/plot/data/ring_buffer.hpp deps/plotter/tests/test_ring_buffer.cpp
|
||
git commit -m "feat(plotter): a retained slot can be backfilled without pushing"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 10: Wire the backfill
|
||
|
||
**Files:**
|
||
- Modify: `src/channel.hpp:48` and `src/channel.cpp:32-63` (`acquire()` drains the queue)
|
||
- Modify: `src/model.hpp`, `src/model.cpp` (the queue and `BufferStore::acquire_all`)
|
||
- Modify: `src/acquisition.cpp` (handle a late frame)
|
||
- Test: the `late` run's `backfilled` counter and a NaN-free trace
|
||
|
||
**Interfaces:**
|
||
- Consumes: `RingBuffer::amend` (Task 9), `MinMaxPyramid::amend` (Task 8), `ScopeModel::holes()` (Task 6).
|
||
- Produces: `ChannelBuffer::queue_amend(u64 absolute, f32 v)` (producer side, under `mutex()`) and the drain inside `acquire()`. No new thread primitive: `acquire_all()` runs under `model.mutex()` (`src/ui.cpp:6263-6265`) and the producer holds that same mutex for all of `on_data` (`src/acquisition.cpp:386`), so a plain `std::vector` member is correctly serialised.
|
||
|
||
- [ ] **Step 1: Give `ChannelBuffer` the queue and the drain**
|
||
|
||
In `src/channel.hpp`:
|
||
|
||
```cpp
|
||
/// Producer thread, under `ScopeModel::mutex()`: records that absolute sample `i` has been
|
||
/// backfilled, for `acquire()` to fold into the pyramid.
|
||
///
|
||
/// The pyramid is mutated *only* on the consumer thread -- `acquire()` is called from the
|
||
/// frame path -- and a bucket is 32 bytes and not atomic, so amending it from the producer
|
||
/// would let a reader see a half-updated bucket whose `sum` did not match its `count`: a
|
||
/// wrong mean out of a structure whose readers trust `count == 0` to mean absent. The value
|
||
/// ring is amended by the producer (one lock-free store); the pyramid waits for here.
|
||
void queue_amend(u64 absolute, f32 v) { pending_amends_.push_back({absolute, v}); }
|
||
```
|
||
|
||
with the member:
|
||
|
||
```cpp
|
||
struct Amend {
|
||
u64 absolute;
|
||
f32 value;
|
||
};
|
||
std::vector<Amend> pending_amends_;
|
||
```
|
||
|
||
In `src/channel.cpp:acquire()`, **after** `published_ = produced;` and the `drop_before` call, so every amended index is already in the pyramid:
|
||
|
||
```cpp
|
||
// Backfills, applied after the append pass that published their indices: an amend whose
|
||
// index the append had not yet reached would be dropped by the pyramid's bounds test, and
|
||
// a hole's tail can be pushed and backfilled inside one frame interval.
|
||
if (!pending_amends_.empty()) {
|
||
for (const Amend& a : pending_amends_) pyramid_.amend(a.absolute, a.value);
|
||
pending_amends_.clear();
|
||
}
|
||
```
|
||
|
||
`invalidate()` clears it too — a rebind's queued amends address slots that no longer exist.
|
||
|
||
- [ ] **Step 2: Handle a late frame in `on_data`**
|
||
|
||
After the counter tracking from Task 6 Step 4, and before the stamping, add an early branch: a late frame does not advance the timeline, it repairs it.
|
||
|
||
```cpp
|
||
// A frame delivered behind the sequence: the slots it owns were reserved as NaN, so it
|
||
// writes values and nothing else. It must not be stamped, must not advance `tick_` and
|
||
// must not reach the trigger or the math channels -- the time ring was pre-stamped at
|
||
// reservation and the stateful consumers have already run past this region, which is the
|
||
// trade the design takes: the raw trace repairs, a derived trace keeps the gap, and the
|
||
// display therefore says "this arrived too late to be computed with".
|
||
if (frame->late) {
|
||
backfill_from(frame, part, channels, store, g);
|
||
return;
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 3: Write `backfill_from()`**
|
||
|
||
```cpp
|
||
void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part,
|
||
const std::vector<ChannelInfo>& channels, BufferStore& store,
|
||
u64 g) {
|
||
// Which hole this packet belongs to, by counter: that is the whole reason relative timing
|
||
// and resequencing are one design. A stamp could not answer it -- the packet's own stamps
|
||
// are a claim about a timeline the ring has already moved past -- but the counter indexes
|
||
// the reservation table directly.
|
||
const Hole* hole = nullptr;
|
||
for (const Hole& h : model_.holes()) {
|
||
if (frame->counter >= h.counter_first &&
|
||
frame->counter < h.counter_first + h.packets) {
|
||
hole = &h;
|
||
break;
|
||
}
|
||
}
|
||
if (hole == nullptr) return; // its hole has already slid out of the ring
|
||
|
||
const u64 offset_raw = u64(frame->counter - hole->counter_first) * u64(batch_);
|
||
u64 filled = 0;
|
||
for (std::size_t i = 0; i < channels.size(); ++i) {
|
||
if (channels[i].math) continue;
|
||
const u64 period = g * part.stride[i];
|
||
// Only slots lying *wholly* inside this packet's raw span are repaired. A slot
|
||
// straddling the span's edge mixes this packet's samples with ones still missing, and
|
||
// its real members were never retained -- so it stays NaN, and a displayed hole is at
|
||
// most one slot wider per edge than the true loss.
|
||
const u64 first_slot = (offset_raw + period - 1) / period;
|
||
const u64 last_slot = (offset_raw + u64(batch_)) / period;
|
||
for (u64 s = first_slot; s < last_slot; ++s) {
|
||
double acc = 0.0;
|
||
for (u64 j = 0; j < period; ++j) {
|
||
const u64 k = s * period + j - offset_raw;
|
||
const double raw =
|
||
channels[i].elements > 1
|
||
? udps_frame_value(frame, static_cast<std::uint32_t>(channels[i].signal), 0,
|
||
std::min<std::uint32_t>(static_cast<std::uint32_t>(k),
|
||
channels[i].elements - 1))
|
||
: udps_frame_value(frame, static_cast<std::uint32_t>(channels[i].signal),
|
||
static_cast<std::uint32_t>(k), 0);
|
||
acc += raw * channels[i].gain + channels[i].offset;
|
||
}
|
||
const f32 value = static_cast<f32>(acc / double(period));
|
||
const u64 absolute = hole->slot_begin + offset_raw / period + (s - first_slot);
|
||
// The ring first, by the producer that owns its writes; the pyramid is queued for the
|
||
// consumer. A slot the ring has already discarded returns false, and then there is no
|
||
// aggregate to amend either.
|
||
if (store.values(i).amend(absolute, value)) {
|
||
store.channel(i).queue_amend(absolute, value);
|
||
++filled;
|
||
}
|
||
}
|
||
}
|
||
model_.stats().backfilled.fetch_add(filled, std::memory_order_relaxed);
|
||
}
|
||
```
|
||
|
||
Declare it in `src/acquisition.hpp`.
|
||
|
||
- [ ] **Step 4: Build and measure the repair**
|
||
|
||
```bash
|
||
cmake --build build
|
||
sed -i 's/^run .*$/run late --reorder-every 10 --reorder-delay-ms 50/' /tmp/udps_reseq.sh
|
||
/tmp/udps_reseq.sh && grep -E "reseq=|backfilled=" /tmp/fault_late.log
|
||
```
|
||
|
||
Expected: `late` near 102, `holes` near 102, and **`backfilled` near `102 × 100 / stride`** — the slots the late packets repaired. `criticals=0`.
|
||
|
||
- [ ] **Step 5: Verify the repair is real, not just counted**
|
||
|
||
The `holes` and `backfilled` counters agreeing is necessary but not sufficient — a wrong `absolute` would amend the wrong slots and still count. Assert on the data: run with `--reorder-delay-ms 50` and compare the exported values against the clean run's, which must agree sample for sample where the hole was.
|
||
|
||
```bash
|
||
cat >> /tmp/udps_reseq.sh <<'EOF'
|
||
python3 - <<'PY'
|
||
import csv
|
||
def col(p, i=1):
|
||
with open(p) as f:
|
||
return [float(r[i]) for r in csv.reader(f) if r and not r[0].startswith('#')]
|
||
import math
|
||
v = col('/tmp/exp_late.csv')
|
||
print('nan slots in a backfilled export:', sum(1 for x in v if math.isnan(x)), 'of', len(v))
|
||
PY
|
||
EOF
|
||
```
|
||
|
||
Expected: the NaN count is a small multiple of the hole count (the one straddling slot per edge), **not** `102 × 100 / stride`. A NaN count equal to `hole_slots` means the amend wrote nowhere; a NaN count of zero across a boundary means the edge rule was not applied.
|
||
|
||
- [ ] **Step 6: Commit**
|
||
|
||
```bash
|
||
git add src/channel.hpp src/channel.cpp src/model.hpp src/model.cpp \
|
||
src/acquisition.hpp src/acquisition.cpp
|
||
git commit -m "feat: a late packet backfills the hole it left, values only"
|
||
```
|
||
|
||
---
|
||
|
||
## Phase 5 — The band, the probe and the documentation
|
||
|
||
### Task 11: A shaded band over an invalid span
|
||
|
||
**Files:**
|
||
- Modify: `deps/plotter/include/plot/plot.hpp`, `deps/plotter/src/render/renderer.cpp`
|
||
- Modify: `src/plot_host.cpp` (pass the model's holes to each row)
|
||
- Test: `deps/plotter/tests/test_render_smoke.cpp`
|
||
|
||
**Interfaces:**
|
||
- Produces: `void Plot::set_invalid_spans(std::span<const std::pair<f64, f64>>)` and `Plot::invalid_spans()`. Drawn behind the trace, in **data** coordinates.
|
||
|
||
- [ ] **Step 1: Write the failing test**
|
||
|
||
Append to `deps/plotter/tests/test_render_smoke.cpp`:
|
||
|
||
```cpp
|
||
PLOT_TEST(an_invalid_span_is_shaded_and_moves_with_the_data) {
|
||
// Unlike the x-note caption, a band marks a *region of the data*, so it must pan and zoom
|
||
// with it: a band that stayed put under a pan would accuse the wrong samples. It also must
|
||
// reserve no layout space -- the labels were measured without it, and a hole must not be
|
||
// able to move them.
|
||
plottest::HeadlessContext ctx(kW, kH);
|
||
// ... build a figure with one row and a full-width trace, as the neighbouring tests do ...
|
||
// 1. ink with no span, 2. ink with a span over the middle third, 3. ink with the same span
|
||
// after a pan of a known number of pixels.
|
||
// Assert: (2) differs from (1) only inside the middle third; (3)'s differing region is
|
||
// (2)'s shifted by exactly the pan; the axes rect is identical in all three.
|
||
}
|
||
```
|
||
|
||
Fill the body against the neighbouring tests' setup, and against `the_x_note_is_chrome_that_a_pan_or_a_zoom_does_not_move`, which is the *opposite* assertion and shows how the footprint is measured.
|
||
|
||
- [ ] **Step 2: Run it to verify it fails**
|
||
|
||
```bash
|
||
cmake --build deps/plotter/build/debug 2>&1 | tail -5
|
||
```
|
||
|
||
Expected: a compile error, `'set_invalid_spans' is not a member of 'plot::Plot'`.
|
||
|
||
- [ ] **Step 3: Add the API and the draw**
|
||
|
||
`Plot` stores `std::vector<std::pair<f64, f64>> invalid_spans_`. The renderer draws each span as a quad in the axes rect, clipped to it, **before** the traces and after the grid, transformed through the same x mapping the traces use — so it is data-anchored by construction rather than by a second code path. Colour is the readout backplate's tint at low alpha, so it reads as "no data here" and not as a trace.
|
||
|
||
- [ ] **Step 4: Run it to verify it passes**
|
||
|
||
```bash
|
||
cmake --build deps/plotter/build/debug && \
|
||
xvfb-run -a env PLOT_TEST_FILTER=invalid_span deps/plotter/build/debug/tests/plot_tests
|
||
```
|
||
|
||
Expected: `0 failed`.
|
||
|
||
- [ ] **Step 5: Feed it from the model**
|
||
|
||
In `src/plot_host.cpp`, where each row's per-frame state is written under `model.mutex()`, convert `model.holes()` to `(x_begin, x_end)` pairs in display coordinates (`+ display_epoch`) and call `set_invalid_spans()`. Only in `Live` and `Capturing`: a `Held` or `Paused` snapshot has its own stamps and the live hole table does not address it.
|
||
|
||
- [ ] **Step 6: Build, run and look**
|
||
|
||
```bash
|
||
cmake --build build
|
||
sed -i 's/^run .*$/run late --reorder-every 10 --reorder-delay-ms 50/' /tmp/udps_reseq.sh
|
||
/tmp/udps_reseq.sh
|
||
```
|
||
|
||
Expected: `criticals=0`, warnings at the baseline. Capture a frame with `UDPSCOPE_SCREENSHOT_PATH` if that probe exists and confirm the band sits over the gap.
|
||
|
||
- [ ] **Step 7: Commit**
|
||
|
||
```bash
|
||
git add deps/plotter/include/plot/plot.hpp deps/plotter/src/render/renderer.cpp \
|
||
deps/plotter/tests/test_render_smoke.cpp src/plot_host.cpp
|
||
git commit -m "feat: a reserved hole is shaded, so a sub-pixel gap is still a report"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 12: The probe and the documentation
|
||
|
||
**Files:**
|
||
- Modify: `src/ui.cpp` (`UDPSCOPE_PROBE_RESEQ_MS`)
|
||
- Modify: `docs/claude/verification.md`, `probe-readings.md`, `invariants-data-path.md`, `invariants-plot-and-layout.md`, `invariants-trigger-and-pinning.md`, `build-and-run.md`, `README.md`, `SPECS.md`, `todo.md`
|
||
- Modify: `docs/superpowers/specs/2026-10-08-resequencing-and-relative-timing-design.md` (the SPSC correction)
|
||
|
||
- [ ] **Step 1: Add the probe**
|
||
|
||
Following the pattern of `UDPSCOPE_PROBE_STATS_MS` at `src/ui.cpp:10648`, a one-shot that prints:
|
||
|
||
```
|
||
[reseq-probe] reseq=N late=N holes=N hole_slots=N backfilled=N nan_slots=N trigger_in_hole=0
|
||
```
|
||
|
||
`nan_slots` is counted by scanning the newest channel's retained window; `trigger_in_hole` asserts the engine did not latch inside a reserved span, by comparing the last trigger time against the hole table.
|
||
|
||
- [ ] **Step 2: Run it, three cases, serially**
|
||
|
||
```bash
|
||
for t in clean "--reorder-every 10" "--reorder-every 10 --reorder-delay-ms 50"; do : ; done
|
||
```
|
||
|
||
Run each as its own `/tmp/udps_reseq.sh` invocation with `UDPSCOPE_PROBE_RESEQ_MS=5000` added, and record the measured numbers. **Do not batch them** — four `xvfb` runs back to back starve the mock past its 2 s silence timeout and read as a reconnect.
|
||
|
||
- [ ] **Step 3: Write the readings down**
|
||
|
||
Add the measured table to `docs/claude/verification.md` beside the existing fault-injection section, and a per-line entry to `docs/claude/probe-readings.md` saying what each field must read and which defect a wrong reading names. **Every figure must be one measured on this machine** — no estimates.
|
||
|
||
- [ ] **Step 4: Update the invariants**
|
||
|
||
- `invariants-data-path.md`: append-only is no longer the whole truth. State `amend`'s non-finite precondition, that the ring is amended by the producer and the pyramid by the consumer, and why (the 32-byte non-atomic bucket).
|
||
- `invariants-plot-and-layout.md`: the band is data-anchored and reserves no layout space — the opposite of the `t0` caption, and both rules in one place so neither is applied to the other.
|
||
- `invariants-trigger-and-pinning.md`: the engine is not fed a reserved hole's samples, and why (a NaN comparison is false anyway, but feeding it would advance the hold-off clock over time nothing was sampled in).
|
||
- `build-and-run.md`: the vendored `udpsclient` diff — the reorder ring, late delivery replacing the discard, the third `clamp_deadline()`, and `reorder_window_s = 0` as the exact-restore switch.
|
||
|
||
- [ ] **Step 5: Update `README.md`, `SPECS.md` and `todo.md`**
|
||
|
||
`README.md` gains `--relative-time` and a paragraph on what recovery does and does not repair (values yes, math and trigger no). `SPECS.md` gets the deviation note. `todo.md` item 2 is struck.
|
||
|
||
- [ ] **Step 6: Correct the spec's SPSC line**
|
||
|
||
In the design document's Backfill section, replace *"the producer enqueues `(slot, value)` on an SPSC queue"* with the measured truth: `acquire_all()` is called under `model.mutex()` (`src/ui.cpp:6263`) and the producer holds that same mutex for all of `on_data` (`src/acquisition.cpp:386`), so a plain `std::vector` member is correctly serialised and no lock-free queue is needed.
|
||
|
||
- [ ] **Step 7: Full clean build, read both baselines**
|
||
|
||
```bash
|
||
rm -rf build && cmake -S . -B build -G Ninja -DCMAKE_BUILD_TYPE=RelWithDebInfo && \
|
||
cmake --build build 2>&1 | grep -c "warning:"
|
||
```
|
||
|
||
Expected: **4**. More is a regression introduced by this work and is fixed before the commit, not noted.
|
||
|
||
- [ ] **Step 8: Commit**
|
||
|
||
```bash
|
||
git add src/ui.cpp docs/ README.md SPECS.md todo.md
|
||
git commit -m "docs: the reseq probe's readings, and the invariants resequencing changes"
|
||
```
|
||
|
||
---
|
||
|
||
## Self-review
|
||
|
||
**Spec coverage.** Relative timing → Tasks 1, 2; the declared-rate-only rule → Task 2 Step 2; the absolute ring for export → Task 3; the reorder ring and the three arrival cases → Task 4; `clamp_deadline` → Task 4 Step 5; the ring bounds and early commit → Task 4 Step 4; `reorder_window_s` including the 0 kill switch → Task 4 Steps 1, 2, 7; hole reservation through the existing loop → Task 6 Step 3; the 256-packet cap → Task 6 Steps 2, 6; math NaN-through and the trigger not latching → Task 6 Step 3 and Task 12 Step 1; the pyramid amend → Task 8; the ring amend → Task 9; the thread split → Task 10 Step 1; counter-addressed backfill and the whole-slot-only edge rule → Task 10 Step 3; the counters including `stale_packets` → Task 7; the band → Task 11; the mock's missing option → Task 5; probe and docs → Task 12.
|
||
|
||
**Two spec items deliberately deferred, both named in the spec's own out-of-scope list:** replaying math/trigger over a backfill, and the `(min, max, mean, rms)` gather mode.
|
||
|
||
**Type consistency.** `amend` returns `bool` on `RingBuffer` (the caller branches on it, Task 10 Step 3) and `void` on `MinMaxPyramid` (bounds failure is a no-op by contract, Task 8 Step 3) — deliberately different, and both call sites match. `Hole::counter_first`/`packets` are the fields Task 10 searches on. `StreamStats::backfilled` is declared in Task 7 and incremented in Task 10.
|
||
|
||
**Known soft spots, flagged rather than papered over.** Task 3 Step 5 and Task 11 Steps 1 and 3 point at code I have not read line by line (`src/export.cpp`'s column assembly, `Plot`'s state block, the smoke test's figure setup) and say what to read rather than quoting code that might not compile. Everything else is complete.
|