PipePort.h

Go to the documentation of this file.

 
#pragma once
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
 
#include "HfCanPort.h"
 
namespace hf::can {
 
class PipeHub {
 public:
  static constexpr std::size_t kMaxBacklog = 1U << 20U;
 
  PipeHub() = default;
  ~PipeHub();
  PipeHub(const PipeHub&) = delete;
  PipeHub& operator=(const PipeHub&) = delete;
 
  bool Start(const std::string& path);
  void Stop();
  bool Running() const noexcept { return running_.load(); }
  uint64_t Forwarded() const noexcept { return forwarded_.load(); }
  std::size_t Participants() const noexcept { return participants_.load(); }
 
 private:
  struct Client;
  void Loop();
 
  std::string path_;
  int listen_fd_{-1};
  int wake_[2]{-1, -1};
  std::thread thread_;
  std::atomic<bool> running_{false};
  std::atomic<uint64_t> forwarded_{0};
  std::atomic<std::size_t> participants_{0};
};
 
class PipePort final : public ICanPort {
 public:
  struct Options {
    std::string path;               
    bool host_hub_if_absent{true};  
    std::size_t tx_queue_frames{256};
    bool fd{false};                 
  };
 
  explicit PipePort(Options options);
  ~PipePort() override;
  PipePort(const PipePort&) = delete;
  PipePort& operator=(const PipePort&) = delete;
 
  bool Open();
  void Close();
  bool IsOpen() const noexcept { return connected_.load(); }
  bool HostsHub() const noexcept { return hub_ != nullptr; }
 
  PortCaps Caps() const override;
  TxResult Send(const CanFrame& frame) override;
  void SetRxSink(RxSink sink, void* ctx) override;
  PortStatus Status() const override;
  bool Recover() override;
  void SetTxDoneSink(TxDoneSink sink, void* ctx) override;
  bool RxBackpressure() const override { return true; }
 
 private:
  bool Connect();
  void IoLoop();
  void Wake() const;
  bool Deliver(const CanFrame& f);
 
  Options opt_;
  std::unique_ptr<PipeHub> hub_;
  int fd_{-1};
  int wake_[2]{-1, -1};
  std::thread io_;
  std::atomic<bool> run_{false};
  std::atomic<bool> connected_{false};
 
  // TX queue (Send → I/O thread).
  std::mutex tx_mu_;
  std::vector<CanFrame> tx_ring_;
  std::size_t tx_head_{0};
  std::size_t tx_count_{0};
  bool tx_was_full_{false};
 
  // Sinks; the mutex keeps a sink alive while it runs (SetRxSink(nullptr) waits).
  std::mutex sink_mu_;
  RxSink rx_sink_{nullptr};
  void* rx_ctx_{nullptr};
  TxDoneSink tx_sink_{nullptr};
  void* tx_ctx_{nullptr};
 
  std::atomic<uint32_t> tx_queue_full_{0};
  uint16_t seq_{0};
  uint32_t epoch_{0};
};
 
}  // namespace hf::can