spoe_process_data.h

Go to the documentation of this file.

#pragma once
 
// The pieces that carry SPoE process data between the RT thread and the exchange thread of one
// drive. The RT thread never waits on either of them.
//
// The drive buffers its input frames and returns them in batches, so the inputs travel through a
// queue and the RT thread takes one frame per cycle. That paces the frames to the RT cycle, which
// is what lets monitoring see every frame in order. The outputs are latest-wins: a frame the
// exchange thread did not send yet is replaced by a newer one.
 
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <span>
#include <vector>
 
namespace mm::comm::spoe {
 
constexpr std::size_t kInputPopLimit = 500;
 
class SpoeOutputSlot {
 public:
  void reset(std::size_t frameBytes);
 
  void write(std::span<const uint8_t> frame);
 
  std::span<const uint8_t> read();
 
 private:
  static constexpr uint8_t kFresh = 0x04;
  static constexpr uint8_t kIndexMask = 0x03;
 
  std::array<std::vector<uint8_t>, 3> buffers_;
  uint8_t writeIndex_ = 0;
  uint8_t readIndex_ = 1;
  std::atomic<uint8_t> middle_{2};
};
 
class SpoeInputQueue {
 public:
  void reset(std::size_t frameBytes, std::size_t capacity);
 
  bool push(std::span<const uint8_t> frame);
 
  bool pop(std::span<uint8_t> frame);
 
  std::size_t dropOldest(std::size_t keep);
 
  std::size_t size() const;
 
 private:
  std::vector<uint8_t> storage_;
  std::size_t frameBytes_ = 0;
  std::size_t capacity_ = 0;
  std::atomic<std::size_t> head_{0};  // written by the producer
  std::atomic<std::size_t> tail_{0};  // written by the consumer
};
 
class SpoeFrameAssembler {
 public:
  void reset(std::size_t frameBytes);
 
  std::vector<uint8_t> add(std::span<const uint8_t> reply);
 
  uint64_t framingLosses() const { return framingLosses_; }
 
 private:
  std::size_t frameBytes_ = 0;
  std::vector<uint8_t> partial_;
  uint64_t framingLosses_ = 0;
};
 
}  // namespace mm::comm::spoe