libxr_rw.hpp

#pragma once
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <limits>
#include <string_view>
#include <tuple>
#include <utility>
 
#include "libxr_cb.hpp"
#include "libxr_def.hpp"
#include "libxr_mem.hpp"
#include "print/print_api.hpp"
#include "libxr_type.hpp"
#include "lockfree_queue.hpp"
#include "mutex.hpp"
#include "semaphore.hpp"
 
namespace LibXR
{
 
template <typename Args>
class Operation
{
 public:
  using Callback = LibXR::Callback<Args>;
 
  enum class OperationType : uint8_t
  {
    CALLBACK,
    BLOCK,
    POLLING,
    NONE
  };
 
  enum class OperationPollingStatus : uint8_t
  {
    READY,
    RUNNING,
    DONE,
    ERROR
  };
 
  Operation() : data{nullptr}, type(OperationType::NONE) {}
 
  Operation(Semaphore& sem, uint32_t timeout = UINT32_MAX)
      : data{.sem_info = {&sem, timeout}}, type(OperationType::BLOCK)
  {}
 
  Operation(Callback& callback)
      : data{.callback = &callback}, type(OperationType::CALLBACK)
  {}
 
  Operation(OperationPollingStatus& status)
      : data{.status = &status}, type(OperationType::POLLING)
  {}
 
  Operation(const Operation& op) : data{nullptr}, type(OperationType::NONE)
  {
    *this = op;
  }
 
  Operation(Operation&& op) noexcept : data{nullptr}, type(OperationType::NONE)
  {
    *this = std::move(op);
  }
 
  Operation& operator=(const Operation& op)
  {
    if (this != &op)
    {
      type = op.type;
      switch (type)
      {
        case OperationType::CALLBACK:
          data.callback = op.data.callback;
          break;
        case OperationType::BLOCK:
          data.sem_info.sem = op.data.sem_info.sem;
          data.sem_info.timeout = op.data.sem_info.timeout;
          break;
        case OperationType::POLLING:
          data.status = op.data.status;
          break;
        case OperationType::NONE:
          data.callback = nullptr;
          break;
      }
    }
    return *this;
  }
 
  Operation& operator=(Operation&& op) noexcept
  {
    if (this != &op)
    {
      type = op.type;
      switch (type)
      {
        case OperationType::CALLBACK:
          data.callback = op.data.callback;
          break;
        case OperationType::BLOCK:
          data.sem_info.sem = op.data.sem_info.sem;
          data.sem_info.timeout = op.data.sem_info.timeout;
          break;
        case OperationType::POLLING:
          data.status = op.data.status;
          break;
        case OperationType::NONE:
          data.callback = nullptr;
          break;
      }
    }
    return *this;
  }
 
  template <typename Status>
  void UpdateStatus(bool in_isr, Status&& status)
  {
    switch (type)
    {
      case OperationType::CALLBACK:
        data.callback->Run(in_isr, std::forward<Status>(status));
        break;
      case OperationType::BLOCK:
        // BLOCK waits are signaled by semaphore only; the owning port keeps the
        // final ErrorCode in its block_result_ handoff state.
        // BLOCK 只通过信号量唤醒；最终 ErrorCode 由端口侧 block_result_ 交接。
        data.sem_info.sem->PostFromCallback(in_isr);
        break;
      case OperationType::POLLING:
        *data.status = (status == ErrorCode::OK) ? OperationPollingStatus::DONE
                                                 : OperationPollingStatus::ERROR;
        break;
      case OperationType::NONE:
        break;
    }
  }
 
  void MarkAsRunning()
  {
    if (type == OperationType::POLLING)
    {
      *data.status = OperationPollingStatus::RUNNING;
    }
  }
 
  union
  {
    Callback* callback;
    struct
    {
      Semaphore* sem;
      uint32_t timeout;
    } sem_info;
    OperationPollingStatus* status;
  } data;
 
  OperationType type;
};
 
class AsyncBlockWait
{
 public:
  // Keep the waiter state 32-bit wide so STM32 builds stay within the
  // project-wide atomic shim boundary.
  enum class State : uint32_t
  {
    IDLE = 0,
    PENDING = 1,
    CLAIMED = 2,
    DETACHED = 3,
  };
 
  void Start(Semaphore& sem)
  {
    sem_ = &sem;
    result_ = ErrorCode::OK;
    state_.store(State::PENDING, std::memory_order_release);
  }
 
