cell M1: per-UE state in the DU library (UeRegistry UeContext: TA loop, uplink-quality window)

docs/cell-m1.md

@@ -90,6 +90,36 @@ clock. On the rig it is driven by testrig `scripts/m1_cell_2ue.sh`.
   schedule (Kestrel's TBTT behaviour after a TSF step), which is at most one
   step-size off.
 
+## Per-UE state (`UeRegistry`)
+
+Each UE id the cell schedules has a `UeContext` in the registry
+(`src/cell/UeRegistry.h`). It holds the UE's timing-advance loop
+(`src/cell/TimingAdvance.h`, moved from the example) and its latest
+uplink-quality window: RSSI, SNR and EVM of its own frames, attributed per
+transmitter (`UeRxAttribution`, M0), not the device-wide `GetRxQuality`.
+- A context comes with a bind or an admission, and is dropped with a
+  release. An id reused by another UE starts fresh; before, the DU's TA map
+  kept a departed UE's TA for the next holder.
+- The DU closes the uplink window on its TA cadence (`CELL_TA_EVERY_S`) and
+  logs it per UE (`cell.ue_rx`). Link adaptation still steers on the SNR the
+  UE reports for the downlink; the uplink window is the per-UE measurement
+  the DU has of each UE, for whoever needs it (#566: EVM over F1).
+
+**Rig** (one run each, on this code):
+- `m1_cell_2ue` passes: both UEs in slot both ways, no overlaps. The DU
+  logged 60 uplink windows per UE (2 s each). Full windows held 20 frames,
+  the median for both UEs; the edge windows were partial (UE1 9 at the end,
+  UE2 4 at the start), from a UE starting or stopping mid-window. Last full
+  windows: RSSI −51 / −53 dBm, SNR 38 / 39 dB, EVM −33 / −31 dB.
+- `m3_cell_rach --leave-at` passes: the joiner's context was created at its
+  join and dropped with its release, 51 superframes after its last uplink.
+- `m6_handover` passes (arrivals, and an aborted arrival's reset).
+- **A reused id:** its frames carry the same transmitter address
+  (02:ce:11:cell:ue) as the previous holder's, so the first window after a
+  context begins may mix both; it is skipped (`rx_fresh`), at the cost of a
+  new UE's first window. Not run on the rig: a UE leaving and another taking
+  its id within one window.
+
 ## QoS classes (`CellScheduler`)
 
 M1's grid is fixed (one DL and one UL opportunity per UE per superframe),

examples/cell/cell.h

@@ -57,6 +57,7 @@
 #include "cell/GrantMap.h"
 #include "cell/Harq.h"
 #include "cell/SlotGrid.h"
+#include "cell/TimingAdvance.h"
 
 namespace cellnode {
 
@@ -723,61 +724,8 @@ inline int64_t ue_tsf_step(int64_t ue_tsf, int64_t du_tsf, int64_t ul_off, int64
   return (err > deadband_us || err < -deadband_us) ? -err : 0;
 }
 
-// DU timing-advance controller for one UE: the measured arrival error (DU
-// clock, arrival phase minus target) is averaged over a window; at each
-// update() the TA moves by -gain * mean, rounded to whole us, clamped to
-// +-max_ta. Positive TA = the UE fires earlier.
-class TaLoop {
-public:
-  TaLoop(double gain, int64_t max_ta_us) : gain_(gain), max_(max_ta_us) {}
-  /* The window keeps the newest kMaxWindow samples (a loop that is never
-   * updated, e.g. CELL_TA=0, must not grow without bound). */
-  static constexpr size_t kMaxWindow = 256;
-  void sample(int64_t err_us) {
-    if (win_.size() == kMaxWindow)
-      win_.erase(win_.begin());
-    win_.push_back(err_us);
-  }
-  size_t pending() const { return win_.size(); }
-  // Drop the window without moving the TA (samples taken under an old map).
-  void discard() { win_.clear(); }
-  // Returns true when the TA changed. min_samples guards a noisy window.
-  // The window's MEDIAN is the estimate: one wild arrival (a UE's beacon
-  // before its first TSF step, a TBTT on the old schedule after one) must
-  // not steer the TA — a mean let a single 39 ms sample push a joining UE
-  // out of its slot on the rig.
-  bool update(size_t min_samples) {
-    if (win_.size() < min_samples || win_.empty())
-      return false;
-    std::vector<int64_t> v;
-    v.swap(win_);
-    std::nth_element(v.begin(), v.begin() + static_cast<std::ptrdiff_t>(v.size() / 2), v.end());
-    const double mean = static_cast<double>(v[v.size() / 2]); /* the median */
-    last_mean_ = mean;
-    const int64_t step = static_cast<int64_t>(std::llround(gain_ * mean));
-    if (step == 0)
-      return false;
-    int64_t t = ta_ + step;  // arrival late (mean > 0) -> fire earlier -> TA up
-    if (t > max_)
-      t = max_;
-    if (t < -max_)
-      t = -max_;
-    const bool changed = t != ta_;
-    ta_ = t;
-    return changed;
-  }
-  int64_t ta() const { return ta_; }
-  // Restore a TA (a map that carried a moved TA failed to publish).
-  void set_ta(int64_t t) { ta_ = t; }
-  double last_mean() const { return last_mean_; }
-
-private:
-  double gain_;
-  int64_t max_;
-  int64_t ta_ = 0;
-  std::vector<int64_t> win_;
-  double last_mean_ = 0;
-};
+// The DU's per-UE timing-advance loop is src/cell/TimingAdvance.h (TaLoop).
+using devourer::cell::TaLoop;
 
 }  // namespace cellnode
 

examples/cell/main.cpp

@@ -899,6 +899,7 @@ int run_du(IRadio *dev, const Cfg &c) {
     positions.push_back({static_cast<uint8_t>(*grid->find(SlotKind::Downlink, static_cast<uint16_t>(i))),
                          static_cast<uint8_t>(*grid->find(SlotKind::Uplink, static_cast<uint16_t>(i)))});
   devourer::cell::UeRegistry reg(positions);
+  reg.set_ta_config(ta_gain, P / 4); /* each UE's TA loop (its UeContext) */
   /* Each UE's home identity (src/cell/Keys.h), recorded at its admission —
    * static, a RACH joiner's MAC, or the home a handover command carries —
    * and reported to the CU (F1 Report.homes) for the UE's handover context.
@@ -1232,11 +1233,10 @@ int run_du(IRadio *dev, const Cfg &c) {
     emit(l);
   }
 
-  std::mutex mu;  // guards ta, ue_epoch
-  std::map<uint16_t, TaLoop> ta;
+  std::mutex mu;  // guards reg's per-UE contexts (TA, uplink window), ue_epoch
   std::map<uint16_t, uint32_t> ue_epoch;
   for (uint16_t ue : ues)
-    ta.emplace(ue, TaLoop(ta_gain, P / 4));
+    reg.context(ue);
   /* Per-UE DL reliability state (guarded by mu): HARQ sender, link
    * adaptation, the UE-reported DL SNR over the current window, and the
    * HARQ counts at the last adaptation window. */
@@ -1748,12 +1748,14 @@ int run_du(IRadio *dev, const Cfg &c) {
       int64_t ta_now = 0;
       {
         std::lock_guard<std::mutex> lk(mu);
-        auto it = ta.find(ub->ue);
-        if (it != ta.end()) {
+        if (const auto *cx = reg.find_context(ub->ue)) {
           if (ta_on) /* CELL_TA=0: no loop to feed */
-            it->second.sample(err);
-          ta_now = it->second.ta();
+            reg.context(ub->ue).ta.sample(err);
+          ta_now = cx->ta.ta();
         }
+        /* its uplink quality, attributed to it (UeRegistry's window) */
+        reg.rx_frame(p.Data.data(), p.Data.size(), p.RxAtrib.rssi[0], p.RxAtrib.snr[0], p.RxAtrib.evm[0],
+                     p.RxAtrib.tsfl);
         ue_epoch[ub->ue] = ub->epoch;
         last_ul_sf[ub->ue] = pos.superframe;
         heard_ul_sf[ub->ue] = pos.superframe;
@@ -1824,6 +1826,7 @@ int run_du(IRadio *dev, const Cfg &c) {
     emit(l);
   }
   auto next_ta = std::chrono::steady_clock::now() + std::chrono::seconds(ta_every_s);
+  auto next_rxw = next_ta; /* the per-UE uplink windows' cadence */
   auto next_sync = std::chrono::steady_clock::now() + std::chrono::milliseconds(sync_loop_ms);
   int64_t sf = grid->at(static_cast<int64_t>(dev->ReadTsf())).superframe + 1;
   bool cu_alive_prev = !f1_on;
@@ -2671,7 +2674,7 @@ int run_du(IRadio *dev, const Cfg &c) {
               continue; /* not yet (or another instant is being published, or no clock) */
             set_position(g, hi->second.pos);
             next.grants.push_back(g);
-            ta.emplace(u.ue, TaLoop(ta_gain, P / 4));
+            reg.context(u.ue);
             /* Fresh DL state (an id reused after another UE left keeps
              * nothing of it); its DL queue is the DAPS buffer: kept. */
             dlue.insert_or_assign(u.ue, make_dlue(u.ue));
@@ -2691,7 +2694,7 @@ int run_du(IRadio *dev, const Cfg &c) {
               break;
             }
           next.grants.push_back(g);
-          ta.emplace(u.ue, TaLoop(ta_gain, P / 4));
+          reg.context(u.ue);
           /* Fresh DL state; a joiner has no DL stream yet (the CU starts
            * one once this DU reports it): a queue left under its id is
            * another UE's. */
@@ -2776,7 +2779,7 @@ int run_du(IRadio *dev, const Cfg &c) {
       {
         std::lock_guard<std::mutex> lk(mu);
         for (Grant &g : next.grants) {
-          TaLoop &l = ta.at(g.ue);
+          TaLoop &l = reg.context(g.ue).ta;
           // Only judge a UE running the newest map: its arrivals reflect it.
           const bool current = ue_epoch.count(g.ue) && ue_epoch[g.ue] == map.epoch;
           if (!current)
@@ -2825,6 +2828,27 @@ int run_du(IRadio *dev, const Cfg &c) {
         }
       }
     }
+    /* Each UE's uplink quality over the window, from its own frames
+     * (UeRegistry; the SA is 02:ce:11:<cell>:<ue>): closed on the TA cadence,
+     * whether or not the TA loop runs (CELL_TA=0). */
+    if (std::chrono::steady_clock::now() >= next_rxw) {
+      next_rxw += std::chrono::seconds(ta_every_s);
+      std::lock_guard<std::mutex> lk(mu);
+      const auto wins = reg.close_rx_window([&](const std::array<uint8_t, 6> &ta) -> uint16_t {
+        const bool ours = ta[0] == 0x02 && ta[1] == 0xce && ta[2] == 0x11 &&
+                          static_cast<uint16_t>(ta[3] << 8 | ta[4]) == c.cell;
+        return ours ? ta[5] : 0;
+      });
+      for (const auto &[ue, w] : wins) {
+        char l[220];
+        snprintf(l, sizeof l,
+                 "{\"ev\":\"cell.ue_rx\",\"ue\":%u,\"frames\":%u,\"rssi_dbm\":%d,\"snr_db\":%.1f,"
+                 "\"snr_min_db\":%.1f,\"snr_ok\":%d,\"evm_db\":%.1f,\"evm_ok\":%d}",
+                 ue, w.frames, w.rssi_mean_dbm, w.snr_mean_db, w.snr_min_db, w.snr_valid ? 1 : 0, w.evm_mean_db,
+                 w.evm_valid ? 1 : 0);
+        emit(l);
+      }
+    }
     /* A due F1 schedule: every UE's position, on its network-wide instant. */
     std::optional<uint64_t> f1_eff;
     if (f1_due) {
@@ -2878,7 +2902,7 @@ int run_du(IRadio *dev, const Cfg &c) {
     if (changed && static_cast<uint64_t>(sf) < map.effective_sf) {
       std::lock_guard<std::mutex> lk(mu);
       for (const Grant &g : map.grants)
-        ta.at(g.ue).set_ta(g.ta_us);
+        reg.context(g.ue).ta.set_ta(g.ta_us);
       next = map;
       changed = false;
       to_release.clear();
@@ -2903,7 +2927,7 @@ int run_du(IRadio *dev, const Cfg &c) {
         }
       {
         std::lock_guard<std::mutex> lk(mu);
-        ta.erase(ho_eff_ue);
+        reg.reset_context(ho_eff_ue);
         dlue.erase(ho_eff_ue);
         last_ul_sf.erase(ho_eff_ue);
       }
@@ -3062,7 +3086,7 @@ int run_du(IRadio *dev, const Cfg &c) {
            * next TA pass retries. */
           std::lock_guard<std::mutex> lk(mu);
           for (const Grant &g : map.grants)
-            ta.at(g.ue).set_ta(g.ta_us);
+            reg.context(g.ue).ta.set_ta(g.ta_us);
         }
         const GrantMap &shown = ok ? map : next;
         char l[320];

src/cell/TimingAdvance.h

@@ -0,0 +1,81 @@
+/* TimingAdvance.h — M1 (#255) the DU's per-UE timing-advance loop.
+ *
+ * A UE's uplink is its hardware beacon, steered through its TSF to the slot
+ * its grant gives it (examples/cell); the DU measures where each uplink
+ * actually arrives against that slot and moves the UE's TA, which the grant
+ * map carries back. The TA corrects clock and actuation phase, not
+ * propagation (indoors that is ns, below this design's µs).
+ *
+ * Moved from examples/cell/cell.h: per-UE state belongs to the DU library
+ * (src/cell/UeRegistry.h UeContext holds one per UE). Pure header; the
+ * cell_node ctest covers it. */
+#ifndef DEVOURER_CELL_TIMINGADVANCE_H
+#define DEVOURER_CELL_TIMINGADVANCE_H
+
+#include <algorithm>
+#include <cmath>
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
+namespace devourer::cell {
+
+// DU timing-advance controller for one UE: the measured arrival error (DU
+// clock, arrival phase minus target) is averaged over a window; at each
+// update() the TA moves by -gain * mean, rounded to whole us, clamped to
+// +-max_ta. Positive TA = the UE fires earlier.
+class TaLoop {
+public:
+  TaLoop(double gain, int64_t max_ta_us) : gain_(gain), max_(max_ta_us) {}
+  /* The window keeps the newest kMaxWindow samples (a loop that is never
+   * updated, e.g. CELL_TA=0, must not grow without bound). */
+  static constexpr size_t kMaxWindow = 256;
+  void sample(int64_t err_us) {
+    if (win_.size() == kMaxWindow)
+      win_.erase(win_.begin());
+    win_.push_back(err_us);
+  }
+  size_t pending() const { return win_.size(); }
+  // Drop the window without moving the TA (samples taken under an old map).
+  void discard() { win_.clear(); }
+  // Returns true when the TA changed. min_samples guards a noisy window.
+  // The window's MEDIAN is the estimate: one wild arrival (a UE's beacon
+  // before its first TSF step, a TBTT on the old schedule after one) must
+  // not steer the TA — a mean let a single 39 ms sample push a joining UE
+  // out of its slot on the rig.
+  bool update(size_t min_samples) {
+    if (win_.size() < min_samples || win_.empty())
+      return false;
+    std::vector<int64_t> v;
+    v.swap(win_);
+    std::nth_element(v.begin(), v.begin() + static_cast<std::ptrdiff_t>(v.size() / 2), v.end());
+    const double mean = static_cast<double>(v[v.size() / 2]); /* the median */
+    last_mean_ = mean;
+    const int64_t step = static_cast<int64_t>(std::llround(gain_ * mean));
+    if (step == 0)
+      return false;
+    int64_t t = ta_ + step;  // arrival late (mean > 0) -> fire earlier -> TA up
+    if (t > max_)
+      t = max_;
+    if (t < -max_)
+      t = -max_;
+    const bool changed = t != ta_;
+    ta_ = t;
+    return changed;
+  }
+  int64_t ta() const { return ta_; }
+  // Restore a TA (a map that carried a moved TA failed to publish).
+  void set_ta(int64_t t) { ta_ = t; }
+  double last_mean() const { return last_mean_; }
+
+private:
+  double gain_;
+  int64_t max_;
+  int64_t ta_ = 0;
+  std::vector<int64_t> win_;
+  double last_mean_ = 0;
+};
+
+} /* namespace devourer::cell */
+
+#endif /* DEVOURER_CELL_TIMINGADVANCE_H */