sync.hpp

#pragma once
 
#include "../topic.hpp"
 
namespace LibXR
{
struct Topic::SyncBlock : public SuberBlock
{
  enum WaitState : uint32_t
  {
    WAIT_IDLE = 0,    
    WAITING = 1,      
    WAIT_CLAIMED = 2  
  };
 
  void* buff_addr;  
  void (*copy_payload)(
      void* dst,
      void* payload_addr);  
  MicrosecondTimestamp
      timestamp;  
  std::atomic<uint32_t> wait_state =
      WAIT_IDLE;  
  Semaphore sem;  
};
 
template <typename Data>
class Topic::SyncSubscriber
{
 public:
  SyncSubscriber(const char* name, Data& data, Domain* domain = nullptr)
      : SyncSubscriber(Topic(WaitTopic(name, UINT32_MAX, domain)), data)
  {
  }
 
  SyncSubscriber(Topic topic, Data& data)
  {
    Topic::CheckSubscriberType<Data>(topic);
 
    block_ = new LockFreeList::Node<SyncBlock>;
    block_->data_.type = SuberType::SYNC;
    block_->data_.timestamp = MicrosecondTimestamp();
    block_->data_.wait_state.store(SyncBlock::WAIT_IDLE, std::memory_order_relaxed);
    block_->data_.buff_addr = &data;
    block_->data_.copy_payload = &Topic::CopyPayload<Data>;
    topic.block_->data_.subers.Add(*block_);
  }
 
  SyncSubscriber(const SyncSubscriber& other) = delete;
 
  SyncSubscriber& operator=(const SyncSubscriber& other) = delete;
 
  SyncSubscriber(SyncSubscriber&& other) noexcept : block_(other.block_)
  {
    other.block_ = nullptr;
  }
 
  SyncSubscriber& operator=(SyncSubscriber&& other) noexcept
  {
    if (this != &other)
    {
      block_ = other.block_;
      other.block_ = nullptr;
    }
    return *this;
  }
 
  ErrorCode Wait(uint32_t timeout = UINT32_MAX)
  {
    ASSERT(block_ != nullptr);
 
    auto& data = block_->data_;
    uint32_t expected = SyncBlock::WAIT_IDLE;
    if (!data.wait_state.compare_exchange_strong(expected, SyncBlock::WAITING,
                                                 std::memory_order_acq_rel,
                                                 std::memory_order_acquire))
    {
      return ErrorCode::BUSY;
    }
 
    auto wait_ans = data.sem.Wait(timeout);
    if (wait_ans == ErrorCode::OK)
    {
      data.wait_state.store(SyncBlock::WAIT_IDLE, std::memory_order_release);
      return ErrorCode::OK;
    }
 
    expected = SyncBlock::WAITING;
    if (data.wait_state.compare_exchange_strong(expected, SyncBlock::WAIT_IDLE,
                                                std::memory_order_acq_rel,
                                                std::memory_order_acquire))
    {
      return wait_ans;
    }
 
    ASSERT(data.wait_state.load(std::memory_order_acquire) == SyncBlock::WAIT_CLAIMED);
 
    auto finish_wait_ans = data.sem.Wait(UINT32_MAX);
    UNUSED(finish_wait_ans);
    ASSERT(finish_wait_ans == ErrorCode::OK);
    data.wait_state.store(SyncBlock::WAIT_IDLE, std::memory_order_release);
    return ErrorCode::OK;
  }
 
  MicrosecondTimestamp GetTimestamp() const { return block_->data_.timestamp; }
 
  LockFreeList::Node<SyncBlock>* block_ =
      nullptr;  
};
}  // namespace LibXR