operation.hpp

#pragma once
 
#include <atomic>
#include <cstdint>
#include <type_traits>
#include <utility>
 
#include "libxr_cb.hpp"
#include "libxr_def.hpp"
#include "libxr_type.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 : uint32_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(std::atomic<OperationPollingStatus>& status)
      : data{.status = &status}, type(OperationType::POLLING)
  {
  }
 
  Operation(const Operation&) = default;
  Operation(Operation&&) noexcept = default;
 
  Operation& operator=(const Operation&) = default;
 
  Operation& operator=(Operation&&) noexcept = default;
 
  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:
        // 信号量只负责唤醒，结果由调用方另行保存。
        // The semaphore only wakes the caller; the result is stored separately.
        data.sem_info.sem->PostFromCallback(in_isr);
        break;
      case OperationType::POLLING:
        data.status->store((status == ErrorCode::OK) ? OperationPollingStatus::DONE
                                                     : OperationPollingStatus::ERROR,
                           std::memory_order_release);
        break;
      case OperationType::NONE:
        break;
    }
  }
 
  void MarkAsRunning()
  {
    if (type == OperationType::POLLING)
    {
      data.status->store(OperationPollingStatus::RUNNING, std::memory_order_release);
    }
  }
 
  union
  {
    Callback* callback;  
    struct
    {
      Semaphore* sem;    
      uint32_t timeout;  
    } sem_info;
    std::atomic<OperationPollingStatus>* status;
  } data;
 
  OperationType type;
};
 
class AsyncBlockWait
{
 public:
  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)
  {
    DEV_ASSERT(sem_ != nullptr);
    auto wait_ans = sem_->Wait(timeout);
    if (wait_ans == ErrorCode::OK)
    {
      DEV_ASSERT(state_.load(std::memory_order_acquire) == State::CLAIMED);
      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;
    }
 
    DEV_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;
    }
 
    ErrorCode finish_wait_ans;
    do
    {
      finish_wait_ans = sem_->Wait(UINT32_MAX);
    } while (finish_wait_ans == ErrorCode::TIMEOUT);
    REQUIRE(finish_wait_ans == ErrorCode::OK);
    state_.store(State::IDLE, std::memory_order_release);
    return result_;
  }
 
  bool TryPost(bool in_isr, ErrorCode ec)
  {
    DEV_ASSERT_FROM_CALLBACK(sem_ != nullptr, in_isr);
 
    State expected = State::PENDING;
    if (!state_.compare_exchange_strong(expected, State::CLAIMED,
                                        std::memory_order_acq_rel,
                                        std::memory_order_acquire))
    {
      DEV_ASSERT_FROM_CALLBACK(expected == State::DETACHED || expected == State::IDLE,
                               in_isr);
      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;
 
// RW 元数据存放在字节型 SPSC 环中，因此其指针/枚举操作描述必须保持可平凡复制。
// Byte-backed SPSC storage requires trivially copyable operation descriptors.
static_assert(std::is_trivially_copyable_v<ReadOperation>);
static_assert(std::is_trivially_copyable_v<WriteOperation>);
static_assert(std::is_trivially_destructible_v<ReadOperation>);
static_assert(std::is_trivially_destructible_v<WriteOperation>);
 
typedef void (*WriteFun)(WritePort& port, bool in_isr);
 
}  // namespace LibXR