diff --git a/src/acquisition.cpp b/src/acquisition.cpp index a00dcb8..3fd1f9c 100644 --- a/src/acquisition.cpp +++ b/src/acquisition.cpp @@ -5,9 +5,8 @@ #include -#include - #include "layout_store.hpp" +#include "source/registry.hpp" #include @@ -29,22 +28,6 @@ namespace { /// ring to make room for samples that are not coming. constexpr double kGlitchPackets = 256.0; -SignalInfo to_info(const udps_signal_t& s) { - SignalInfo info; - info.name = s.name; - info.unit = s.unit; - info.type_code = s.type_code; - info.quant_type = s.quant_type; - info.rows = s.num_rows; - info.cols = s.num_cols; - info.range_min = s.range_min; - info.range_max = s.range_max; - info.time_mode = s.time_mode; - info.sampling_rate = s.sampling_rate; - info.time_signal_idx = s.time_signal_idx; - return info; -} - } // namespace Acquisition::Acquisition(ScopeModel& model, const Config& cfg) : model_(model), cfg_(cfg) {} @@ -52,31 +35,21 @@ Acquisition::Acquisition(ScopeModel& model, const Config& cfg) : model_(model), Acquisition::~Acquisition() { stop(); } void Acquisition::start() { - // A started -- or *restarted* -- client is a new stream. Going live from a file view is the restart that + // A started -- or *restarted* -- source is a new stream. Going live from a file view is the restart that // matters: the store still holds the event's samples, and the new stream's clock can sit behind their // stamps, so the guard below has to re-base rather than clamp. `ring=[2.5200..2.5200]` with `n` climbing // was the symptom -- every packet pinned flat at the file view's last stamp. model_.reset_stream(); last_stamp_ = -std::numeric_limits::infinity(); - udps_client_config_t c; - udps_client_config_init(&c); - c.server_addr = cfg_.server.c_str(); - c.server_port = cfg_.port; - c.multicast_group = cfg_.multicast.empty() ? nullptr : cfg_.multicast.c_str(); - c.interface_addr = cfg_.interface_addr.empty() ? nullptr : cfg_.interface_addr.c_str(); - c.recv_buffer_bytes = static_cast(cfg_.recv_buffer_bytes); - c.silence_timeout_s = 2.0; - c.reconnect_delay_s = 0.5; - c.keepalive_interval_s = 1.0; - - client_ = udps_client_create(&c); - if (client_ == nullptr) { - throw std::runtime_error("could not create the UDPS client for " + cfg_.server + ":" + - std::to_string(cfg_.port)); + if (!source_) { + SourceKind* kind = find_kind(cfg_.source_kind); + if (kind == nullptr) throw std::runtime_error("unknown source kind: " + cfg_.source_kind); + source_ = kind->make(cfg_.source_config()); } - udps_client_set_callbacks(client_, &Acquisition::config_trampoline, &Acquisition::data_trampoline, - &Acquisition::event_trampoline, this); - udps_client_set_datagram_cb(client_, &Acquisition::datagram_trampoline, this); + source_->set_datagram_tap([this](const std::uint8_t* bytes, std::size_t len, double recv_time) { + on_datagram(bytes, len, recv_time); + }); + source_->open(*this); // `--capture-packets`: armed before the first datagram, so the first CONNECT and // its CONFIG -- the hardest strange behaviours to reproduce -- are in the ring. // The open itself still happens on the producer thread, at the first datagram. @@ -90,35 +63,24 @@ void Acquisition::start() { } void Acquisition::stop() { - if (!running_.exchange(false, std::memory_order_acq_rel)) { - if (client_ != nullptr) { - udps_client_destroy(client_); - client_ = nullptr; - } - return; - } - if (thread_.joinable()) thread_.join(); - if (client_ != nullptr) { - udps_client_destroy(client_); - client_ = nullptr; - } + if (running_.exchange(false, std::memory_order_acq_rel) && thread_.joinable()) thread_.join(); + if (source_) source_->close(); } void Acquisition::run() { while (running_.load(std::memory_order_acquire)) { - udps_client_poll(client_, 50); - udps_stats_t s; - udps_client_stats(client_, &s); + source_->poll(50); + const SourceStats s = source_->stats(); StreamStats& out = model_.stats(); - out.packets.store(s.packets_received, std::memory_order_relaxed); - out.frames.store(s.frames_delivered, std::memory_order_relaxed); - out.gaps.store(s.counter_gaps, std::memory_order_relaxed); - out.resequenced.store(s.late_resequenced, std::memory_order_relaxed); - out.late.store(s.late_delivered, std::memory_order_relaxed); - out.stale.store(s.stale_packets, std::memory_order_relaxed); - out.overflows.store(s.reorder_overflows, std::memory_order_relaxed); + out.packets.store(s.packets, std::memory_order_relaxed); + out.frames.store(s.frames, std::memory_order_relaxed); + out.gaps.store(s.gaps, std::memory_order_relaxed); + out.resequenced.store(s.resequenced, std::memory_order_relaxed); + out.late.store(s.late, std::memory_order_relaxed); + out.stale.store(s.stale, std::memory_order_relaxed); + out.overflows.store(s.overflows, std::memory_order_relaxed); out.reconnects.store(s.reconnects, std::memory_order_relaxed); - out.connected.store(udps_client_is_connected(client_) != 0, std::memory_order_relaxed); + out.connected.store(s.connected, std::memory_order_relaxed); } } @@ -127,19 +89,19 @@ void Acquisition::set_capture_want(bool want, std::uint32_t slots) { if (slots > 0) capture_slots_ = slots; if (capture_path_.empty()) { // Keyed on the source: one path per source, the previous capture kept one deep. - std::string host = cfg_.server; + std::string host = cfg_.session_host(); for (char& ch : host) { if (ch == ':' || ch == '/' || ch == ' ') ch = '_'; } - std::string dir = cache_dir() + "/" + host + "-" + std::to_string(cfg_.port); + std::string dir = cache_dir() + "/" + host + "-" + std::to_string(cfg_.session_port()); g_mkdir_with_parents(dir.c_str(), 0700); capture_path_ = dir + "/capture.bin"; std::uint32_t addr = 0; struct ::in_addr a; - if (::inet_pton(AF_INET, cfg_.server.c_str(), &a) == 1) { + if (cfg_.source_kind == "udps" && ::inet_pton(AF_INET, cfg_.server.c_str(), &a) == 1) { addr = a.s_addr; // already network order, which is what the header wants } - capture_.set_peer(addr, cfg_.port); + capture_.set_peer(addr, cfg_.source_kind == "udps" ? cfg_.port : 0); } } capture_want_.store(want, std::memory_order_release); @@ -171,29 +133,20 @@ void Acquisition::on_datagram(const std::uint8_t* bytes, std::size_t len, double } } -void Acquisition::datagram_trampoline(const std::uint8_t* bytes, std::size_t len, - double recv_time, void* user) { - static_cast(user)->on_datagram(bytes, len, recv_time); -} - -void Acquisition::on_config(const udps_signal_t* signals, std::uint32_t count, - std::uint8_t publish_mode) { - std::vector info; - info.reserve(count); - for (std::uint32_t i = 0; i < count; ++i) info.push_back(to_info(signals[i])); - +void Acquisition::on_config(const std::vector& info, std::uint8_t publish_mode) { // One line per CONFIG on stderr: what the stream says it is. This is the first // thing to look at when a signal refuses to plot. static const char* const kModes[] = {"strict", "accumulate", "decimate"}; - std::fprintf(stderr, "[udpscope] CONFIG: %u signals, publish=%s\n", count, + std::fprintf(stderr, "[udpscope] CONFIG: %u signals, publish=%s\n", + static_cast(info.size()), publish_mode < 3 ? kModes[publish_mode] : "?"); for (const SignalInfo& sig : info) { std::fprintf(stderr, "[udpscope] '%s' %ux%u (%u elem) rate=%g Hz time=%s idx=%u unit='%s'\n", sig.name.c_str(), sig.rows, sig.cols, sig.elements(), sig.sampling_rate, - sig.time_mode == UDPS_TIME_PACKET ? "packet" - : sig.time_mode == UDPS_TIME_FULL_ARRAY ? "full-array" - : sig.time_mode == UDPS_TIME_FIRST_SAMPLE ? "first-sample" + sig.time_mode == kTimePacket ? "packet" + : sig.time_mode == kTimeFullArray ? "full-array" + : sig.time_mode == kTimeFirstSample ? "first-sample" : "last-sample", sig.time_signal_idx, sig.unit.c_str()); } @@ -213,14 +166,14 @@ void Acquisition::on_config(const udps_signal_t* signals, std::uint32_t count, time_source_ = 0; reset_session_state(); model_.stats().have_config.store(true, std::memory_order_relaxed); - model_.set_signals(std::move(info)); + model_.set_signals(info); } -void Acquisition::on_event(int event, const char* detail) { +void Acquisition::on_event(SourceEvent event, const char* detail) { (void)detail; - if (event == UDPS_EVENT_CONNECTED) { + if (event == SourceEvent::Connected) { model_.stats().connected.store(true, std::memory_order_relaxed); - } else if (event == UDPS_EVENT_DISCONNECTED) { + } else if (event == SourceEvent::Disconnected) { model_.stats().connected.store(false, std::memory_order_relaxed); // The session dropped mid-flight: everything the producer knows about *that* // session's numbering and clocks is now a lie about the next one. Clearing it here @@ -300,15 +253,15 @@ bool Acquisition::gap_plausible(std::uint32_t missing, double elapsed, const Par return true; } -std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_signal, +std::uint32_t Acquisition::build_stamps(const Frame& frame, int time_signal, bool counter_gap_refused) { // The frame's span in base ticks: a scalar signal carries `num_samples` values // (one per batched cycle), a block signal (rows*cols > 1) carries its whole // element block. The largest of those is how many base ticks this frame spans -- // in `strict` mode `num_samples` is 1 while a 1x1000 block carries 1000 samples. - std::uint32_t n = frame->num_samples; - for (std::uint32_t i = 0; i < frame->num_signals; ++i) { - n = std::max(n, udps_signal_num_elements(&frame->signals[i])); + std::uint32_t n = frame.num_samples; + for (std::uint32_t i = 0; i < frame.num_signals; ++i) { + n = std::max(n, frame.signals[i].elements()); } stamps_.assign(n, 0.0); abs_stamps_.assign(n, 0.0); @@ -359,9 +312,9 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign // refused, because the advance and the reservation are the *same* claim about the wire: // believing a phantom gap here would stretch the timeline by exactly the loss that did not // happen. - if (!frame->late && frame->lost > 0 && batch_ > 0 && !counter_gap_refused && - double(frame->lost) <= kGlitchPackets) { - rel_base_s_ += double(frame->lost) * double(batch_) * dt; + if (!frame.late && frame.lost > 0 && batch_ > 0 && !counter_gap_refused && + double(frame.lost) <= kGlitchPackets) { + rel_base_s_ += double(frame.lost) * double(batch_) * dt; } for (std::uint32_t k = 0; k < n; ++k) { stamps_[k] = rel_base_s_ + double(k) * dt; @@ -370,14 +323,14 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign // 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(frame->num_signals)) { - const udps_signal_t& ts = frame->signals[time_signal]; - const bool scalar_time = udps_signal_num_elements(&ts) == 1; + if (time_signal >= 0 && time_signal < static_cast(frame.num_signals)) { + const SignalInfo& ts = frame.signals[time_signal]; + const bool scalar_time = ts.elements() == 1; for (std::uint32_t k = 0; k < n; ++k) { abs_stamps_[k] = scalar_time - ? udps_frame_value(frame, static_cast(time_signal), k, 0) - : udps_frame_value(frame, static_cast(time_signal), 0, k); + ? frame.value(static_cast(time_signal), k, 0) + : frame.value(static_cast(time_signal), 0, k); } } else { for (std::uint32_t k = 0; k < n; ++k) abs_stamps_[k] = stamps_[k]; @@ -411,8 +364,8 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign // arrival clock for the rest of its life (until a CONFIG says otherwise): letting // the decision flip per frame is what interleaved two timelines in one ring. if (time_source_ != 2 && time_signal >= 0 && - time_signal < static_cast(frame->num_signals)) { - const udps_signal_t& ts = frame->signals[time_signal]; + time_signal < static_cast(frame.num_signals)) { + const SignalInfo& ts = frame.signals[time_signal]; // How the client indexes a signal's values depends on its shape // (`udps_frame_value`): a scalar carries one value per batched cycle, so the // stamp of sample k is (sample=k, elem=0); a multi-element signal carries one @@ -420,18 +373,18 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign // (sample=0, elem=k). Passing the scalar form for both is why FULL_ARRAY time // signals -- the only mode that carries a stamp per element -- read as // element 0 for every sample and were then judged degenerate. - const bool scalar_time = udps_signal_num_elements(&ts) == 1; + const bool scalar_time = ts.elements() == 1; auto value_of = [&](std::uint32_t k) { return scalar_time - ? udps_frame_value(frame, static_cast(time_signal), k, 0) - : udps_frame_value(frame, static_cast(time_signal), 0, k); + ? frame.value(static_cast(time_signal), k, 0) + : frame.value(static_cast(time_signal), 0, k); }; switch (ts.time_mode) { - case UDPS_TIME_FULL_ARRAY: + case kTimeFullArray: for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = value_of(k); used_stream_time = true; break; - case UDPS_TIME_FIRST_SAMPLE: { + case kTimeFirstSample: { // Only element 0 is meaningful here: the producer stamps the batch's first // sample and says nothing about the rest, so the cadence has to space them -- // exactly as LAST_SAMPLE below dates its batch backwards from its anchor. @@ -445,7 +398,7 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign used_stream_time = true; break; } - case UDPS_TIME_LAST_SAMPLE: { + case kTimeLastSample: { const double anchor = value_of(n > 0 ? n - 1 : 0); for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = anchor - double(n - 1 - k) * cadence; @@ -549,7 +502,7 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign // was produced, and the earlier ones are dated backwards. A gap (`lost`) // shows up as extra elapsed time between arrivals, which is exactly the // spacing the samples had. - const double end = frame->recv_time; + const double end = frame.recv_time; if (!have_arrival_epoch_) { arrival_epoch_ = end; have_arrival_epoch_ = true; @@ -605,7 +558,7 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign // to a dead clock; they do not, and the wipe is what made a single glitching packet // look like the stream restarting from zero. time_source_ = 2; - const double end = frame->recv_time; + const double end = frame.recv_time; arrival_epoch_ = end - (last_stamp_ + model_.epoch()); arrival_prev_ = end; } @@ -625,7 +578,7 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign return n; } -void Acquisition::on_data(const udps_frame_t* frame) { +void Acquisition::on_data(const Frame& frame) { std::lock_guard lock(model_.mutex()); if (!running_.load(std::memory_order_relaxed)) return; @@ -633,8 +586,8 @@ void Acquisition::on_data(const udps_frame_t* frame) { // The counter is this frame's identity and the hole's address. A *late* frame's counter is // behind the sequence, so it must not move the mark -- the hole it belongs in was addressed // from the counter that revealed it. - if (!have_counter_ || static_cast(frame->counter - last_counter_) > 0) { - last_counter_ = frame->counter; + if (!have_counter_ || static_cast(frame.counter - last_counter_) > 0) { + last_counter_ = frame.counter; have_counter_ = true; } @@ -668,7 +621,7 @@ void Acquisition::on_data(const udps_frame_t* frame) { // Placed here and not earlier because `part`/`channels` are not in scope above, and not // later because `build_stamps()` is stateful: a late frame's stamps would walk the clock // backwards and can throw the whole stream onto the arrival clock. - if (frame->late) { + if (frame.late) { backfill_from(frame, part, channels, model_.live(), part.time_stride); return; } @@ -688,7 +641,7 @@ void Acquisition::on_data(const udps_frame_t* frame) { // (full-array, more than one element, no declared rate). for (std::size_t i = 0; i < model_.signals().size(); ++i) { const SignalInfo& sig = model_.signals()[i]; - if (sig.time_mode == UDPS_TIME_FULL_ARRAY && sig.elements() > 1 && + if (sig.time_mode == kTimeFullArray && sig.elements() > 1 && sig.sampling_rate == 0.0) { time_signal = static_cast(i); break; @@ -702,11 +655,11 @@ void Acquisition::on_data(const udps_frame_t* frame) { // let one burst raise the very number that is supposed to corroborate it -- a 200-packet outage // smooths the interval to 40 ms and then refuses itself. const bool first_frame = !have_prev_frame_recv_; - const double since_prev = first_frame ? 0.0 : frame->recv_time - prev_frame_recv_; - const std::uint32_t missing = frame->lost; + const double since_prev = first_frame ? 0.0 : frame.recv_time - prev_frame_recv_; + const std::uint32_t missing = frame.lost; const bool gap_refused = missing > 0 && !gap_plausible(missing, since_prev, part, part.time_stride); - note_frame_arrival(frame->recv_time); + note_frame_arrival(frame.recv_time); const std::uint32_t n = build_stamps(frame, time_signal, gap_refused); if (n == 0) return; @@ -721,7 +674,7 @@ void Acquisition::on_data(const udps_frame_t* frame) { epoch_adopted_ = true; model_.set_epoch_if_unset(stamps_[0]); std::fprintf(stderr, "[udpscope] time base: epoch=%.3f, first frame recv=%.3f, n=%u\n", - model_.epoch(), frame->recv_time, n); + model_.epoch(), frame.recv_time, n); std::fflush(stderr); } const double epoch = model_.epoch(); @@ -930,7 +883,7 @@ void Acquisition::on_data(const udps_frame_t* frame) { edge, last_stamp_, span - post_s, post_s, snap ? snap->pre_s : 0.0, snap ? snap->post_s : 0.0, static_cast(snap ? snap->stamps.size() : 0), - frame->recv_time, static_cast(tick_), + frame.recv_time, static_cast(tick_), std::chrono::duration(std::chrono::steady_clock::now() - started_) .count()); std::fflush(stderr); @@ -1012,9 +965,9 @@ void Acquisition::on_data(const udps_frame_t* frame) { // the cycle's sample at element `k`. Either way element 0 is the channel. const double raw = ch.elements > 1 - ? udps_frame_value(frame, static_cast(ch.signal), 0, + ? frame.value(static_cast(ch.signal), 0, std::min(k, ch.elements - 1)) - : udps_frame_value(frame, static_cast(ch.signal), k, 0); + : frame.value(static_cast(ch.signal), k, 0); // Every sample is read and accumulated even when only one in `period` is stored: // that is the whole point of the filter. Skipping the read -- which is what the // old stride did -- is what made decimation alias. @@ -1091,7 +1044,7 @@ void Acquisition::on_data(const udps_frame_t* frame) { if (engine.pending()) model_.set_capture_edge(engine.last_trigger_time()); model_.set_trigger_pending(engine.pending()); static std::int64_t next_engine_log = 0; - const std::int64_t now_e = static_cast(frame->recv_time * 1.0e6); + const std::int64_t now_e = static_cast(frame.recv_time * 1.0e6); if (now_e >= next_engine_log) { next_engine_log = now_e + 1000000; // `elapsed/post_want` is the capture's progress in the only unit that matters now @@ -1196,7 +1149,7 @@ void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part, model_.stats().hole_slots.fetch_add(h.slot_count, std::memory_order_relaxed); } -void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part, +void Acquisition::backfill_from(const Frame& frame, const Partition& part, const std::vector& channels, BufferStore& store, u64 g) { // `batch_` is learned in `build_stamps()`, which a late frame never reaches. A late frame @@ -1213,7 +1166,7 @@ void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part // late packets would be dropped on the next line. Hole* hole = nullptr; for (Hole& h : model_.holes()) { - const std::int32_t into = static_cast(frame->counter - h.counter_first); + const std::int32_t into = static_cast(frame.counter - h.counter_first); if (into >= 0 && into < static_cast(h.packets)) { hole = &h; break; @@ -1224,7 +1177,7 @@ void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part // The packet's own span in absolute base ticks. `tick_begin` is the hole's first raw sample // and the packets inside a hole are consecutive, so the counter's distance from // `counter_first` places this one exactly. - const u64 lo = hole->tick_begin + u64(frame->counter - hole->counter_first) * u64(batch_); + const u64 lo = hole->tick_begin + u64(frame.counter - hole->counter_first) * u64(batch_); const u64 hi = lo + u64(batch_); u64 filled = 0; @@ -1248,10 +1201,10 @@ void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part const u64 k = s * period + j - lo; // this packet's own sample index, 0..batch_-1 const double raw = channels[i].elements > 1 - ? udps_frame_value(frame, static_cast(channels[i].signal), 0, + ? frame.value(static_cast(channels[i].signal), 0, std::min(static_cast(k), channels[i].elements - 1)) - : udps_frame_value(frame, static_cast(channels[i].signal), + : frame.value(static_cast(channels[i].signal), static_cast(k), 0); acc += raw * channels[i].gain + channels[i].offset; } @@ -1274,18 +1227,4 @@ void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part model_.stats().backfilled.fetch_add(filled, std::memory_order_relaxed); } -// Static trampolines: the client's C ABI cannot call a member function directly. -void Acquisition::config_trampoline(const udps_signal_t* signals, std::uint32_t count, - std::uint8_t publish_mode, void* user) { - static_cast(user)->on_config(signals, count, publish_mode); -} - -void Acquisition::data_trampoline(const udps_frame_t* frame, void* user) { - static_cast(user)->on_data(frame); -} - -void Acquisition::event_trampoline(udps_event_t event, const char* detail, void* user) { - static_cast(user)->on_event(static_cast(event), detail); -} - } // namespace udpscope diff --git a/src/acquisition.hpp b/src/acquisition.hpp index c2fa94e..c9e72c2 100644 --- a/src/acquisition.hpp +++ b/src/acquisition.hpp @@ -8,21 +8,22 @@ #include "model.hpp" #include "trigger.hpp" -#include +#include "source/data_source.hpp" #include #include #include #include +#include #include #include namespace udpscope { -/// Owns the UDPStreamer client and the producer thread that feeds the model's +/// Owns the data source and the producer thread that feeds the model's /// acquisition buffer. Everything it touches from the thread is serialised on /// `ScopeModel::mutex()`, so a UI-side structural change cannot race a frame. -class Acquisition { +class Acquisition : private SourceSink { public: Acquisition(ScopeModel& model, const Config& cfg); ~Acquisition(); @@ -62,22 +63,19 @@ class Acquisition { /// read by a packet whose engine was still disarmed, and then dropped. void request_force() noexcept { force_want_.store(true, std::memory_order_release); } + /// The source this acquisition drives, for the UI's capability panels: it may only call + /// `post()` on it. Null until `start()`. + DataSource* source() noexcept { return source_.get(); } + /// The capture, for the popover's row and the probe. Counters and errors are read /// cross-thread through PacketCapture's own published fields. const PacketCapture& capture() const noexcept { return capture_; } private: void run(); - void on_config(const udps_signal_t* signals, std::uint32_t count, std::uint8_t publish_mode); - void on_data(const udps_frame_t* frame); - void on_event(int event, const char* detail); - - static void config_trampoline(const udps_signal_t* signals, std::uint32_t count, - std::uint8_t publish_mode, void* user); - static void data_trampoline(const udps_frame_t* frame, void* user); - static void event_trampoline(udps_event_t event, const char* detail, void* user); - static void datagram_trampoline(const std::uint8_t* bytes, std::size_t len, - double recv_time, void* user); + void on_config(const std::vector& signals, std::uint8_t publish_mode) override; + void on_data(const Frame& frame) override; + void on_event(SourceEvent event, const char* detail) override; /// Drops the producer state that is per *connection*: the counter mark, the time /// anchors and the clock choice. A reconnect re-sends the CONFIG and the stream may /// have restarted its numbering; carrying the old session's marks across makes the @@ -89,9 +87,9 @@ class Acquisition { void on_datagram(const std::uint8_t* bytes, std::size_t len, double recv_time); /// Fills `stamps_` with this frame's base-tick timestamps, in stream seconds, - /// and returns how many were written (`frame->num_samples` unless the frame is + /// and returns how many were written (`frame.num_samples` unless the frame is /// malformed). - std::uint32_t build_stamps(const udps_frame_t* frame, int time_signal, + std::uint32_t build_stamps(const Frame& frame, int time_signal, bool counter_gap_refused); /// Pushes NaN for the `missing` packets' worth of samples that were never delivered, @@ -104,7 +102,7 @@ class Acquisition { /// Writes a late frame's values into the NaN-filled holes their counter addresses, /// amending the value ring directly (producer path) and queuing pyramid amends for /// `ChannelBuffer::acquire()` to apply on the consumer thread. - void backfill_from(const udps_frame_t* frame, const Partition& part, + void backfill_from(const Frame& frame, const Partition& part, const std::vector& channels, BufferStore& store, u64 g); /// Keeps the wire's own cadence up to date from the arrival times, in the modes where @@ -118,7 +116,7 @@ class Acquisition { ScopeModel& model_; Config cfg_; - udps_client* client_ = nullptr; + std::unique_ptr source_; std::thread thread_; std::atomic running_{false}; std::atomic paused_{false}; diff --git a/src/config.cpp b/src/config.cpp index 1b71402..73ab661 100644 --- a/src/config.cpp +++ b/src/config.cpp @@ -6,6 +6,7 @@ // For `RowSet::kMaxRows`: `--plots` names a row count, and one constant for the ceiling // is what stops the message and the clamp from drifting apart. #include "plot_host.hpp" +#include "source/registry.hpp" #include #include @@ -40,14 +41,35 @@ double parse_double(const char* name, const std::string& v) { } // namespace +SourceConfig Config::source_config() const { + SourceConfig out; + if (source_kind == "udps") { + out.set("server", server); + out.set("port", std::to_string(port)); + if (!multicast.empty()) out.set("multicast", multicast); + if (!interface_addr.empty()) out.set("interface", interface_addr); + out.set("recv_buffer", std::to_string(recv_buffer_bytes)); + } + for (const auto& kv : source_opts.items()) out.set(kv.first, kv.second); + return out; +} + +std::string Config::session_host() const { return udpscope::session_host(source_kind, source_config()); } +std::uint16_t Config::session_port() const { return udpscope::session_port(source_kind, source_config()); } + void print_usage(std::FILE* out, const char* argv0) { + std::string names; + for (const SourceKind* k : kinds()) names += (names.empty() ? "" : ", ") + k->id(); std::fprintf(out, - "usage: %s --server ADDR --port PORT [options]\n" + "usage: %s --server ADDR --port PORT [--source NAME] [options]\n" "\n" " --server ADDR stream source IPv4 address (default 127.0.0.1)\n" " --port PORT UDP port (unicast) or TCP port (multicast) (default 8000)\n" " --multicast GROUP join GROUP for DATA instead of unicast\n" " --interface ADDR local IPv4 of the interface to join on\n" + " --source NAME source kind (%s; default udps). --server/--port/\n" + " --multicast/--interface configure udps\n" + " --source-opt K=V a setting for the source kind; repeatable\n" " --buffer-mb MB size of each of the two sample buffers (default 200)\n" " --max-window SEC largest time window the toolbar offers (default 600)\n" " --holdoff SEC default trigger hold-off (default 0.010)\n" @@ -69,7 +91,7 @@ void print_usage(std::FILE* out, const char* argv0) { " and the GL area has no hardware context, so the window\n" " never paints or allocates\n" " -h, --help this text\n", - argv0); + argv0, names.c_str()); } Config parse_args(int argc, char** argv) { @@ -91,6 +113,18 @@ Config parse_args(int argc, char** argv) { c.multicast = value_of(argc, argv, i, "--multicast"); } else if (a == "--interface" || a.rfind("--interface=", 0) == 0) { c.interface_addr = value_of(argc, argv, i, "--interface"); + } else if (a == "--source" || a.rfind("--source=", 0) == 0) { + const std::string v = value_of(argc, argv, i, "--source"); + if (find_kind(v) == nullptr) { + std::string names; + for (const SourceKind* k : kinds()) names += (names.empty() ? "" : ", ") + k->id(); + fail("unknown source '" + v + "' (available: " + names + ")"); + } + c.source_kind = v; + c.source_specified = true; + } else if (a == "--source-opt" || a.rfind("--source-opt=", 0) == 0) { + const std::string v = value_of(argc, argv, i, "--source-opt"); + if (!c.source_opts.parse_assignment(v)) fail("--source-opt must be key=value: " + v); } else if (a == "--buffer-mb" || a.rfind("--buffer-mb=", 0) == 0) { c.buffer_mb = parse_double("--buffer-mb", value_of(argc, argv, i, "--buffer-mb")); c.buffer_mb_specified = true; diff --git a/src/config.hpp b/src/config.hpp index 3304834..7144a0b 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -6,6 +6,8 @@ #include #include +#include "source/source_config.hpp" + namespace udpscope { /// Everything the command line can say. Defaults are the ones the spec names @@ -21,6 +23,23 @@ struct Config { /// `server` valid". The default stays a usable value for every other path. bool server_specified = false; + /// `--source NAME`: which registered kind opens. `udps` is the default, and `--server/--port/ + /// --multicast/--interface` keep meaning UDPS. + std::string source_kind = "udps"; + /// Whether `--source` was given (the other half of "did the user choose a source"). + bool source_specified = false; + /// Repeatable `--source-opt k=v`: the kind's own settings. For `udps` they overlay the fields above. + SourceConfig source_opts; + + /// A source was named on the command line (`--server` or `--source`), so the chooser is skipped. + bool has_source() const noexcept { return server_specified || source_specified; } + /// The settings the kind's `make()` receives: for UDPS the four connection fields plus the receive + /// buffer, then `source_opts` on top. + SourceConfig source_config() const; + /// The (host, port) that name this source's session file; see `session_host()` in the registry. + std::string session_host() const; + std::uint16_t session_port() const; + /// Size of *each* of the two sample buffers, in megabytes. The application /// always allocates two identical buffers (spec item 1). double buffer_mb = 200.0; diff --git a/src/main.cpp b/src/main.cpp index b4e439c..c6be5c6 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -7,6 +7,7 @@ #include "waveform.hpp" #include "model.hpp" #include "plot_host.hpp" +#include "source/registry.hpp" #include "source_dialog.hpp" #include "ui.hpp" @@ -18,6 +19,7 @@ int main(int argc, char** argv) { udpscope::Config cfg; try { + udpscope::register_builtin_sources(); cfg = udpscope::parse_args(argc, argv); // The preferences fill in what the command line did not decide. `--relative-time` is an explicit // instruction for this run, so a preference can only turn the mode on, never off -- which is also @@ -57,13 +59,13 @@ int main(int argc, char** argv) { // declining must cost no buffer and no socket, and nothing below has to cope with "no source". // `--open` is the other answer to that question, and the chooser knows it: its Waveforms page // fills `open_waveform` in instead of the connection fields. - const bool asked = !cfg.server_specified && cfg.open_waveform.empty(); + const bool asked = !cfg.has_source() && cfg.open_waveform.empty(); if (asked && !udpscope::choose_source(cfg)) return 0; // A source named on the command line skips the chooser, but its *stored* configuration still // applies: the buffer size, the clock and the capture ring are decided here, before the model -- // which is the only moment they can be -- and a saved host whose buffer was ignored because the // address came from the command line is an edit that looks like it never took effect. - if (cfg.server_specified) udpscope::apply_stored_settings(cfg); + if (cfg.has_source()) udpscope::apply_stored_settings(cfg); // Everything from here to `gtk_window_present()` inside `run_ui()` runs with no main loop: the // buffer is allocated, a recording is read, then the whole UI is built. The chooser's window is diff --git a/src/source_dialog.cpp b/src/source_dialog.cpp index ee31ac6..b2a9dd8 100644 --- a/src/source_dialog.cpp +++ b/src/source_dialog.cpp @@ -1129,10 +1129,10 @@ gboolean pick_probe(gpointer user) { } // namespace void apply_stored_settings(Config& cfg) { - const std::string want = active_configuration(cfg.server, cfg.port); + const std::string want = active_configuration(cfg.session_host(), cfg.session_port()); if (want.empty()) return; LayoutState st; - if (!load_layout(configuration_path(cfg.server, cfg.port, want), st)) return; + if (!load_layout(configuration_path(cfg.session_host(), cfg.session_port(), want), st)) return; // Only where nothing newer was said: a flag the user typed, or the preferences they just applied, // wins -- because it is the more recent statement of intent, and the file is not a place to hide a // surprise from whoever just made it. diff --git a/src/ui.cpp b/src/ui.cpp index 19d4f8e..959554e 100644 --- a/src/ui.cpp +++ b/src/ui.cpp @@ -2494,7 +2494,7 @@ void on_about_clicked(GtkButton*, gpointer user) { App* app = static_cast(user); close_overflow_menu(*app); show_about(GTK_WINDOW(app->window), - layout_path(app->model->config().server, app->model->config().port)); + layout_path(app->model->config().session_host(), app->model->config().session_port())); } void on_help_clicked(GtkButton*, gpointer user) { @@ -5914,8 +5914,8 @@ int run_ui(int argc, char** argv, const Config& cfg, ScopeModel& model, Acquisit // both overrode the arrangement the file was opened with and could clobber a real session. With no // port, `save_layout_now()` refuses and the first-CONFIG layout load never runs, which is exactly // a source-less run. - app.source_host = cfg.open_waveform.empty() ? cfg.server : std::string(); - app.source_port = cfg.open_waveform.empty() ? cfg.port : 0; + app.source_host = cfg.open_waveform.empty() ? cfg.session_host() : std::string(); + app.source_port = cfg.open_waveform.empty() ? cfg.session_port() : 0; app.source_multicast = cfg.open_waveform.empty() ? cfg.multicast : std::string(); app.source_interface = cfg.interface_addr; // The event cache is keyed on the source, like the packet capture. A waveform has no source, so a diff --git a/src/ui_probe.cpp b/src/ui_probe.cpp index 20cdc1c..9dca011 100644 --- a/src/ui_probe.cpp +++ b/src/ui_probe.cpp @@ -5411,7 +5411,7 @@ gboolean help_probe(gpointer user) { App* a = static_cast(user); const HelpReport report = help_self_check(); const std::string session = - layout_path(a->model->config().server, a->model->config().port); + layout_path(a->model->config().session_host(), a->model->config().session_port()); // The manual is *launched* here and read no further: it opens in another process (Yelp, through // the `help:` URI) and neither it nor this box can draw a window for it, so there is nothing to