write_port.hpp

#pragma once
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <string_view>
 
#include "lockfree_queue.hpp"
#include "operation.hpp"
 
namespace LibXR
{
 
class WritePort
{
 public:
  // Exposed low-level state and helpers for the write-path core. Some members stay public
  // because low-level libxr tests and backend glue inspect them directly. Keep the
  // boundary explicit instead of introducing a fake private wall that tests cannot use.
  // 写路径核心的低层状态与辅助类型。部分成员保持 public，
  // 是因为 libxr 的底层测试与后端胶水层会直接检查它们。
  // 这里保持显式边界即可，不做测试本身也用不上的“伪私有化”。
 
  // Write BLOCK states:
  // LOCKED = submit path owns queue mutation
  // BLOCK_PUBLISHING = BLOCK submit path is publishing queue metadata
  // BLOCK_WAITING = waiter armed, completion not claimed yet
  // BLOCK_CLAIMED = final wakeup belongs to the waiter
  // BLOCK_DETACHED = timeout detached the waiter
  // RESETTING = fail-and-clear path owns queue mutation
  // The same semaphore may be reused only after the previous BLOCK call
  // returns and the port goes back to IDLE.
  // 写 BLOCK 状态：
  // LOCKED = 提交路径占有队列修改权
  // BLOCK_PUBLISHING = BLOCK 提交路径正在发布队列元数据
  // BLOCK_WAITING = waiter 已挂起，完成尚未 claim
  // BLOCK_CLAIMED = 最终唤醒已经归 waiter 所有
  // BLOCK_DETACHED = timeout 已把 waiter 分离
  // RESETTING = fail-and-clear 路径占有队列修改权
  // 同一个信号量只能在上一次 BLOCK 调用返回、端口回到 IDLE 后复用。
  enum class BusyState : uint32_t
  {
    LOCKED =
        0,  
    BLOCK_PUBLISHING = 1,  
    BLOCK_WAITING = 2,  
    BLOCK_CLAIMED =
        3,  
    BLOCK_DETACHED = 4,  
    RESETTING = 5,        
    IDLE = UINT32_MAX    
  };
 
  WriteFun write_fun_ = nullptr;  
  LockFreeQueue<WriteInfoBlock>* queue_info_ =
      nullptr;  
  LockFreeQueue<uint8_t>* queue_data_ =
      nullptr;  
  std::atomic<BusyState> busy_{BusyState::IDLE};  
  ErrorCode block_result_ = ErrorCode::OK;  
 
  // Stream batch facade.
  // Stream 负责一次批次的累积写入与提交。
  class Stream
  {
   public:
    Stream(LibXR::WritePort* port, LibXR::WriteOperation op);
 
    ~Stream();
 
    [[nodiscard]] ErrorCode Write(ConstRawData data);
 
    [[nodiscard]] ErrorCode Write(std::string_view text)
    {
      return Write(ConstRawData{text.data(), text.size()});
    }
 
    Stream& operator<<(const ConstRawData& data);
 
    ErrorCode Commit();
 
    [[nodiscard]] ErrorCode Acquire();
 
    [[nodiscard]] size_t EmptySize() const
    {
      return owns_port_ ? (batch_capacity_ - buffered_size_) : 0;
    }
 
   private:
    [[nodiscard]] ErrorCode SubmitBuffered();
 
    void Release();
 
    LibXR::WritePort* port_;  
    LibXR::WriteOperation op_;  
    size_t batch_capacity_ = 0;  
    size_t buffered_size_ = 0;  
    bool owns_port_ = false;  
  };
 
  WritePort(size_t queue_size = 3, size_t buffer_size = 128);
 
  size_t EmptySize();
 
  size_t Size();
 
  bool Writable();
 
  WritePort& operator=(WriteFun fun);
 
  void Finish(bool in_isr, ErrorCode ans, WriteInfoBlock& info);
 
  void MarkAsRunning(WriteOperation& op);
 
  ErrorCode operator()(ConstRawData data, WriteOperation& op, bool in_isr = false);
 
  void FailAndClearAll(ErrorCode reason, bool in_isr);
 
  ErrorCode CommitWrite(ConstRawData data, WriteOperation& op, bool pushed = false,
                        bool in_isr = false);
};
 
}  // namespace LibXR