martino/udpscope
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

acquisition: drive a DataSource from the registry; --source and --source-opt

Browse Source
This commit is contained in:
Martino Ferrari committed 2026-10-31 19:59:51 +01:00
1 parent 46076312cc
commit 504bc193ea
8 files changed
+156 -164
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/acquisition.cpp
+76 -137
View File
@@ -5,9 +5,8 @@
#include <limits>
#include <udps_client.h>
#include "layout_store.hpp"
#include "source/registry.hpp"
#include <glib/gstdio.h>
@@ -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<double>::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<std::uint32_t>(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<Acquisition*>(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<SignalInfo> 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<SignalInfo>& 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<unsigned>(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<int>(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<int>(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<std::uint32_t>(time_signal), k, 0)
: udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), 0, k);
? frame.value(static_cast<std::uint32_t>(time_signal), k, 0)
: frame.value(static_cast<std::uint32_t>(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<int>(frame->num_signals)) {
const udps_signal_t& ts = frame->signals[time_signal];
time_signal < static_cast<int>(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<std::uint32_t>(time_signal), k, 0)
: udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), 0, k);
? frame.value(static_cast<std::uint32_t>(time_signal), k, 0)
: frame.value(static_cast<std::uint32_t>(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<std::mutex> 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<std::int32_t>(frame->counter - last_counter_) > 0) {
last_counter_ = frame->counter;
if (!have_counter_ || static_cast<std::int32_t>(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<int>(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<unsigned long long>(snap ? snap->stamps.size() : 0),
frame->recv_time, static_cast<unsigned long long>(tick_),
frame.recv_time, static_cast<unsigned long long>(tick_),
std::chrono::duration<double>(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<std::uint32_t>(ch.signal), 0,
? frame.value(static_cast<std::uint32_t>(ch.signal), 0,
std::min<std::uint32_t>(k, ch.elements - 1))
: udps_frame_value(frame, static_cast<std::uint32_t>(ch.signal), k, 0);
: frame.value(static_cast<std::uint32_t>(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<std::int64_t>(frame->recv_time * 1.0e6);
const std::int64_t now_e = static_cast<std::int64_t>(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<ChannelInfo>& 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<std::int32_t>(frame->counter - h.counter_first);
const std::int32_t into = static_cast<std::int32_t>(frame.counter - h.counter_first);
if (into >= 0 && into < static_cast<std::int32_t>(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<std::uint32_t>(channels[i].signal), 0,
? frame.value(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),
: frame.value(static_cast<std::uint32_t>(channels[i].signal),
static_cast<std::uint32_t>(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<Acquisition*>(user)->on_config(signals, count, publish_mode);
}
void Acquisition::data_trampoline(const udps_frame_t* frame, void* user) {
static_cast<Acquisition*>(user)->on_data(frame);
}
void Acquisition::event_trampoline(udps_event_t event, const char* detail, void* user) {
static_cast<Acquisition*>(user)->on_event(static_cast<int>(event), detail);
}
} // namespace udpscope