  void Cancel() { state_.store(State::IDLE, std::memory_order_release); }
 
  ErrorCode Wait(uint32_t timeout)
  {
    ASSERT(sem_ != nullptr);
    auto wait_ans = sem_->Wait(timeout);
    if (wait_ans == ErrorCode::OK)
    {
#ifdef LIBXR_DEBUG_BUILD
      ASSERT(state_.load(std::memory_order_acquire) == State::CLAIMED);
#endif
      state_.store(State::IDLE, std::memory_order_release);
      return result_;
    }
 
    State expected = State::PENDING;
    if (state_.compare_exchange_strong(expected, State::DETACHED,
                                       std::memory_order_acq_rel,
                                       std::memory_order_acquire))
    {
      return ErrorCode::TIMEOUT;
    }
 
    ASSERT(expected == State::CLAIMED || expected == State::DETACHED ||
           expected == State::IDLE);
    if (expected == State::DETACHED)
    {
      state_.store(State::IDLE, std::memory_order_release);
      return ErrorCode::TIMEOUT;
    }
    if (expected == State::IDLE)
    {
      return ErrorCode::TIMEOUT;
    }
 
    auto finish_wait_ans = sem_->Wait(UINT32_MAX);
    UNUSED(finish_wait_ans);
    ASSERT(finish_wait_ans == ErrorCode::OK);
    state_.store(State::IDLE, std::memory_order_release);
    return result_;
  }
 
  bool TryPost(bool in_isr, ErrorCode ec)
  {
    ASSERT(sem_ != nullptr);
 
    State expected = State::PENDING;
    if (!state_.compare_exchange_strong(expected, State::CLAIMED,
                                        std::memory_order_acq_rel,
                                        std::memory_order_acquire))
    {
      ASSERT(expected == State::DETACHED || expected == State::IDLE);
      if (expected == State::DETACHED)
      {
        expected = State::DETACHED;
        (void)state_.compare_exchange_strong(expected, State::IDLE,
                                             std::memory_order_acq_rel,
                                             std::memory_order_acquire);
      }
      return false;
    }
 
    result_ = ec;
    sem_->PostFromCallback(in_isr);
    return true;
  }
 
 private:
  Semaphore* sem_ = nullptr;
  std::atomic<State> state_{State::IDLE};
  ErrorCode result_ = ErrorCode::OK;
};
 
class ReadPort;
class WritePort;
 
typedef Operation<ErrorCode> ReadOperation;
 
typedef Operation<ErrorCode> WriteOperation;
 
typedef ErrorCode (*WriteFun)(WritePort& port, bool in_isr);
 
typedef ErrorCode (*ReadFun)(ReadPort& port, bool in_isr);
 
typedef struct
{
  RawData data;      
  ReadOperation op;  
} ReadInfoBlock;
 
typedef struct
{
  ConstRawData data;  
  WriteOperation op;  
} WriteInfoBlock;
 
class ReadPort
{
 public:
  // Exposed low-level state and helpers for the read-path core. Some members stay public
  // because low-level libxr tests and driver glue inspect them directly.
  // 读路径核心的低层状态与辅助类型。部分成员保持 public，
  // 是因为 libxr 的底层测试与驱动胶水层会直接检查它们。
 
  // Read BLOCK states:
  // PENDING = waiting for queue-fed completion after read_fun_ was notified
  // BLOCK_CLAIMED = wakeup now belongs to the waiter
  // BLOCK_DETACHED = timeout/reset detached the waiter
  // The same semaphore may be reused only after the previous BLOCK call
  // returns and the port goes back to IDLE.
  // 读 BLOCK 状态：
  // PENDING = 已通知 read_fun_，等待队列侧完成
  // BLOCK_CLAIMED = 唤醒已经归当前 waiter 所有
  // BLOCK_DETACHED = timeout/reset 已把 waiter 分离
  // 同一个信号量只能在上一次 BLOCK 调用返回、端口回到 IDLE 后复用。
  enum class BusyState : uint32_t
  {
    IDLE = 0,     
    PENDING = 1,  
    BLOCK_CLAIMED = 2,   
    BLOCK_DETACHED = 3,  
    EVENT = UINT32_MAX   
  };
 
  ReadFun read_fun_ = nullptr;  
  LockFreeQueue<uint8_t>* queue_data_ = nullptr;  
  ReadInfoBlock info_{};  
  std::atomic<BusyState> busy_{BusyState::IDLE};  
  ErrorCode block_result_ = ErrorCode::OK;  
 
