libxr_rw.hpp

#pragma once
 
#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 "queue.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
  };
 
  Operation() : type(OperationType::NONE) {
    memset(&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)
  {
    switch (type)
    {
      case OperationType::CALLBACK:
        data.callback->Run(in_isr, std::forward<Args>(status)...);
        break;
      case OperationType::BLOCK:
        data.sem_info.sem->PostFromCallback(in_isr);
        break;
      case OperationType::POLLING:
        *data.status = OperationPollingStatus::DONE;
        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 ReadPort;
class WritePort;
 
typedef Operation<ErrorCode> ReadOperation;
 
typedef Operation<ErrorCode> WriteOperation;
 
typedef ErrorCode (*WriteFun)(WritePort &port);
 
typedef ErrorCode (*ReadFun)(ReadPort &port);
 
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 = 2
  };
 
  ReadFun read_fun_ = nullptr;
  LockFreeQueue<uint8_t> *queue_data_ = nullptr;
  size_t read_size_ = 0;
  Mutex mutex_;
  ReadInfoBlock info_;
  std::atomic<BusyState> busy_{BusyState::Idle};
 
  ReadPort(size_t buffer_size = 128)
      : queue_data_(buffer_size > 0 ? new(std::align_val_t(LIBXR_CACHE_LINE_SIZE))
                                          LockFreeQueue<uint8_t>(buffer_size)
                                    : nullptr)
  {
  }
 
  virtual size_t EmptySize()
  {
    ASSERT(queue_data_ != nullptr);
    return queue_data_->EmptySize();
  }
 
  virtual size_t Size()
  {
    ASSERT(queue_data_ != nullptr);
    return queue_data_->Size();
  }
 
  bool Readable() { return read_fun_ != nullptr; }
 
  ReadPort &operator=(ReadFun fun)
  {
    read_fun_ = fun;
    return *this;
  }
 
  void Finish(bool in_isr, ErrorCode ans, ReadInfoBlock &info, uint32_t size)
  {
    read_size_ = size;
    busy_.store(BusyState::Idle, std::memory_order_release);
    info.op.UpdateStatus(in_isr, std::forward<ErrorCode>(ans));
  }
 
  void MarkAsRunning(ReadInfoBlock &info) { info.op.MarkAsRunning(); }
 
  ErrorCode operator()(RawData data, ReadOperation &op)
  {
    if (Readable())
    {
      mutex_.Lock();
 
      BusyState is_busy = busy_.load(std::memory_order_relaxed);
 
      if (is_busy == BusyState::Pending)
      {
        mutex_.Unlock();
        return ErrorCode::BUSY;
      }
 
      while (true)
      {
        busy_.store(BusyState::Idle, std::memory_order_release);
 
        if (queue_data_ != nullptr)
        {
          auto readable_size = queue_data_->Size();
 
          if (readable_size >= data.size_ && readable_size != 0)
          {
            auto ans = queue_data_->PopBatch(reinterpret_cast<uint8_t *>(data.addr_),
                                             data.size_);
            UNUSED(ans);
            read_size_ = data.size_;
            ASSERT(ans == ErrorCode::OK);
            if (op.type != ReadOperation::OperationType::BLOCK)
            {
              op.UpdateStatus(false, ErrorCode::OK);
            }
            mutex_.Unlock();
            return ErrorCode::OK;
          }
        }
 
        info_ = ReadInfoBlock{data, op};
 
        op.MarkAsRunning();
 
        auto ans = read_fun_(*this);
 
        if (ans != ErrorCode::OK)
        {
          BusyState expected = BusyState::Idle;
          if (busy_.compare_exchange_strong(expected, BusyState::Pending,
                                            std::memory_order_acq_rel,
                                            std::memory_order_acquire))
          {
            break;
          }
          else
          {
            expected = BusyState::Pending;
            continue;
          }
        }
        else
        {
          read_size_ = data.size_;
          if (op.type != ReadOperation::OperationType::BLOCK)
          {
            op.UpdateStatus(false, ErrorCode::OK);
          }
          mutex_.Unlock();
          return ErrorCode::OK;
        }
      }
 
      mutex_.Unlock();
 
      if (op.type == ReadOperation::OperationType::BLOCK)
      {
        return op.data.sem_info.sem->Wait(op.data.sem_info.timeout);
      }
      else
      {
        return ErrorCode::OK;
      }
    }
    else
    {
      return ErrorCode::NOT_SUPPORT;
    }
  }
 
  virtual void ProcessPendingReads(bool in_isr)
  {
    ASSERT(queue_data_ != nullptr);
 
    if (in_isr)
    {
      auto is_busy = busy_.load(std::memory_order_relaxed);
 
      if (is_busy == BusyState::Pending)
      {
        if (queue_data_->Size() >= info_.data.size_)
        {
          if (info_.data.size_ > 0)
          {
            auto ans = queue_data_->PopBatch(
                reinterpret_cast<uint8_t *>(info_.data.addr_), info_.data.size_);
            UNUSED(ans);
            ASSERT(ans == ErrorCode::OK);
          }
          Finish(in_isr, ErrorCode::OK, info_, info_.data.size_);
        }
      }
      else if (is_busy == BusyState::Idle)
      {
        busy_.store(BusyState::Event, std::memory_order_release);
      }
    }
    else
    {
      LibXR::Mutex::LockGuard lock_guard(mutex_);
      if (busy_.load(std::memory_order_relaxed) == BusyState::Pending)
      {
        if (queue_data_->Size() >= info_.data.size_)
        {
          if (info_.data.size_ > 0)
          {
            auto ans = queue_data_->PopBatch(
                reinterpret_cast<uint8_t *>(info_.data.addr_), info_.data.size_);
            UNUSED(ans);
            ASSERT(ans == ErrorCode::OK);
          }
          Finish(in_isr, ErrorCode::OK, info_, info_.data.size_);
        }
      }
    }
  }
 
  virtual void Reset()
  {
    ASSERT(queue_data_ != nullptr);
    Mutex::LockGuard lock_guard(mutex_);
    queue_data_->Reset();
    read_size_ = 0;
  }
};
 
class WritePort
{
 public:
  WriteFun write_fun_ = nullptr;
  LockFreeQueue<WriteInfoBlock> *queue_info_;
  LockFreeQueue<uint8_t> *queue_data_;
  Mutex mutex_;
  size_t write_size_ = 0;
 
  WritePort(size_t queue_size = 3, size_t buffer_size = 128)
      : queue_info_(new(std::align_val_t(LIBXR_CACHE_LINE_SIZE))
                        LockFreeQueue<WriteInfoBlock>(queue_size)),
        queue_data_(buffer_size > 0 ? new(std::align_val_t(LIBXR_CACHE_LINE_SIZE))
                                          LockFreeQueue<uint8_t>(buffer_size)
                                    : nullptr)
  {
  }
 
  virtual size_t EmptySize()
  {
    ASSERT(queue_data_ != nullptr);
    return queue_data_->EmptySize();
  }
 
  virtual size_t Size()
  {
    ASSERT(queue_data_ != nullptr);
    return queue_data_->Size();
  }
 
  bool Writable() { return write_fun_ != nullptr; }
 
  WritePort &operator=(WriteFun fun)
  {
    write_fun_ = fun;
    return *this;
  }
 
  void Finish(bool in_isr, ErrorCode ans, WriteInfoBlock &info, uint32_t size)
  {
    write_size_ = size;
    info.op.UpdateStatus(in_isr, std::forward<ErrorCode>(ans));
  }
 
  void MarkAsRunning(WriteOperation &op) { op.MarkAsRunning(); }
 
  ErrorCode operator()(ConstRawData data, WriteOperation &op)
  {
    if (Writable())
    {
      if (data.size_ == 0)
      {
        write_size_ = 0;
        if (op.type != WriteOperation::OperationType::BLOCK)
        {
          op.UpdateStatus(false, ErrorCode::OK);
        }
        return ErrorCode::OK;
      }
 
      mutex_.Lock();
 
      if (queue_info_->EmptySize() < 1)
      {
        mutex_.Unlock();
        return ErrorCode::FULL;
      }
 
      if (queue_data_)
      {
        if (queue_data_->EmptySize() < data.size_)
        {
          mutex_.Unlock();
          return ErrorCode::FULL;
        }
 
        auto ans = queue_data_->PushBatch(reinterpret_cast<const uint8_t *>(data.addr_),
                                          data.size_);
        UNUSED(ans);
        ASSERT(ans == ErrorCode::OK);
 
        WriteInfoBlock info{data, op};
        ans = queue_info_->Push(info);
 
        ASSERT(ans == ErrorCode::OK);
 
        op.MarkAsRunning();
 
        ans = write_fun_(*this);
 
        mutex_.Unlock();
 
        if (ans == ErrorCode::OK)
        {
          write_size_ = data.size_;
          if (op.type != WriteOperation::OperationType::BLOCK)
          {
            op.UpdateStatus(false, ErrorCode::OK);
          }
          return ErrorCode::OK;
        }
 
        if (op.type == WriteOperation::OperationType::BLOCK)
        {
          return op.data.sem_info.sem->Wait(op.data.sem_info.timeout);
        }
 
        return ErrorCode::OK;
      }
      else
      {
        WriteInfoBlock info{data, op};
        auto ans = queue_info_->Push(info);
 
        ASSERT(ans == ErrorCode::OK);
 
        op.MarkAsRunning();
 
        ans = write_fun_(*this);
 
        mutex_.Unlock();
 
        if (ans == ErrorCode::OK)
        {
          write_size_ = data.size_;
          if (op.type != WriteOperation::OperationType::BLOCK)
          {
            op.UpdateStatus(false, ErrorCode::OK);
          }
          return ErrorCode::OK;
        }
 
        if (op.type == WriteOperation::OperationType::BLOCK)
        {
          return op.data.sem_info.sem->Wait(op.data.sem_info.timeout);
        }
        else
        {
          return ErrorCode::OK;
        }
      }
    }
    else
    {
      return ErrorCode::NOT_SUPPORT;
    }
  }
 
  virtual void Reset()
  {
    ASSERT(queue_data_ != nullptr);
    Mutex::LockGuard lock_guard(mutex_);
    queue_info_->Reset();
    queue_data_->Reset();
    write_size_ = 0;
  }
};
 
class STDIO
{
 public:
  // NOLINTBEGIN
  static inline ReadPort *read_ = nullptr;    
  static inline WritePort *write_ = 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
  {
#if LIBXR_PRINTF_BUFFER_SIZE > 0
    if (!STDIO::write_ || !STDIO::write_->Writable())
    {
      return -1;
    }
 
    static LibXR::Mutex mutex;  // NOLINT
 
    LibXR::Mutex::LockGuard lock_guard(mutex);
 
    va_list args;
    va_start(args, fmt);
    int len = vsnprintf(STDIO::printf_buff_, LIBXR_PRINTF_BUFFER_SIZE, fmt, args);
    va_end(args);
 
    // Check result and limit length
    if (len < 0)
    {
      return -1;
    }
    if (static_cast<size_t>(len) >= LIBXR_PRINTF_BUFFER_SIZE)
    {
      len = LIBXR_PRINTF_BUFFER_SIZE - 1;
    }
 
    ConstRawData data = {reinterpret_cast<const uint8_t *>(STDIO::printf_buff_),
                         static_cast<size_t>(len)};
 
    static WriteOperation op;  // NOLINT
    return static_cast<int>(STDIO::write_->operator()(data, op));
#else
    UNUSED(fmt);
    return 0;
#endif
  }
};
}  // namespace LibXR