flatudp: vector rows as channels or as a waveform, a re-openable source, probes for both

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/source/flatudp/flatudp_signals.cpp

@@ -87,12 +87,23 @@ bool all_scalar(const std::vector<Row>& rows) {
   return std::all_of(rows.begin(), rows.end(), [](const Row& r) { return r.count == 1; });
 }
 
+bool any_wave(const std::vector<Row>& rows) {
+  return std::any_of(rows.begin(), rows.end(), [](const Row& r) { return r.wave; });
+}
+
+int channel_count(const std::vector<Row>& rows) {
+  long n = 0;
+  for (const Row& r : rows) n += r.wave ? 1 : static_cast<long>(r.count);
+  return static_cast<int>(std::min<long>(n, 1L << 30));
+}
+
 std::string format(const std::vector<Row>& rows) {
   std::string out;
   for (const Row& r : rows) {
     if (!out.empty()) out += ',';
     out += r.name + ':' + type_name(r.type);
-    if (r.count != 1) out += '[' + std::to_string(r.count) + ']';
+    if (r.count != 1 || r.wave) out += '[' + std::to_string(r.count) + ']';
+    if (r.wave) out += 'w';
   }
   return out;
 }
@@ -118,11 +129,19 @@ std::string parse(const std::string& text, std::vector<Row>& rows) {
     std::string ty = trim(tok.substr(colon + 1));
     const std::size_t br = ty.find('[');
     if (br != std::string::npos) {
-      if (ty.back() != ']') {
+      const std::size_t cl = ty.rfind(']');
+      if (cl == std::string::npos || cl < br) {
         rows.clear();
         return where + "missing ']' after the count";
       }
-      const std::string n = trim(ty.substr(br + 1, ty.size() - br - 2));
+      const std::string tail = trim(ty.substr(cl + 1));
+      if (tail == "w") {
+        r.wave = true;
+      } else if (!tail.empty()) {
+        rows.clear();
+        return where + "unexpected '" + tail + "' after the count (only w, for a waveform)";
+      }
+      const std::string n = trim(ty.substr(br + 1, cl - br - 1));
       const bool digits = !n.empty() && n.size() <= 6 &&
                           std::all_of(n.begin(), n.end(), [](char c) { return c >= '0' && c <= '9'; });
       if (!digits) {
@@ -159,6 +178,30 @@ std::string problem(const std::vector<Row>& rows) {
   if (n > kMaxDatagram)
     return std::to_string(n) + " bytes per datagram exceed the " + std::to_string(kMaxDatagram) +
            " a UDP datagram can carry";
+
+  // A waveform is one channel of `count` samples per datagram; it cannot share a frame with a
+  // channel of another length, so a list of waveforms is all waveforms of one length.
+  const Row* first_wave = nullptr;
+  for (const Row& r : rows) {
+    if (!r.wave) continue;
+    if (r.count < 2) return "signal '" + r.name + "' is a waveform and needs a count above 1";
+    if (first_wave == nullptr) first_wave = &r;
+  }
+  if (first_wave != nullptr) {
+    for (const Row& r : rows) {
+      if (!r.wave)
+        return "signal '" + r.name + "' is not a waveform but '" + first_wave->name +
+               "' is; every row must be a waveform of the same length";
+      if (r.count != first_wave->count)
+        return "signal '" + r.name + "' is a waveform of " + std::to_string(r.count) +
+               " values but '" + first_wave->name + "' has " + std::to_string(first_wave->count) +
+               "; every row must be a waveform of the same length";
+    }
+  }
+  const int channels = channel_count(rows);
+  if (channels > kMaxChannels)
+    return std::to_string(channels) + " channels exceed the " + std::to_string(kMaxChannels) +
+           " one list may make (a vector row makes one per element unless it is a waveform)";
   return {};
 }
 
@@ -176,4 +219,20 @@ void decode_datagram(const std::vector<Row>& rows, const std::uint8_t* data, boo
   }
 }
 
+void decode_channels(const std::vector<Row>& rows, const std::uint8_t* data, bool big_endian,
+                     std::vector<std::vector<double>>& cols) {
+  const std::size_t channels = static_cast<std::size_t>(channel_count(rows));
+  if (cols.size() < channels) cols.resize(channels);
+  const bool swap = big_endian != (std::endian::native == std::endian::big);
+  std::size_t off = 0, ch = 0;
+  for (const Row& r : rows) {
+    const std::size_t w = type_size(r.type);
+    for (std::uint32_t k = 0; k < r.count; ++k) {
+      cols[r.wave ? ch : ch + k].push_back(load_typed(r.type, data + off, swap));
+      off += w;
+    }
+    ch += r.wave ? 1 : r.count;
+  }
+}
+
 }  // namespace udpscope::flatudp

src/source/flatudp/flatudp_signals.hpp

@@ -15,12 +15,16 @@ enum class Type : std::uint8_t { U8 = 0, I8, U16, I16, U32, I32, U64, I64, F32,
 
 constexpr std::uint32_t kMaxCount = 4096;
 constexpr std::size_t kMaxDatagram = 65507;  ///< the largest UDP payload over IPv4
+constexpr int kMaxChannels = 256;            ///< the most channels one list may make
 
 /// One signal of the sender's datagram: `count` consecutive values of `type`.
 struct Row {
   std::string name;
   Type type = Type::F32;
   std::uint32_t count = 1;
+  /// A vector row (count > 1) only: true is ONE channel of `count` consecutive samples per
+  /// datagram (a waveform); false is `count` channels `name[0]`.. of one sample each.
+  bool wave = false;
   bool operator==(const Row&) const = default;
 };
 
@@ -33,7 +37,13 @@ std::size_t datagram_size(const std::vector<Row>& rows);
 /// True when no row is a vector. Such a list can be batched, several datagrams to a frame.
 bool all_scalar(const std::vector<Row>& rows);
 
-/// `name:type[,name:type[count]...]`; a count of 1 is written without brackets.
+/// True when some row is a waveform (`wave`).
+bool any_wave(const std::vector<Row>& rows);
+/// The channels the list makes: a scalar 1, a non-wave vector `count`, a waveform 1.
+int channel_count(const std::vector<Row>& rows);
+
+/// `name:type[,name:type[count][w]...]`; a count of 1 is written without brackets, and the `w`
+/// of a waveform row follows its count.
 std::string format(const std::vector<Row>& rows);
 /// The inverse of `format`. Whitespace around names, types and counts is ignored. An empty text
 /// is an empty list (and not an error: `problem` reports it). Returns an error in words, empty
@@ -43,6 +53,12 @@ std::string parse(const std::string& text, std::vector<Row>& rows);
 /// The first reason this list cannot describe a datagram, in words for the user; empty when valid.
 std::string problem(const std::vector<Row>& rows);
 
+/// Decodes one datagram of `datagram_size(rows)` bytes to one column per *channel* (see
+/// `channel_count`): a non-wave vector row feeds `count` columns of one value each, a waveform row
+/// one column of `count` values. Appends; the caller clears the columns between uses.
+void decode_channels(const std::vector<Row>& rows, const std::uint8_t* data, bool big_endian,
+                     std::vector<std::vector<double>>& cols);
+
 /// Decodes one datagram of `datagram_size(rows)` bytes to doubles, row by row, byte-swapping when
 /// `big_endian` differs from the host. Appends to `cols[s]`; the caller clears them between uses.
 void decode_datagram(const std::vector<Row>& rows, const std::uint8_t* data, bool big_endian,

src/source/flatudp/flatudp_source.cpp

@@ -90,7 +90,7 @@ std::string FlatUdpSettings::identity() const {
 }
 
 std::size_t FlatUdpSettings::block() const {
-  if (!all_scalar(rows) || !(rate > 0.0) || !std::isfinite(rate)) return 1;
+  if (any_wave(rows) || !(rate > 0.0) || !std::isfinite(rate)) return 1;
   const long b = std::lround(rate / 100.0);
   return static_cast<std::size_t>(std::min(1024L, std::max(1L, b)));
 }
@@ -127,20 +127,44 @@ void FlatUdpSource::open(SourceSink& sink) {
   if (!p.empty()) throw std::runtime_error(p);
   sink_ = &sink;
 
+  // A fresh session: Stop then Go live opens the same object again, and the model needs its CONFIG.
+  configured_ = false;
+  connected_.store(false, std::memory_order_relaxed);
+  ever_connected_ = false;
+  recv_error_noted_ = false;
+  reported_sizes_.clear();
+  frame_counter_ = 0;
+  frames_ = 0;
+  reconnects_ = 0;
+  valid_.store(0, std::memory_order_relaxed);
+  wrong_.store(0, std::memory_order_relaxed);
+
+  // One SignalInfo per channel: a scalar, one element of a `channels` vector (`Name[k]`), or a whole
+  // waveform (rows = count consecutive samples at rate * count).
   signals_.clear();
-  for (const Row& r : cfg_.rows) {
+  const auto add = [&](const std::string& name, const Row& r, std::uint32_t rows, double rate) {
     SignalInfo s;
-    s.name = r.name;
+    s.name = name;
     s.type_code = static_cast<std::uint8_t>(r.type);
-    s.rows = r.count;
+    s.rows = rows;
     s.cols = 1;
     s.time_mode = kTimePacket;
-    s.sampling_rate = cfg_.rate;
+    s.sampling_rate = rate;
     signals_.push_back(s);
+  };
+  for (const Row& r : cfg_.rows) {
+    if (r.wave) {
+      add(r.name, r, r.count, cfg_.rate * static_cast<double>(r.count));
+    } else if (r.count == 1) {
+      add(r.name, r, 1, cfg_.rate);
+    } else {
+      for (std::uint32_t k = 0; k < r.count; ++k)
+        add(r.name + "[" + std::to_string(k) + "]", r, 1, cfg_.rate);
+    }
   }
   expected_ = datagram_size(cfg_.rows);
   block_ = cfg_.block();
-  pending_.assign(cfg_.rows.size(), {});
+  pending_.assign(signals_.size(), {});
   pending_n_ = 0;
   rx_.assign(65536, 0);
 
@@ -223,14 +247,19 @@ void FlatUdpSource::poll(int timeout_ms) {
 
 void FlatUdpSource::drain_socket() {
   for (int i = 0; i < 8192; ++i) {  // bounded, so a flood cannot starve the timers
-    const ssize_t n = ::recv(fd_, rx_.data(), rx_.size(), 0);
+    // MSG_TRUNC: the return value is the datagram's real length even if it did not fit `rx_`.
+    const ssize_t n = ::recv(fd_, rx_.data(), rx_.size(), MSG_TRUNC);
     if (n < 0) {
       if (errno == EINTR) continue;
+      if (errno != EAGAIN && errno != EWOULDBLOCK && !recv_error_noted_) {
+        recv_error_noted_ = true;
+        note(std::string("receive failed: ") + std::strerror(errno));
+      }
       break;  // EAGAIN: drained
     }
     const double t = epoch_seconds();
     const std::size_t len = static_cast<std::size_t>(n);
-    if (tap_) tap_(rx_.data(), len, t);
+    if (tap_) tap_(rx_.data(), std::min(len, rx_.size()), t);
     if (len == expected_) {
       accept(rx_.data());
     } else {
@@ -248,7 +277,7 @@ void FlatUdpSource::accept(const std::uint8_t* data) {
     connected_.store(true, std::memory_order_relaxed);
     sink_->on_event(SourceEvent::Connected, "");
   }
-  decode_datagram(cfg_.rows, data, cfg_.big_endian, pending_);
+  decode_channels(cfg_.rows, data, cfg_.big_endian, pending_);
   if (pending_n_ == 0) batch_start_ = last_valid_;
   if (++pending_n_ >= block_) flush();
 }

src/source/flatudp/flatudp_source.hpp

@@ -40,7 +40,7 @@ struct FlatUdpSettings {
   std::string problem() const;
   /// `[multicast:]port`; names the session file.
   std::string identity() const;
-  /// Datagrams per frame: 1 for a list with a vector, else `clamp(round(rate / 100), 1, 1024)`.
+  /// Datagrams per frame: 1 for a list with a waveform row, else `clamp(round(rate / 100), 1, 1024)`.
   std::size_t block() const;
 };
 
@@ -78,6 +78,7 @@ class FlatUdpSource final : public DataSource {
   int fd_ = -1;
   bool configured_ = false;
   bool ever_connected_ = false;
+  bool recv_error_noted_ = false;  ///< a failing `recv` is logged once per open
   std::size_t expected_ = 0;  ///< bytes of a valid datagram
   std::size_t block_ = 1;