working on todo.md features

src/acquisition.cpp

@@ -5,6 +5,13 @@
 
 #include <udps_client.h>
 
+#include "layout_store.hpp"
+
+#include <glib/gstdio.h>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+
 #include <algorithm>
 #include <cmath>
 #include <cstring>
@@ -61,6 +68,13 @@ void Acquisition::start() {
   }
   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);
+  // `--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.
+  if (cfg_.capture_packets > 0) {
+    set_capture_want(true, cfg_.capture_packets);
+  }
 
   started_ = std::chrono::steady_clock::now();
   running_.store(true, std::memory_order_release);
@@ -100,6 +114,60 @@ void Acquisition::run() {
   }
 }
 
+void Acquisition::set_capture_want(bool want, std::uint32_t slots) {
+  if (want) {
+    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;
+      for (char& ch : host) {
+        if (ch == ':' || ch == '/' || ch == ' ') ch = '_';
+      }
+      std::string dir = cache_dir() + "/" + host + "-" + std::to_string(cfg_.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) {
+        addr = a.s_addr;  // already network order, which is what the header wants
+      }
+      capture_.set_peer(addr, cfg_.port);
+    }
+  }
+  capture_want_.store(want, std::memory_order_release);
+}
+
+void Acquisition::on_datagram(const std::uint8_t* bytes, std::size_t len, double recv_time) {
+  const bool want = capture_want_.load(std::memory_order_acquire);
+  if (want && !capture_.enabled()) {
+    // A user-initiated act that happens once per toggle, on this thread: a rename, a
+    // create, a 62.6 MiB reserve and an mmap. It can cost the packets in flight, and
+    // the measured cost is `capture_.open_ms()`.
+    if (!capture_.open(capture_path_, capture_slots_)) {
+      char err[128];
+      capture_.error(err, sizeof(err));
+      std::fprintf(stderr, "[udpscope] packet capture failed to open: %s\n", err);
+      // Stay off rather than retrying every datagram against a full disk; the GTK
+      // thread reads the flag back and returns the button to off.
+      capture_want_.store(false, std::memory_order_release);
+      return;
+    }
+    std::fprintf(stderr, "[udpscope] packet capture on: %s (%u slots, %.1f ms to open)\n",
+                 capture_.path().c_str(), capture_.slot_count(), capture_.open_ms());
+  } else if (!want && capture_.enabled()) {
+    capture_.close();
+    std::fprintf(stderr, "[udpscope] packet capture off\n");
+  }
+  if (capture_.enabled()) {
+    capture_.store(bytes, len, recv_time);
+  }
+}
+
+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;
@@ -129,6 +197,11 @@ void Acquisition::on_config(const udps_signal_t* signals, std::uint32_t count,
   warned_outage_ = false;
   warned_bridge_ = false;
   warned_time_source_ = false;
+  // The clock choice is per connection too: a reconnect re-derives it from the frames
+  // the new session actually delivers, rather than trusting a decision made about a
+  // session that dropped.
+  time_source_ = 0;
+  reset_session_state();
   model_.stats().have_config.store(true, std::memory_order_relaxed);
   model_.set_signals(std::move(info));
 }
@@ -139,9 +212,31 @@ void Acquisition::on_event(int event, const char* detail) {
     model_.stats().connected.store(true, std::memory_order_relaxed);
   } else if (event == UDPS_EVENT_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
+    // is what makes the next frame re-anchor instead of reading as late, as a glitch or
+    // as an outage -- and nothing here touches the buffer, whose history survives the
+    // drop (see `ScopeModel::set_signals`'s unchanged-CONFIG path).
+    reset_session_state();
   }
 }
 
+void Acquisition::reset_session_state() noexcept {
+  have_counter_ = false;
+  last_counter_ = 0;
+  have_stream_anchor_ = false;
+  last_stream_anchor_ = 0.0;
+  stamp_failures_ = 0;
+  // The arrival clock's *anchor* stays -- stamps are small seconds the ring's timeline
+  // is already scaled against, and re-anchoring mid-stream would move x -- but the
+  // smoothed interval and the previous arrival are per-connection measurements, and
+  // carrying them across a multi-second outage makes the first new frame read as a
+  // clock fault rather than as the resumption it is.
+  have_arrival_interval_ = false;
+  arrival_interval_ = 0.0;
+  arrival_prev_ = 0.0;
+}
+
 std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_signal) {
   // 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
@@ -537,8 +632,9 @@ void Acquisition::on_data(const udps_frame_t* frame) {
   // only an absolute slot index while `tick_ / period == produced()`, so the tick counter has
   // to restart with the buffer -- otherwise the amend lands outside the ring, repairs nothing,
   // and the hole is still marked repaired, which is the band saying the opposite of the truth.
-  // The generation is the signal rather than the stride/channel-count comparison below: a
-  // reconnect with an identical CONFIG rebuilds the store leaving both unchanged.
+  // The generation is the signal rather than the stride/channel-count comparison below: only
+  // a CONFIG that *changes* the stream re-plans the store -- a reconnect's identical CONFIG
+  // keeps it, history and all (see `ScopeModel::set_signals`).
   // `slot_base_` is deliberately *not* reset -- it is relative mode's clock origin, samples
   // since the stream began, and relative time must not jump backwards when the user retypes
   // the window.