libxr_rw.hpp

#pragma once
 
#include <atomic>
#include <cstdarg>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <utility>
 
#include "libxr_cb.hpp"
#include "libxr_def.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() : type(OperationType::NONE) { Memory::FastSet(&data, 0, sizeof(data)); }
 
  Operation(Semaphore& sem, uint32_t timeout = UINT32_MAX) : type(OperationType::BLOCK)
  {
    data.sem_info.sem = &sem;
    data.sem_info.timeout = timeout;
  }
 
  Operation(Callback& callback) : type(OperationType::CALLBACK)
  {
    data.callback = &callback;
  }
 
  Operation(OperationPollingStatus& status) : type(OperationType::POLLING)
  {
    data.status = &status;
  }
 
  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:
          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:
          break;
      }
    }
    return *this;
  }
 
  template <
      typename InitOperation,
      typename = std::enable_if_t<std::is_same_v<std::decay_t<InitOperation>, Operation>>>
  Operation(InitOperation&& op)
  {
    *this = std::forward<InitOperation>(op);
  }
 
  template <typename Status>
  void UpdateStatus(bool in_isr, Status&& status)
  {
    // TODO: 任何操作类型都应拿到执行结果。
    switch (type)
    {
      case OperationType::CALLBACK:
        data.callback->Run(in_isr, std::forward<Status>(status));
        break;
      case OperationType::BLOCK:
        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;
    // TODO: state
  } data;
 
  OperationType type;
};
 
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:
  enum class BusyState : uint32_t
  {
    IDLE = 0,
    PENDING = 1,
    EVENT = UINT32_MAX
  };
 
  ReadFun read_fun_ = nullptr;
  LockFreeQueue<uint8_t>* queue_data_ = nullptr;
  ReadInfoBlock info_;
  std::atomic<BusyState> busy_{BusyState::IDLE};
 
  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:
  enum class LockState : uint32_t
  {
    LOCKED = 0,
    UNLOCKED = UINT32_MAX
  };
 
  WriteFun write_fun_ = nullptr;
  LockFreeQueue<WriteInfoBlock>* queue_info_;
  LockFreeQueue<uint8_t>* queue_data_;
  std::atomic<LockState> lock_{LockState::UNLOCKED};
 
  class Stream
  {
   public:
    Stream(LibXR::WritePort* port, LibXR::WriteOperation op);
 
    ~Stream();
 
    Stream& operator<<(const ConstRawData& data);
 
    ErrorCode Commit();
 
   private:
    LibXR::WritePort* port_;    
    LibXR::WriteOperation op_;  
    size_t cap_;                
    size_t size_ = 0;  
    bool locked_ = 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 Reset();
 
  ErrorCode CommitWrite(ConstRawData data, WriteOperation& op, bool pushed = false,
                        bool in_isr = false);
};
 
class STDIO
{
 public:
  // 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;
#if LIBXR_PRINTF_BUFFER_SIZE > 0
  static inline char
      printf_buff_[LIBXR_PRINTF_BUFFER_SIZE];  
#endif
  // NOLINTEND
 
  static int Printf(const char* fmt, ...);  // NOLINT
};
}  // namespace LibXR