WritePort class for handling write operations. More...

#include <libxr_rw.hpp>

Collaboration diagram for LibXR::WritePort:

Data Structures
class	Stream
	WritePort 的流式写入操作器，支持链式 << 操作和批量提交。 More...

Public Types
enum class	BusyState : uint32_t { LOCKED , BLOCK_WAITING = 1 , BLOCK_CLAIMED , BLOCK_DETACHED = 3 , IDLE = UINT32_MAX }

Public Member Functions
	WritePort (size_t queue_size=3, size_t buffer_size=128)
	构造一个新的 WritePort 对象。 Constructs a new WritePort object.

size_t	EmptySize ()
	获取数据队列的剩余可用空间。 Gets the remaining available space in the data queue.

size_t	Size ()
	获取当前数据队列的已使用大小。 Gets the used size of the current data queue.

bool	Writable ()
	判断端口是否可写。 Checks whether the port is writable.

WritePort &	operator= (WriteFun fun)
	赋值运算符重载，用于设置写入函数。 Overloaded assignment operator to set the write function.

void	Finish (bool in_isr, ErrorCode ans, WriteInfoBlock &info)
	更新写入操作的状态。 Updates the status of the write operation.

void	MarkAsRunning (WriteOperation &op)
	标记写入操作为运行中。 Marks the write operation as running.

ErrorCode	operator() (ConstRawData data, WriteOperation &op, bool in_isr=false)
	执行写入操作。 Performs a write operation.

void	Reset ()
	Resets the WritePort.

ErrorCode	CommitWrite (ConstRawData data, WriteOperation &op, bool pushed=false, bool in_isr=false)
	提交写入操作。 Commits a write operation.

Data Fields
WriteFun	write_fun_ = nullptr

LockFreeQueue< WriteInfoBlock > *	queue_info_

LockFreeQueue< uint8_t > *	queue_data_

std::atomic< BusyState >	busy_ {BusyState::IDLE}

ErrorCode	block_result_ = ErrorCode::OK
	Final status for the current BLOCK write.

Detailed Description

WritePort class for handling write operations.

处理写入操作的WritePort类。

Definition at line 537 of file libxr_rw.hpp.

Member Enumeration Documentation

◆ BusyState

enum class LibXR::WritePort::BusyState : uint32_t

strong

Enumerator
LOCKED	Submission path owns queue mutation. 提交路径占有写队列/元数据修改权。
BLOCK_WAITING	One BLOCK waiter is armed and waiting for final completion. 一个 BLOCK 等待者已经挂起，等待最终完成。
BLOCK_CLAIMED	Final wakeup belongs to the current waiter. 最终唤醒已归当前等待者所有。
BLOCK_DETACHED	Waiter already timed out/reset; completion must not post. 等待者已超时/被分离，完成侧不能再 Post。
IDLE	No active submitter and no armed BLOCK waiter. 没有活动提交者，也没有挂起中的 BLOCK 等待者。

Definition at line 553 of file libxr_rw.hpp.

  {
    LOCKED =
        0,  
    BLOCK_WAITING = 1,  
    BLOCK_CLAIMED =
        2,  
    BLOCK_DETACHED = 3,  
    IDLE = UINT32_MAX    
  };

Constructor & Destructor Documentation

◆ WritePort()

WritePort::WritePort	(	size_t	queue_size = 3,
		size_t	buffer_size = 128 )

构造一个新的 WritePort 对象。 Constructs a new WritePort object.

该构造函数初始化无锁操作队列 queue_info_ 和数据块队列 queue_data_。 This constructor initializes the lock-free operation queue queue_info_ and the data block queue queue_data_.

Parameters

queue_size	队列的大小，默认为 3。 The size of the queue, default is 3.
block_size	缓存的数据的最大大小，默认为 128。 The maximum size of cached data, default is 128.

Note: 包含动态内存分配。 Contains dynamic memory allocation.

Definition at line 265 of file libxr_rw.cpp.

    : 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)
{
}

Member Function Documentation

◆ CommitWrite()

