perf: the producer's group test is a mask, not a 64-bit division

`plan_partition()` only ever chooses powers of two for `time_stride`, and the
comment there already claimed the test was cheap -- but the stride is a runtime
value, so `tick % period` compiled to a real division that the hot loop paid
once per channel per wire sample. The per-channel masks are built where the
partition is already reacted to, and the shared clock's group end is computed
once per sample instead of three times.

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

src/acquisition.cpp

@@ -626,6 +626,11 @@ void Acquisition::on_data(const udps_frame_t* frame) {
   // event window out into an immutable `EventSnapshot`.
   BufferStore& store = model_.live();
   const u64 g = part.time_stride;
+  // `tick % g == g - 1` as a mask. `plan_partition()` doubles `time_stride` from 1, so `g` is
+  // a power of two -- but it is a *runtime* value, so the modulo compiled to a real 64-bit
+  // division rather than the cheap test its comment there claims. See `acc_mask_` for the
+  // per-channel form and for what that cost on the hot loop.
+  const u64 g_mask = g - 1;
 
   // A re-plan replaced the store, so every ring's `produced()` restarted at 0 while `tick_`
   // kept counting. `backfill_from()` derives a slot index from a hole's `tick_begin`, which is
@@ -655,6 +660,13 @@ void Acquisition::on_data(const udps_frame_t* frame) {
     value_acc_.assign(channels.size(), 0.0);
     value_acc_n_.assign(channels.size(), 0);
     math_sample_.assign(channels.size(), std::nanf(""));
+    // The group boundaries, as masks, for the per-sample loops below. Here rather than in
+    // them: `g` and the strides are properties of the partition, and a partition change is
+    // exactly what this block already reacts to.
+    acc_mask_.resize(channels.size());
+    for (std::size_t i = 0; i < channels.size(); ++i) {
+      acc_mask_[i] = g * std::max<u64>(1, part.stride[i]) - 1;
+    }
     time_acc_ = 0.0;
     abs_time_acc_ = 0.0;
     time_acc_n_ = 0;
@@ -797,6 +809,9 @@ void Acquisition::on_data(const udps_frame_t* frame) {
 
   for (std::uint32_t k = 0; k < n; ++k) {
     const u64 tick = tick_++;
+    // The shared clock's group boundary, read once and reused by the time ring, the channel
+    // loop's own per-channel test and the math push: three tests of the same thing.
+    const bool group_end = (tick & g_mask) == g_mask;
     const double rel = std::max(stamps_[k] - epoch, last_stamp_);
     last_stamp_used = rel;
     last_stamp_ = rel;
@@ -808,7 +823,7 @@ void Acquisition::on_data(const udps_frame_t* frame) {
     last_abs_stamp_ = abs_stamps_[k] - epoch;
     abs_time_acc_ += last_abs_stamp_;
     ++time_acc_n_;
-    if (tick % g == g - 1) {
+    if (group_end) {
       store.time().push(time_acc_ / static_cast<double>(time_acc_n_));
       time_acc_ = 0.0;
       // The absolute clock on the same slot and under the same group mean, so an export can
@@ -853,8 +868,8 @@ void Acquisition::on_data(const udps_frame_t* frame) {
       value_acc_[i] += raw * ch.gain + ch.offset;
       ++value_acc_n_[i];
 
-      const u64 period = g * part.stride[i];
-      if (tick % period != period - 1) continue;
+      const u64 mask = acc_mask_[i];
+      if ((tick & mask) != mask) continue;
       const f32 value =
           static_cast<f32>(value_acc_[i] / static_cast<double>(value_acc_n_[i]));
       value_acc_[i] = 0.0;
@@ -878,7 +893,7 @@ void Acquisition::on_data(const udps_frame_t* frame) {
     // `publish_event()` clamps its snapshot to every value ring's `produced()`, so a derived
     // channel pushed after it would be missing its newest sample from the capture -- the same
     // one-sample clamp the deferred latch above exists for.
-    if (tick % g == g - 1) model_.push_math_sample(math_sample_);
+    if (group_end) model_.push_math_sample(math_sample_);
     if (latch_here) latch(latch_edge);
   }
   last_stamp_ = last_stamp_used;
@@ -929,8 +944,12 @@ void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part,
 
   double dt = part.base_dt;
   if (!(dt > 0.0)) dt = 1.0e-3;
+  // The same group tests as the live loop, written the same way: a hole's samples go through
+  // the same accumulators, so the two phases have to agree sample for sample.
+  const u64 g_mask = g - 1;
   for (u64 j = 0; j < raw; ++j) {
     const u64 tick = tick_++;
+    const bool group_end = (tick & g_mask) == g_mask;
     // The hole is stamped at the declared cadence from the last good stamp -- exact in
     // relative mode, and in stream mode the same continuation the glitch bridge makes.
     // It must be *stamped* rather than left short: the time ring is binary-searched by
@@ -945,7 +964,7 @@ void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part,
     time_acc_ += rel;
     abs_time_acc_ += last_abs_stamp_;
     ++time_acc_n_;
-    if (tick % g == g - 1) {
+    if (group_end) {
       store.time().push(time_acc_ / double(time_acc_n_));
       // Null in stream mode, where `time()` already is the absolute clock.
       if (plot::data::RingBuffer<f64>* abs = store.abs_time(); abs != nullptr) {
@@ -963,8 +982,8 @@ void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part,
       // pretend to. One slot per edge, against a hole thousands of slots wide.
       value_acc_[i] += std::numeric_limits<double>::quiet_NaN();
       ++value_acc_n_[i];
-      const u64 period = g * part.stride[i];
-      if (tick % period != period - 1) continue;
+      const u64 mask = acc_mask_[i];
+      if ((tick & mask) != mask) continue;
       store.values(i).push(static_cast<f32>(value_acc_[i] / double(value_acc_n_[i])));
       value_acc_[i] = 0.0;
       value_acc_n_[i] = 0;
@@ -973,7 +992,7 @@ void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part,
     // its source had. The trigger is *not* fed: a NaN comparison is false, so the engine
     // could not latch here anyway, and feeding it would advance its hold-off clock over
     // time nothing was sampled in.
-    if (tick % g == g - 1) {
+    if (group_end) {
       std::fill(math_sample_.begin(), math_sample_.end(), std::nanf(""));
       model_.push_math_sample(math_sample_);
     }