  ReadPort(size_t buffer_size = 128);
 
  size_t EmptySize();
 
  size_t Size();
 
  bool Readable();
 
  ReadPort& operator=(ReadFun fun);
 
  void Finish(bool in_isr, ErrorCode ans, ReadInfoBlock& info);
 
  void MarkAsRunning(ReadInfoBlock& info);
 
  ErrorCode operator()(RawData data, ReadOperation& op, bool in_isr = false);
 
  virtual void OnRxDequeue(bool) {}
 
  void ProcessPendingReads(bool in_isr);
 
  void Reset();
};
 
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/reset detached the waiter
  // RESETTING = reset 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/reset 已把 waiter 分离
  // RESETTING = reset 路径占有队列修改权
  // 同一个信号量只能在上一次 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();
 
    void Discard();
 
    [[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;  
  };
 
  // Direct WritePort operations and state-machine entrypoints.
  // WritePort 本体的直接写接口与状态机入口。
  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 Reset();
 
  ErrorCode CommitWrite(ConstRawData data, WriteOperation& op, bool pushed = false,
                        bool in_isr = false);
};
 
class STDIO
{
 public:
  // Shared global stdio binding state.
  // 共享的全局 stdio 绑定状态。
  // NOLINTBEGIN
  static inline ReadPort* read_ = nullptr;    
  static inline WritePort* write_ = nullptr;  
  static inline LibXR::Mutex* write_mutex_ =
      nullptr;  
  static inline LibXR::WritePort::Stream* write_stream_ =
      nullptr;  
  // NOLINTEND
 
 private:
  // Compiled-format bridge shared by the brace and printf frontends.
  // brace 与 printf 两个前端共用的编译格式桥接层。
 
  class CompiledSink
  {
   public:
    explicit CompiledSink(WritePort::Stream& stream);
 
    [[nodiscard]] ErrorCode Write(std::string_view chunk);
 
    [[nodiscard]] size_t RetainedSize() const { return retained_size_; }
 
   private:
    WritePort::Stream& stream_;  
    size_t retained_size_ = 0;  
    bool saturated_ = false;  
  };
 
  using CompiledWriteFun = ErrorCode (*)(void* context, CompiledSink& sink);
 
  template <typename CompiledFormat, typename... Args>
  struct CompiledCall
  {
    const CompiledFormat& format;  
    std::tuple<Args&&...> args;  
 
    [[nodiscard]] static ErrorCode Write(void* context, CompiledSink& sink)
    {
      auto& compiled_call = *static_cast<CompiledCall*>(context);
      return std::apply(
          [&](auto&&... unpacked) -> ErrorCode
          {
            return Print::Write(sink, compiled_call.format,
                                std::forward<decltype(unpacked)>(unpacked)...);
          },
          compiled_call.args);
    }
  };
 
  [[nodiscard]] static int WriteCompiledToStream(WritePort::Stream& stream,
                                                 void* context,
                                                 CompiledWriteFun write_fun);
 
  [[nodiscard]] static int WriteCompiledSession(void* context, CompiledWriteFun write_fun);
 
  template <typename Call>
  [[nodiscard]] static int RunCompiledSession(Call& call)
  {
    if (!BeginWriteSession())
    {
      return -1;
    }
 
    return WriteCompiledSession(&call, &Call::Write);
  }
 
  template <typename CompiledFormat, typename... Args>
  [[nodiscard]] static int RunCompiled(const CompiledFormat& format, Args&&... args)
  {
    CompiledCall<CompiledFormat, Args...> call{
        format, std::forward_as_tuple(std::forward<Args>(args)...)};
    return RunCompiledSession(call);
  }
 
  [[nodiscard]] static bool BeginWriteSession();
 
  [[nodiscard]] static int FinishWriteSession(WritePort::Stream& stream,
                                              size_t retained_size,
                                              ErrorCode format_result);
 
 public:
 
  template <Print::Text Source, typename... Args>
  static int Print(Args&&... args)
  {
    constexpr LibXR::Format<Source> format{};
    return RunCompiled(format, std::forward<Args>(args)...);
  }
 
  template <Print::Text Source, typename... Args>
  static int Printf(Args&&... args)
  {
    constexpr auto format = Print::Printf::Build<Source>();
    return RunCompiled(format, std::forward<Args>(args)...);
  }
};
}  // namespace LibXR