ErrorCode WritePort::CommitWrite	(	ConstRawData	data,
		WriteOperation &	op,
		bool	pushed = false,
		bool	in_isr = false )

提交写入操作。 Commits a write operation.

Parameters

data	写入的原始数据 / Raw data to be written
op	写入操作对象，包含操作类型和同步机制。 Write operation object containing the operation type and synchronization
pushed	数据是否已经推送到缓冲区 / Whether the data has been pushed to the buffer
in_isr	指示是否在中断上下文中执行。 Indicates whether the operation is executed in an interrupt context.

Returns: 返回操作的 ErrorCode，指示操作结果。 Returns an ErrorCode indicating the result of the operation.

Definition at line 356 of file libxr_rw.cpp.

{
  if (!meta_pushed && queue_info_->EmptySize() < 1)
  {
    busy_.store(BusyState::IDLE, std::memory_order_release);
    return ErrorCode::FULL;
  }
 
  ErrorCode ans = ErrorCode::OK;
  if (!meta_pushed)
  {
    if (queue_data_->EmptySize() < data.size_)
    {
      busy_.store(BusyState::IDLE, std::memory_order_release);
      return ErrorCode::FULL;
    }
 
    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();
 
  if (op.type == WriteOperation::OperationType::BLOCK)
  {
    // BLOCK waiter must be armed before write_fun_() runs.
    // 必须先挂起 BLOCK waiter，再调用 write_fun_()。
    busy_.store(BusyState::BLOCK_WAITING, std::memory_order_release);
  }
 
  ans = write_fun_(*this, in_isr);
 
  if (ans != ErrorCode::PENDING)
  {
    if (op.type == WriteOperation::OperationType::BLOCK)
    {
      busy_.store(BusyState::IDLE, std::memory_order_release);
      return ans;
    }
 
    if (!meta_pushed)
    {
      busy_.store(BusyState::IDLE, std::memory_order_release);
    }
 
    if (op.type != WriteOperation::OperationType::BLOCK)
    {
      op.UpdateStatus(in_isr, ans);
    }
    return ErrorCode::OK;
  }
 
  if (op.type == WriteOperation::OperationType::BLOCK)
  {
    ASSERT(!in_isr);
    auto wait_ans = op.data.sem_info.sem->Wait(op.data.sem_info.timeout);
    if (wait_ans == ErrorCode::OK)
    {
      // BLOCK_CLAIMED is always released by the waiter itself.
      // BLOCK_CLAIMED 始终由 waiter 自己释放。
#ifdef LIBXR_DEBUG_BUILD
      auto state = busy_.load(std::memory_order_acquire);
      ASSERT(state == BusyState::BLOCK_CLAIMED);
#endif
      busy_.store(BusyState::IDLE, std::memory_order_release);
      return block_result_;
    }
 
    // Timeout won before completion claimed the waiter.
    // 超时先赢，完成侧还没 claim 当前 waiter。
    BusyState expected = BusyState::BLOCK_WAITING;
    if (busy_.compare_exchange_strong(expected, BusyState::BLOCK_DETACHED,
                                      std::memory_order_acq_rel,
                                      std::memory_order_acquire))
    {
      return ErrorCode::TIMEOUT;
    }
 
    if (expected != BusyState::BLOCK_CLAIMED)
    {
      ASSERT(expected == BusyState::BLOCK_DETACHED);
      busy_.store(BusyState::IDLE, std::memory_order_release);
      return ErrorCode::TIMEOUT;
    }
 
    // Timeout lost after completion had already claimed the waiter.
    // 超时发生得太晚，完成侧已经 claim 了当前 waiter。
    auto finish_wait_ans = op.data.sem_info.sem->Wait(UINT32_MAX);
    UNUSED(finish_wait_ans);
    ASSERT(finish_wait_ans == ErrorCode::OK);
    busy_.store(BusyState::IDLE, std::memory_order_release);
 
    return block_result_;
  }
 
  if (!meta_pushed)
  {
    busy_.store(BusyState::IDLE, std::memory_order_release);
  }
 
  return ErrorCode::OK;
}

◆ EmptySize()

size_t WritePort::EmptySize ( )

获取数据队列的剩余可用空间。 Gets the remaining available space in the data queue.

Returns: 返回数据队列的空闲大小。 Returns the empty size of the data queue.

Definition at line 274 of file libxr_rw.cpp.

{
  ASSERT(queue_data_ != nullptr);
  return queue_data_->EmptySize();
}

◆ Finish()

void WritePort::Finish	(	bool	in_isr,
		ErrorCode	ans,
		WriteInfoBlock &	info )

更新写入操作的状态。 Updates the status of the write operation.

该函数在写入操作过程中更新 write_size_ 并调用 UpdateStatus 方法更新 op 的状态。 This function updates write_size_ and calls UpdateStatus on op during a write operation.

Parameters

in_isr	指示是否在中断上下文中执行。 Indicates whether the operation is executed in an interrupt context.
ans	错误码，用于指示操作的结果。 Error code indicating the result of the operation.
op	需要更新状态的 `WriteOperation` 引用。 Reference to the `WriteOperation` whose status needs to be updated.

Definition at line 294 of file libxr_rw.cpp.

{
  if (info.op.type == WriteOperation::OperationType::BLOCK)
  {
    block_result_ = ans;
 
    // Write completion claims BLOCK_WAITING and hands the wakeup to the waiter.
    // 写完成从 BLOCK_WAITING claim 当前 waiter，并把唤醒交给它。
    BusyState expected = BusyState::BLOCK_WAITING;
    if (busy_.compare_exchange_strong(expected, BusyState::BLOCK_CLAIMED,
                                      std::memory_order_acq_rel,
                                      std::memory_order_acquire))
    {
      info.op.data.sem_info.sem->PostFromCallback(in_isr);
      return;
    }
 
    ASSERT(expected == BusyState::BLOCK_DETACHED);
    if (expected == BusyState::BLOCK_DETACHED)
    {
      expected = BusyState::BLOCK_DETACHED;
      (void)busy_.compare_exchange_strong(expected, BusyState::IDLE,
                                          std::memory_order_acq_rel,
                                          std::memory_order_acquire);
    }
    return;
  }
 
  info.op.UpdateStatus(in_isr, ans);
}

◆ MarkAsRunning()

void WritePort::MarkAsRunning ( WriteOperation & op )

标记写入操作为运行中。 Marks the write operation as running.

该函数用于将 op 标记为运行状态，以指示当前正在进行写入操作。 This function marks op as running to indicate an ongoing write operation.

Parameters

op	需要更新状态的 `WriteOperation` 引用。 Reference to the `WriteOperation` whose status needs to be updated.

Definition at line 325 of file libxr_rw.cpp.

325{ op.MarkAsRunning(); }

◆ operator()()

ErrorCode WritePort::operator()	(	ConstRawData	data,
		WriteOperation &	op,
		bool	in_isr = false )

执行写入操作。 Performs a write operation.

该函数检查端口是否可写，并根据 data.size_ 和 op 的类型执行不同的操作。 This function checks if the port is writable and performs different actions based on data.size_ and the type of op.

Parameters

data	需要写入的原始数据。 Raw data to be written.
op	写入操作对象，包含操作类型和同步机制。 Write operation object containing the operation type and synchronization mechanism.
in_isr	指示是否在中断上下文中执行。 Indicates whether the operation is executed in an interrupt context.

Returns: 返回操作的 ErrorCode，指示操作结果。 Returns an ErrorCode indicating the result of the operation.

Definition at line 327 of file libxr_rw.cpp.

{
  if (Writable())
  {
    if (data.size_ == 0)
    {
      if (op.type != WriteOperation::OperationType::BLOCK)
      {
        op.UpdateStatus(in_isr, ErrorCode::OK);
      }
      return ErrorCode::OK;
    }
 
    BusyState expected = BusyState::IDLE;
    if (!busy_.compare_exchange_strong(expected, BusyState::LOCKED,
                                       std::memory_order_acq_rel,
                                       std::memory_order_acquire))
    {
      return ErrorCode::BUSY;
    }
 
    return CommitWrite(data, op, false, in_isr);
  }
  else
  {
    return ErrorCode::NOT_SUPPORT;
  }
}

◆ operator=()

WritePort & WritePort::operator= ( WriteFun fun )

赋值运算符重载，用于设置写入函数。 Overloaded assignment operator to set the write function.

该函数允许使用 WriteFun 类型的函数对象赋值给 WritePort，从而设置 write_fun_。 This function allows assigning a WriteFun function object to WritePort, setting write_fun_.

Parameters

fun	要分配的写入函数。 The write function to be assigned.

Returns: 返回自身的引用，以支持链式调用。 Returns a reference to itself for chaining.

Definition at line 288 of file libxr_rw.cpp.

{
  write_fun_ = fun;
  return *this;
}

◆ Reset()

void WritePort::Reset ( )

Resets the WritePort.

重置WritePort。

Definition at line 467 of file libxr_rw.cpp.

{
  ASSERT(queue_data_ != nullptr);
  queue_data_->Reset();
  queue_info_->Reset();
 
  auto state = busy_.load(std::memory_order_acquire);
  if (state == BusyState::BLOCK_WAITING)
  {
    // Reset detaches the BLOCK waiter instead of reopening the port directly.
    // Reset 先分离 BLOCK waiter，不直接重开端口。
    BusyState expected = BusyState::BLOCK_WAITING;
    if (busy_.compare_exchange_strong(expected, BusyState::BLOCK_DETACHED,
                                      std::memory_order_acq_rel,
                                      std::memory_order_acquire))
    {
      return;
    }
 
    state = busy_.load(std::memory_order_acquire);
  }
 
  if (state == BusyState::BLOCK_CLAIMED || state == BusyState::BLOCK_DETACHED)
  {
    return;
  }
 
  block_result_ = ErrorCode::OK;
  busy_.store(BusyState::IDLE, std::memory_order_release);
}

◆ Size()

size_t WritePort::Size ( )

获取当前数据队列的已使用大小。 Gets the used size of the current data queue.

Returns: 返回数据队列的已使用大小。 Returns the size of the data queue.

Definition at line 280 of file libxr_rw.cpp.

{
  ASSERT(queue_data_ != nullptr);
  return queue_data_->Size();
}

◆ Writable()

bool WritePort::Writable ( )

判断端口是否可写。 Checks whether the port is writable.

Returns: 如果 write_fun_ 不为空，则返回 true，否则返回 false。 Returns true if write_fun_ is not null, otherwise returns false.

Definition at line 286 of file libxr_rw.cpp.

286{ return write_fun_ != nullptr; }

Field Documentation

◆ block_result_

ErrorCode LibXR::WritePort::block_result_ = ErrorCode::OK

Final status for the current BLOCK write.

Definition at line 571 of file libxr_rw.hpp.

◆ busy_

std::atomic<BusyState> LibXR::WritePort::busy_ {BusyState::IDLE}

Definition at line 570 of file libxr_rw.hpp.

570{BusyState::IDLE};

◆ queue_data_

LockFreeQueue<uint8_t>* LibXR::WritePort::queue_data_

Definition at line 569 of file libxr_rw.hpp.

◆ queue_info_

LockFreeQueue<WriteInfoBlock>* LibXR::WritePort::queue_info_

Definition at line 568 of file libxr_rw.hpp.

◆ write_fun_

WriteFun LibXR::WritePort::write_fun_ = nullptr

Definition at line 567 of file libxr_rw.hpp.

The documentation for this class was generated from the following files:

src/core/libxr_rw.hpp
src/core/libxr_rw.cpp

Data Structures

Public Types

Public Member Functions

Data Fields

Detailed Description

Member Enumeration Documentation

◆ BusyState

Constructor & Destructor Documentation

◆ WritePort()

Member Function Documentation

◆ CommitWrite()

◆ EmptySize()

◆ Finish()

◆ MarkAsRunning()

◆ operator()()

◆ operator=()

◆ Reset()

◆ Size()

◆ Writable()

Field Documentation

◆ block_result_

◆ busy_

◆ queue_data_

◆ queue_info_

◆ write_fun_