feat: the buffer can be searched for the trigger's own condition, and the hits stepped through

This commit is contained in:
Martino Ferrari committed 2026-10-08 09:56:47 +02:00
1 parent 2cee85c52b
commit 415fceca87
4 files changed
+517

No files matched your search

+1
View File
@@ -110,6 +110,7 @@ add_executable(udpscope
src/acquisition.cpp
src/plot_host.cpp
src/export.cpp
src/search.cpp
src/layout_store.cpp
src/source_dialog.cpp
src/measure.cpp
+154
View File
@@ -0,0 +1,154 @@
// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari
#include "search.hpp"
#include <chrono>
#include <cmath>
#include <mutex>
namespace udpscope {
namespace {
double ms_since(const std::chrono::steady_clock::time_point& t0) {
const std::chrono::duration<double, std::milli> d = std::chrono::steady_clock::now() - t0;
return d.count();
}
} // namespace
void BufferSearch::start(const EventCondition& c, int channel) {
clear();
condition_ = c;
channel_ = channel;
detector_.reset();
detector_.configure(c);
// `configure()` is a no-op for an unchanged condition, so the explicit reset above is what
// makes a re-run of the *same* search start from a clean machine.
detector_.reset();
state_ = State::Running;
t_scratch_.resize(static_cast<std::size_t>(kChunk));
v_scratch_.resize(static_cast<std::size_t>(kChunk));
}
void BufferSearch::cancel() noexcept {
if (state_ == State::Running) state_ = State::Cancelled;
}
void BufferSearch::clear() noexcept {
state_ = State::Idle;
started_ = false;
at_ = 0;
end_ = 0;
first_s_ = 0.0;
scanned_s_ = 0.0;
total_s_ = 0.0;
hits_.clear();
cursor_ = -1;
chunks_ = 0;
worst_chunk_ms_ = 0.0;
total_ms_ = 0.0;
}
void BufferSearch::select(int i) noexcept {
if (hits_.empty()) {
cursor_ = -1;
return;
}
if (i < 0) i = 0;
if (static_cast<std::size_t>(i) >= hits_.size()) i = static_cast<int>(hits_.size()) - 1;
cursor_ = i;
}
bool BufferSearch::next() noexcept {
if (hits_.empty() || cursor_ + 1 >= static_cast<int>(hits_.size())) return false;
++cursor_;
return true;
}
bool BufferSearch::prev() noexcept {
if (hits_.empty() || cursor_ <= 0) return false;
--cursor_;
return true;
}
bool BufferSearch::pump(ScopeModel& model) {
if (state_ != State::Running) return false;
const auto chunk_t0 = std::chrono::steady_clock::now();
bool truncated = false;
u64 n = 0;
{
std::lock_guard<std::mutex> lock(model.mutex());
if (channel_ < 0 || static_cast<std::size_t>(channel_) >= model.channel_count()) {
state_ = State::Cancelled;
return false;
}
BufferStore& store = model.live();
if (!started_) {
// The whole retained buffer: the bounds are deliberately absurd, because
// `live_window_locked()` clamps them to the time ring and to *every* value ring, which
// is the one cut that keeps the stamps and the values index-aligned by construction.
u64 first = 0;
u64 last = 0;
model.live_window_locked(-1.0e300, 1.0e300, first, last);
at_ = first;
end_ = last;
started_ = true;
if (end_ > at_) {
f64 stamp = 0.0;
store.time().copy_from(at_, 1, &stamp);
first_s_ = static_cast<double>(stamp);
store.time().copy_from(end_ - 1, 1, &stamp);
total_s_ = static_cast<double>(stamp) - first_s_;
}
}
// The producer keeps pushing while we scan, so the oldest samples go. `copy_from()` fixes
// *which* indices are read, not that the slots still hold them -- a read behind the base
// would silently return whatever overwrote it, which would be a hit at a time that never
// existed. Skipping forward and reporting `Truncated` is the honest answer.
const u64 base = store.time().base();
if (base > at_) {
at_ = base;
truncated = true;
detector_.reset(); // the sequence has a gap in it; no pulse spans it
}
if (at_ >= end_) {
state_ = truncated ? State::Truncated : State::Done;
total_ms_ += ms_since(chunk_t0);
return false;
}
n = end_ - at_ < kChunk ? end_ - at_ : kChunk;
store.time().copy_from(at_, n, t_scratch_.data());
store.values(static_cast<std::size_t>(channel_)).copy_from(at_, n, v_scratch_.data());
}
// The evaluation is outside the lock: it is the expensive half, and it touches only the
// copies above.
for (u64 i = 0; i < n; ++i) {
const double t = static_cast<double>(t_scratch_[static_cast<std::size_t>(i)]);
const double e =
detector_.feed(static_cast<double>(v_scratch_[static_cast<std::size_t>(i)]), t);
if (std::isfinite(e)) {
hits_.push_back(e);
if (hits_.size() >= kHitsMax) {
at_ += i + 1;
scanned_s_ = t - first_s_;
state_ = State::Capped;
++chunks_;
total_ms_ += ms_since(chunk_t0);
return false;
}
}
scanned_s_ = t - first_s_;
}
at_ += n;
++chunks_;
const double cost = ms_since(chunk_t0);
if (cost > worst_chunk_ms_) worst_chunk_ms_ = cost;
total_ms_ += cost;
if (at_ >= end_) {
state_ = truncated ? State::Truncated : State::Done;
return false;
}
return true;
}
} // namespace udpscope
+98
View File
@@ -0,0 +1,98 @@
// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari
#pragma once
#include <cstddef>
#include <cstdint>
#include <vector>
#include "condition.hpp"
#include "model.hpp"
namespace udpscope {
/// A scan of the whole retained buffer for every sample satisfying an `EventCondition`.
///
/// It runs an `EventDetector` -- the *same* evaluator the live trigger runs -- over the ring
/// in stream order, so a hit is by construction an instant the trigger would have fired on.
/// That is the claim the probe's armed-vs-scanned cross-check checks, and it is the reason
/// the detector was separated from the engine at all.
///
/// Everything here is in **ring-stamp space**, which is stream seconds relative to the
/// model's epoch: the clock `TriggerEngine::process()` is fed, the clock
/// `ScopeModel::live_window_locked()` searches, and the clock `publish_pause_view()` cuts in.
/// No conversion to display x happens in this class.
///
/// It runs on the GTK thread in chunks rather than on a thread of its own: the samples live
/// behind `ScopeModel::mutex()`, which the producer holds for each packet, and a scan that
/// took the lock once would stall acquisition for the whole buffer. One chunk per call holds
/// it for `kChunk` samples and no longer.
class BufferSearch {
public:
/// The hit list's ceiling. A condition matching a large share of the samples -- an edge
/// trigger on a noisy channel -- would otherwise fill memory with a list nobody can
/// navigate; the scan stops and says `Capped`, which is information rather than a hang.
static constexpr std::size_t kHitsMax = 10000;
/// Samples per `pump()`, and therefore the only latency knob: the producer waits on the
/// lock for exactly one chunk's scan.
static constexpr u64 kChunk = 8192;
enum class State {
Idle, ///< nothing has been searched.
Running, ///< `pump()` has more to do.
Done, ///< the whole retained buffer was scanned.
Truncated, ///< the producer overwrote samples before they were reached.
Capped, ///< `kHitsMax` was reached.
Cancelled, ///< the user stopped it, or the channel went away.
};
/// Begins a scan of the channel for the condition. Discards any previous result.
void start(const EventCondition& c, int channel);
void cancel() noexcept;
void clear() noexcept;
/// Scans one chunk. Returns true while there is more to do. Takes `model.mutex()` itself,
/// so the caller must *not* hold it.
bool pump(ScopeModel& model);
State state() const noexcept { return state_; }
int channel() const noexcept { return channel_; }
const std::vector<double>& hits() const noexcept { return hits_; }
/// Which hit the view is on, or -1 for none.
int cursor() const noexcept { return cursor_; }
void select(int i) noexcept;
/// Moves the cursor and returns whether it moved; the ends do not wrap, because a wrap
/// from the last hit to the first looks exactly like the view not having moved at all.
bool next() noexcept;
bool prev() noexcept;
/// How much of the buffer has been scanned, and how much there was, in stream seconds.
double scanned_s() const noexcept { return scanned_s_; }
double total_s() const noexcept { return total_s_; }
/// Cost, for the probe: chunks scanned, the worst single chunk, and the whole scan.
std::uint64_t chunks() const noexcept { return chunks_; }
double worst_chunk_ms() const noexcept { return worst_chunk_ms_; }
double total_ms() const noexcept { return total_ms_; }
private:
EventDetector detector_{};
EventCondition condition_{};
int channel_ = -1;
State state_ = State::Idle;
bool started_ = false; ///< the index range has been established.
u64 at_ = 0; ///< next absolute index to scan.
u64 end_ = 0; ///< one past the last absolute index to scan.
double first_s_ = 0.0; ///< stamp of `at_` when the scan began.
double scanned_s_ = 0.0;
double total_s_ = 0.0;
std::vector<double> hits_;
int cursor_ = -1;
std::uint64_t chunks_ = 0;
double worst_chunk_ms_ = 0.0;
double total_ms_ = 0.0;
std::vector<f64> t_scratch_;
std::vector<f32> v_scratch_;
};
} // namespace udpscope
+264
View File
@@ -25,6 +25,7 @@
#include "measure.hpp"
#include "model.hpp"
#include "plot_host.hpp"
#include "search.hpp"
#include <adwaita.h>
#include <gtk/gtk.h>
@@ -157,6 +158,15 @@ struct App {
/// the stream's own clock advances whether the view is frozen or not and would say nothing
/// about how long the user has been looking at a still picture.
gint64 pause_started_us = 0;
/// The buffer search. Driven a few milliseconds at a time from the frame tick, because its
/// samples live behind the model's mutex and the producer needs it back between chunks.
BufferSearch search;
GtkWidget* search_run_button = nullptr;
GtkWidget* search_cancel_button = nullptr;
GtkWidget* search_status_label = nullptr;
GtkWidget* search_pos_label = nullptr;
GtkWidget* search_prev_button = nullptr;
GtkWidget* search_next_button = nullptr;
GtkWidget* menu_button = nullptr;
GtkWidget* cursors_toggle = nullptr;
/// Where the panel controls ended up, decided once at build time: the dump runs after the window
@@ -2341,6 +2351,76 @@ void on_pause_toggled(GtkToggleButton* button, gpointer user) {
: "pause acquisition");
}
/// Freezes the view on a hit, centred, with the hit itself at x = 0.
///
/// This is the existing pause, not a fifth display mode: the toggle has to end up on, or the
/// next frame's `clamp_to_data()` would follow the live edge and walk the frozen window off
/// the hit entirely.
void search_jump(App& app, double t) {
const double w = app.rows->time_window();
if (app.pause_button != nullptr &&
!gtk_toggle_button_get_active(GTK_TOGGLE_BUTTON(app.pause_button))) {
// Its handler publishes the *current* window; the publish below replaces it.
gtk_toggle_button_set_active(GTK_TOGGLE_BUTTON(app.pause_button), TRUE);
}
app.model->publish_pause_view(t - w / 2.0, t + w / 2.0, /*x_offset=*/t);
app.pause_started_us = g_get_monotonic_time();
// Pinned from the snapshot's *own* stamps, never from the window asked for: a cut is
// clamped to what the ring holds, so a hit near the oldest sample yields a shorter
// snapshot, and pinning the request would leave an empty margin that a zoom moves into.
if (std::shared_ptr<const EventSnapshot> snap = app.model->pause_view()) {
if (!snap->stamps.empty()) {
app.rows->pin_pause_window(static_cast<double>(snap->stamps.front()),
static_cast<double>(snap->stamps.back()));
}
}
}
void update_search_panel(App& app) {
if (app.search_status_label == nullptr) return;
const BufferSearch& s = app.search;
const std::size_t n = s.hits().size();
char text[192];
switch (s.state()) {
case BufferSearch::State::Idle:
std::snprintf(text, sizeof(text), "not searched");
break;
case BufferSearch::State::Running:
std::snprintf(text, sizeof(text), "scanning %s of %s, %zu found",
format_window(s.scanned_s()).c_str(), format_window(s.total_s()).c_str(), n);
break;
case BufferSearch::State::Done:
std::snprintf(text, sizeof(text), "%zu found in %s", n, format_window(s.total_s()).c_str());
break;
case BufferSearch::State::Truncated:
std::snprintf(text, sizeof(text), "%zu found; the stream overtook the oldest samples", n);
break;
case BufferSearch::State::Capped:
std::snprintf(text, sizeof(text), "%zu found (the limit) in the newest %s", n,
format_window(s.scanned_s()).c_str());
break;
case BufferSearch::State::Cancelled:
std::snprintf(text, sizeof(text), "stopped after %zu", n);
break;
}
gtk_label_set_text(GTK_LABEL(app.search_status_label), text);
char pos[64];
if (s.cursor() < 0 || n == 0) {
std::snprintf(pos, sizeof(pos), "--");
} else {
std::snprintf(pos, sizeof(pos), "%d of %zu", s.cursor() + 1, n);
}
gtk_label_set_text(GTK_LABEL(app.search_pos_label), pos);
const bool running = s.state() == BufferSearch::State::Running;
gtk_widget_set_sensitive(app.search_run_button, !running);
gtk_widget_set_sensitive(app.search_cancel_button, running);
gtk_widget_set_sensitive(app.search_prev_button, s.cursor() > 0);
gtk_widget_set_sensitive(app.search_next_button,
n > 0 && s.cursor() + 1 < static_cast<int>(n));
}
void on_trigger_toggled(GtkToggleButton* button, gpointer user) {
App* app = static_cast<App*>(user);
const bool on = gtk_toggle_button_get_active(button);
@@ -6014,6 +6094,15 @@ gboolean tick(gpointer user) {
app->dbg_measures_pane_w = gtk_widget_get_width(app->measures_list);
app->dbg_measures_pane_h = gtk_widget_get_height(app->measures_list);
}
// A few milliseconds of scanning per frame: enough that a full buffer finishes in a
// handful of seconds, little enough that neither the producer (which waits on the mutex
// for one chunk) nor the frame budget notices.
if (app->search.state() == BufferSearch::State::Running) {
const gint64 budget_end_us = g_get_monotonic_time() + 5000;
while (app->search.pump(*app->model) && g_get_monotonic_time() < budget_end_us) {
}
update_search_panel(*app);
}
if (model.generation() != app->signal_generation) {
// A CONFIG re-describes the stream and drops the buffer, so nothing measured before it
// belongs in the same series as what comes after.
@@ -6852,6 +6941,94 @@ GtkWidget* build_trigger_panel(App& app) {
G_CALLBACK(on_trigger_drop_changed), &app);
g_signal_connect(app.trigger_mode_drop, "notify::selected",
G_CALLBACK(on_trigger_drop_changed), &app);
// Searching *is* the trigger's condition applied to history, so it belongs in this panel
// rather than in one of its own: a second place to spell the same condition would be a
// second place to get it wrong. Collapsed by default -- it is not part of running a trigger.
GtkWidget* search_exp = gtk_expander_new("Search the buffer");
gtk_widget_set_tooltip_text(search_exp,
"find every place in the retained buffer where the condition "
"above holds, and step through them");
GtkWidget* sbox = gtk_box_new(GTK_ORIENTATION_VERTICAL, 4);
gtk_widget_set_margin_top(sbox, 4);
GtkWidget* srun = gtk_box_new(GTK_ORIENTATION_HORIZONTAL, 6);
app.search_run_button = gtk_button_new_with_label("Search");
gtk_widget_set_hexpand(app.search_run_button, true);
g_signal_connect(app.search_run_button, "clicked", G_CALLBACK(+[](GtkButton*, gpointer user) {
App* a = static_cast<App*>(user);
// Published, not read back off the widgets: the producer's copy is the
// one the live trigger uses, so the search asks exactly its question.
publish_trigger_fields(a);
EventCondition c;
int channel = -1;
{
std::lock_guard<std::mutex> lock(a->model->mutex());
c = a->model->trigger().condition;
channel = a->model->trigger().channel;
}
a->search.start(c, channel);
update_search_panel(*a);
}),
&app);
gtk_box_append(GTK_BOX(srun), app.search_run_button);
app.search_cancel_button = gtk_button_new_with_label("Stop");
gtk_widget_set_sensitive(app.search_cancel_button, false);
g_signal_connect(app.search_cancel_button, "clicked", G_CALLBACK(+[](GtkButton*, gpointer user) {
App* a = static_cast<App*>(user);
a->search.cancel();
update_search_panel(*a);
}),
&app);
gtk_box_append(GTK_BOX(srun), app.search_cancel_button);
gtk_box_append(GTK_BOX(sbox), srun);
app.search_status_label = gtk_label_new("not searched");
gtk_widget_set_halign(app.search_status_label, GTK_ALIGN_START);
gtk_label_set_wrap(GTK_LABEL(app.search_status_label), true);
gtk_widget_add_css_class(app.search_status_label, "dim-label");
gtk_box_append(GTK_BOX(sbox), app.search_status_label);
GtkWidget* snav = gtk_box_new(GTK_ORIENTATION_HORIZONTAL, 6);
app.search_prev_button = gtk_button_new_from_icon_name("go-previous-symbolic");
gtk_widget_set_tooltip_text(app.search_prev_button, "the previous hit");
gtk_widget_set_sensitive(app.search_prev_button, false);
g_signal_connect(app.search_prev_button, "clicked", G_CALLBACK(+[](GtkButton*, gpointer user) {
App* a = static_cast<App*>(user);
if (a->search.prev()) search_jump(*a, a->search.hits()[static_cast<std::size_t>(
a->search.cursor())]);
update_search_panel(*a);
}),
&app);
gtk_box_append(GTK_BOX(snav), app.search_prev_button);
app.search_pos_label = gtk_label_new("--");
gtk_widget_set_hexpand(app.search_pos_label, true);
gtk_box_append(GTK_BOX(snav), app.search_pos_label);
app.search_next_button = gtk_button_new_from_icon_name("go-next-symbolic");
gtk_widget_set_tooltip_text(app.search_next_button, "the next hit");
gtk_widget_set_sensitive(app.search_next_button, false);
g_signal_connect(app.search_next_button, "clicked", G_CALLBACK(+[](GtkButton*, gpointer user) {
App* a = static_cast<App*>(user);
// The first press with no cursor goes to the first hit, not the second.
if (a->search.cursor() < 0) {
a->search.select(0);
} else if (!a->search.next()) {
update_search_panel(*a);
return;
}
search_jump(*a, a->search.hits()[static_cast<std::size_t>(
a->search.cursor())]);
update_search_panel(*a);
}),
&app);
gtk_box_append(GTK_BOX(snav), app.search_next_button);
gtk_box_append(GTK_BOX(sbox), snav);
gtk_expander_set_child(GTK_EXPANDER(search_exp), sbox);
gtk_box_append(GTK_BOX(box), search_exp);
g_signal_connect(app.trigger_type_drop, "notify::selected",
G_CALLBACK(on_trigger_shape_changed), &app);
g_signal_connect(app.trigger_cmp_drop, "notify::selected",
@@ -8830,6 +9007,90 @@ gboolean condition_probe(gpointer user) {
return G_SOURCE_REMOVE;
}
/// Headless self-check for the buffer search, and for the claim that makes it worth having:
/// a scanned hit is an instant the *live trigger* would have fired on.
///
/// It arms the same pulse-width condition `condition_probe` arms, waits for a real latch,
/// then searches the buffer for that condition and compares the latched t0 with the nearest
/// scanned hit. `agree=1` is the assertion; `agree=0` means the two evaluators have drifted
/// apart, which is the whole failure mode separating `EventDetector` out was meant to
/// prevent. It then steps to a hit and reports the pinned axis against it.
int g_search_phase = 0;
double g_search_t0 = 0.0;
gboolean search_probe_report(gpointer user) {
App* a = static_cast<App*>(user);
if (a->window == nullptr) return G_SOURCE_REMOVE;
const BufferSearch& s = a->search;
if (s.state() == BufferSearch::State::Running) return G_SOURCE_CONTINUE;
double nearest = 0.0;
double err = 0.0;
bool have = false;
for (const double h : s.hits()) {
const double d = std::fabs(h - g_search_t0);
if (!have || d < std::fabs(err)) {
have = true;
nearest = h;
err = h - g_search_t0;
}
}
// One stored sample period is the tolerance: the trigger saw the stored (filtered) stream
// and so does the scan, so they must agree to the sample, not merely to the neighbourhood.
const double tol = a->model->partition().stored_rate() > 0.0
? 1.0 / a->model->partition().stored_rate()
: 1.0e-3;
std::fprintf(stderr,
"[search-probe] state=%d hits=%zu scanned=%.4f total=%.4f chunks=%" G_GUINT64_FORMAT
" worst=%.3fms total=%.3fms armed t0=%.6f vs scanned first=%.6f err=%.6g agree=%d\n",
int(s.state()), s.hits().size(), s.scanned_s(), s.total_s(), s.chunks(),
s.worst_chunk_ms(), s.total_ms(), g_search_t0, have ? nearest : 0.0, err,
int(have && std::fabs(err) <= tol));
if (!s.hits().empty()) {
a->search.select(0);
search_jump(*a, a->search.hits()[0]);
update_search_panel(*a);
const plot::Range view = a->rows->view_range();
std::fprintf(stderr,
"[search-probe] jumped to hit 0 t=%.6f mode=%d axis=[%.4f..%.4f] pos=%d\n",
a->search.hits()[0], int(a->model->display_mode()), view.min, view.max,
a->search.cursor() + 1);
}
std::fflush(stderr);
return G_SOURCE_REMOVE;
}
gboolean search_probe_run(gpointer user) {
App* a = static_cast<App*>(user);
if (a->window == nullptr) return G_SOURCE_REMOVE;
g_search_t0 = a->model->trigger_time();
EventCondition c;
int channel = -1;
{
std::lock_guard<std::mutex> lock(a->model->mutex());
c = a->model->trigger().condition;
channel = a->model->trigger().channel;
}
a->search.start(c, channel);
std::fprintf(stderr, "[search-probe] started on channel %d, armed t0=%.6f\n", channel,
g_search_t0);
std::fflush(stderr);
g_timeout_add(100, search_probe_report, a);
return G_SOURCE_REMOVE;
}
gboolean search_probe(gpointer user) {
App* a = static_cast<App*>(user);
if (a->window == nullptr) return G_SOURCE_REMOVE;
// Reuse the condition probe's arming entirely: the point is that one condition gives two
// evaluators the same answer, so the condition must be set in exactly one place.
g_search_phase = 0;
condition_probe(a);
g_timeout_add(6000, search_probe_run, a);
return G_SOURCE_REMOVE;
}
/// Headless self-check: drives the window control the four ways a user can -- type a number,
/// change the unit under it, and step a decade either way -- and prints what each produced.
/// Each line reports the typed request, the window the plot took and the trigger's event
@@ -10212,6 +10473,9 @@ int run_ui(int argc, char** argv, const Config& cfg, ScopeModel& model, Acquisit
if (const char* env = g_getenv("UDPSCOPE_PROBE_STATS_MS")) {
g_timeout_add(static_cast<guint>(std::atoi(env)), stats_probe, &app);
}
if (const char* env = g_getenv("UDPSCOPE_PROBE_SEARCH_MS")) {
g_timeout_add(static_cast<guint>(std::atoi(env)), search_probe, &app);
}
if (const char* env = g_getenv("UDPSCOPE_PROBE_FIELDS_MS")) {
g_timeout_add(static_cast<guint>(std::atoi(env)), fields_probe, &app);
}