process_data_ring.h

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <cstdint>
#include <span>
#include <vector>
 
namespace mm::node {
 
class ProcessDataRing {
 public:
  struct Record {
    uint64_t seq = 0;  
    // Wall-clock epoch nanoseconds at exchange (system clock). u64 holds it until ~year 2554.
    // INTERNAL precision only: epoch-ns today (~1.7e18) far exceeds the 2^53 exact-integer limit
    // of a JavaScript double. The live monitoring row reduces it to epoch MICROSECONDS (ns / 1000)
    // before JSON — epoch-µs stays exact in JS until ~year 2255 and, unlike milliseconds, gives
    // every cycle a distinct timestamp even at sub-millisecond cycle periods (periodUs < 1000).
    // The full ns precision is retained here for the binary dump.
    uint64_t timestampNs = 0;
    int32_t wkc = 0;  
    std::vector<uint8_t> inputs;
    std::vector<uint8_t> outputs;
  };
 
  ProcessDataRing() = default;
  ~ProcessDataRing();
 
  ProcessDataRing(const ProcessDataRing&) = delete;
  ProcessDataRing& operator=(const ProcessDataRing&) = delete;
 
  void allocate(uint32_t inputCap, uint32_t outputCap, size_t capacityCycles);
 
  void clear();
 
  bool allocated() const { return capacity_ != 0; }
 
  void write(uint64_t timestampNs, int wkc, std::span<const uint8_t> inputs,
             std::span<const uint8_t> outputs);
 
  uint64_t head() const { return head_.load(std::memory_order_acquire); }
 
  size_t capacity() const { return capacity_; }
 
  uint64_t oldestValidSeq() const;
 
  bool readRecord(uint64_t seq, Record& out) const;
 
 private:
  // Per-record payload layout in buffer_ (accessed by memcpy, so no alignment constraints):
  //   [0] timestampNs (u64)  [8] wkc (i32)  [12] inputBytes (u32)  [16] outputBytes (u32)
  //   [20] input region (inputCap_ bytes)   [20+inputCap_] output region (outputCap_ bytes)
  static constexpr size_t kHeaderBytes = 20;
 
  std::vector<uint8_t> buffer_;  // capacity_ * stride_ bytes of payload records
  std::vector<std::atomic<uint64_t>>
      seqWords_;  // one publication word per slot; kInvalidSeq = empty
  size_t stride_ = 0;
  size_t capacity_ = 0;
  uint32_t inputCap_ = 0;
  uint32_t outputCap_ = 0;
  std::atomic<uint64_t> head_{0};  // next sequence number to write (monotonic, never wraps)
  bool locked_ = false;            // whether the storage was successfully mlock'd
};
 
}  // namespace mm::node