diff --git a/CMakeLists.txt b/CMakeLists.txt index fa2b67b..6c8dcf8 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -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 diff --git a/src/search.cpp b/src/search.cpp new file mode 100644 index 0000000..01dbd73 --- /dev/null +++ b/src/search.cpp @@ -0,0 +1,154 @@ +// SPDX-License-Identifier: MIT +// SPDX-FileCopyrightText: 2026 Martino Ferrari + +#include "search.hpp" + +#include +#include +#include + +namespace udpscope { + +namespace { +double ms_since(const std::chrono::steady_clock::time_point& t0) { + const std::chrono::duration 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(kChunk)); + v_scratch_.resize(static_cast(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(i) >= hits_.size()) i = static_cast(hits_.size()) - 1; + cursor_ = i; +} + +bool BufferSearch::next() noexcept { + if (hits_.empty() || cursor_ + 1 >= static_cast(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 lock(model.mutex()); + if (channel_ < 0 || static_cast(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(stamp); + store.time().copy_from(end_ - 1, 1, &stamp); + total_s_ = static_cast(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(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(t_scratch_[static_cast(i)]); + const double e = + detector_.feed(static_cast(v_scratch_[static_cast(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 diff --git a/src/search.hpp b/src/search.hpp new file mode 100644 index 0000000..3739a48 --- /dev/null +++ b/src/search.hpp @@ -0,0 +1,98 @@ +// SPDX-License-Identifier: MIT +// SPDX-FileCopyrightText: 2026 Martino Ferrari + +#pragma once + +#include +#include +#include + +#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& 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 hits_; + int cursor_ = -1; + std::uint64_t chunks_ = 0; + double worst_chunk_ms_ = 0.0; + double total_ms_ = 0.0; + std::vector t_scratch_; + std::vector v_scratch_; +}; + +} // namespace udpscope diff --git a/src/ui.cpp b/src/ui.cpp index 7ef4664..0ff1f5a 100644 --- a/src/ui.cpp +++ b/src/ui.cpp @@ -25,6 +25,7 @@ #include "measure.hpp" #include "model.hpp" #include "plot_host.hpp" +#include "search.hpp" #include #include @@ -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 snap = app.model->pause_view()) { + if (!snap->stamps.empty()) { + app.rows->pin_pause_window(static_cast(snap->stamps.front()), + static_cast(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(n)); +} + void on_trigger_toggled(GtkToggleButton* button, gpointer user) { App* app = static_cast(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(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 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(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(user); + if (a->search.prev()) search_jump(*a, a->search.hits()[static_cast( + 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(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( + 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(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(user); + if (a->window == nullptr) return G_SOURCE_REMOVE; + g_search_t0 = a->model->trigger_time(); + EventCondition c; + int channel = -1; + { + std::lock_guard 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(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(std::atoi(env)), stats_probe, &app); } + if (const char* env = g_getenv("UDPSCOPE_PROBE_SEARCH_MS")) { + g_timeout_add(static_cast(std::atoi(env)), search_probe, &app); + } if (const char* env = g_getenv("UDPSCOPE_PROBE_FIELDS_MS")) { g_timeout_add(static_cast(std::atoi(env)), fields_probe, &app); }