WritePort class for handling write operations. More...

#include <write_port.hpp>

Collaboration diagram for LibXR::WritePort:

Data Structures
class	Stream

Public Types
enum class	BusyState : uint32_t { LOCKED , BLOCK_PUBLISHING = 1 , BLOCK_WAITING = 2 , BLOCK_CLAIMED , BLOCK_DETACHED = 4 , RESETTING = 5 , 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	FailAndClearAll (ErrorCode reason, bool in_isr)
	失败完成并清空当前所有挂起写操作。

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

Data Fields
WriteFun	write_fun_
	Driver/backend write entry. 底层驱动或后端写入入口。

SPSCQueue< WriteInfoBlock > *	queue_info_
	Metadata queue for pending write batches. 挂起写批次的元数据队列。

SPSCQueue< uint8_t > *	queue_data_
	Payload queue for pending write bytes. 挂起写入字节的数据队列。

std::atomic< BusyState >	busy_
	Shared submit/wait handoff state. 共享的提交/等待交接状态。

ErrorCode	block_result_ = ErrorCode::OK

Detailed Description

WritePort class for handling write operations.

处理写入操作的WritePort类。

Definition at line 18 of file write_port.hpp.

Member Enumeration Documentation

◆ BusyState

enum class LibXR::WritePort::BusyState : uint32_t

strong

Enumerator
LOCKED	Submission path owns queue mutation. 提交路径占有写队列/元数据修改权。
BLOCK_PUBLISHING	BLOCK submitter is publishing queue metadata. BLOCK 提交者正在发布队列元数据。
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/detached; completion must not post. 等待者已超时/被分离，完成侧不能再 Post。
RESETTING	Fail-and-clear owns queue mutation. Fail-and-clear 路径占有写队列/元数据修改权。
IDLE	No active submitter and no armed BLOCK waiter. 没有活动提交者，也没有挂起中的 BLOCK 等待者。

Definition at line 45 of file write_port.hpp.

  {
    LOCKED =
        0,  
    BLOCK_PUBLISHING = 1,  
    BLOCK_WAITING = 2,  
    BLOCK_CLAIMED =
        3,  
    BLOCK_DETACHED = 4,  
    RESETTING = 5,       
    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.
buffer_size	缓存数据字节队列的容量，默认为 128。 Capacity of the cached-byte queue, default is 128.

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

Definition at line 10 of file write_port.cpp.

    : queue_info_(new (std::align_val_t(LibXR::CONCURRENCY_ALIGNMENT))
                      SPSCQueue<WriteInfoBlock>(queue_size)),
      queue_data_(buffer_size > 0 ? new (std::align_val_t(LibXR::CONCURRENCY_ALIGNMENT))
                                        SPSCQueue<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 117 of file write_port.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);
 
    if (op.type == WriteOperation::OperationType::BLOCK)
    {
      // The BLOCK waiter must be armed before queue_info_ becomes visible to a
      // backend/completion thread.
      // queue_info_ 对后端/完成线程可见前，必须先挂起 BLOCK waiter。
      op.MarkAsRunning();
      busy_.store(BusyState::BLOCK_PUBLISHING, std::memory_order_release);
    }
 
    WriteInfoBlock info{data, op};
    ans = queue_info_->Push(info);
 
    ASSERT(ans == ErrorCode::OK);
  }
 
  // Non-BLOCK ops are armed here. BLOCK ops submitted through Stream::SubmitBuffered()
  // were already armed (MarkAsRunning + BLOCK_PUBLISHING) before the metadata was
  // published, so the meta_pushed re-arm below is an idempotent no-op kept only to cover
  // the direct-submit (meta_pushed == false) BLOCK path, which arms via BLOCK_PUBLISHING
  // above. MarkAsRunning only affects POLLING ops, so re-marking a BLOCK op is harmless.
  // 非 BLOCK op 在此挂起。经 Stream::SubmitBuffered() 提交的 BLOCK op 在发布元数据前
  // 已经挂起（MarkAsRunning + BLOCK_PUBLISHING），因此下面 meta_pushed 分支的重复挂起
  // 是幂等空操作，仅为覆盖直接提交（meta_pushed == false）的 BLOCK 路径而保留。
  // MarkAsRunning 只对 POLLING op 生效，重复标记 BLOCK op 无副作用。
  if (op.type != WriteOperation::OperationType::BLOCK)
  {
    op.MarkAsRunning();
  }
  else if (meta_pushed)
  {
    op.MarkAsRunning();
  }
 
  if (op.type == WriteOperation::OperationType::BLOCK)
  {
    BusyState expected = BusyState::BLOCK_PUBLISHING;
    if (!busy_.compare_exchange_strong(expected, BusyState::BLOCK_WAITING,
                                       std::memory_order_acq_rel,
                                       std::memory_order_acquire))
    {
      ASSERT(expected == BusyState::BLOCK_CLAIMED);
    }
  }
 
  ans = write_fun_(*this, in_isr);
 
  if (ans != ErrorCode::PENDING)
  {
    if (op.type == WriteOperation::OperationType::BLOCK)
    {
      auto state = busy_.load(std::memory_order_acquire);
      while (state == BusyState::RESETTING)
      {
        state = busy_.load(std::memory_order_acquire);
      }
 
      if (state == BusyState::BLOCK_CLAIMED)
      {
        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 (state == BusyState::BLOCK_DETACHED)
      {
        busy_.store(BusyState::IDLE, std::memory_order_release);
        return ErrorCode::TIMEOUT;
      }
 
      ASSERT(state == BusyState::BLOCK_WAITING || state == BusyState::IDLE ||
             state == BusyState::LOCKED);
      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 (static_cast<int8_t>(ans) < 0) ? ans : 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)
    {
      while (expected == BusyState::RESETTING)
      {
        expected = busy_.load(std::memory_order_acquire);
      }
 
      // A detached late completion may already have cleared BLOCK_DETACHED
      // back to IDLE before this waiter wakes from timeout.
      // 分离后的迟到完成可能会在当前 waiter 超时醒来前，先把
      // BLOCK_DETACHED 清回 IDLE。
      ASSERT(expected == BusyState::BLOCK_DETACHED || expected == BusyState::IDLE ||
             expected == BusyState::LOCKED);
      if (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 19 of file write_port.cpp.

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

◆ FailAndClearAll()

void WritePort::FailAndClearAll	(	ErrorCode	reason,
		bool	in_isr )

失败完成并清空当前所有挂起写操作。

Fail-complete and clear all currently pending write operations.

Note: Driver-only: call this only after the backend is known to be unavailable.; 仅供驱动层在后端已明确不可用后调用。; The surrounding driver must already guarantee that no new front-end submissions or back-end completion/data events can still arrive for this port.; 外围驱动还必须先保证：这条端口后续不会再收到新的前端提交，也不会再收到新的后端完成或数据事件。; Seeing LOCKED / BLOCK_PUBLISHING / RESETTING here means the caller violated that driver-side precondition.; 若此时仍看到 LOCKED / BLOCK_PUBLISHING / RESETTING，说明调用方违反了上述驱动层前提。; Dev builds fire DEV_ASSERT, while release builds still back off.; 开发期会触发 DEV_ASSERT，发布构建仍保持直接退回。

Parameters

reason	最终失败原因 / Final failure reason
in_isr	是否在 ISR 上下文 / Whether running in ISR context

Definition at line 293 of file write_port.cpp.

{
  ASSERT(queue_data_ != nullptr);
  WriteInfoBlock info{};
 
  while (true)
  {
    auto state = busy_.load(std::memory_order_acquire);
 
    if (state == BusyState::LOCKED)
    {
      DEV_ASSERT_FROM_CALLBACK(false, in_isr);
      return;
    }
 
    if (state == BusyState::BLOCK_PUBLISHING)
    {
      DEV_ASSERT_FROM_CALLBACK(false, in_isr);
      return;
    }
 
    if (state == BusyState::RESETTING)
    {
      DEV_ASSERT_FROM_CALLBACK(false, in_isr);
      return;
    }
 
    if (state == BusyState::IDLE)
    {
      BusyState expected = BusyState::IDLE;
      if (!busy_.compare_exchange_strong(expected, BusyState::RESETTING,
                                         std::memory_order_acq_rel,
                                         std::memory_order_acquire))
      {
        continue;
      }
 
      queue_data_->Reset();
      while (queue_info_->Pop(info) == ErrorCode::OK)
      {
        Finish(in_isr, reason, info);
      }
      block_result_ = ErrorCode::OK;
      busy_.store(BusyState::IDLE, std::memory_order_release);
      return;
    }
 
    if (state == BusyState::BLOCK_WAITING)
    {
      // Keep BLOCK_WAITING visible until Finish() hands the terminal wakeup to
      // the blocked caller. Switching to RESETTING here would break that
      // existing waiter handoff.
      // 这里必须保留 BLOCK_WAITING，直到 Finish() 把最终唤醒交给当前
      // BLOCK waiter；若先切成 RESETTING，会破坏既有 waiter 交接。
      queue_data_->Reset();
      while (queue_info_->Pop(info) == ErrorCode::OK)
      {
        Finish(in_isr, reason, info);
      }
      return;
    }
 
    if (state == BusyState::BLOCK_DETACHED)
    {
      // The waiter is already gone, but BLOCK_DETACHED still blocks reentrant
      // submissions while old queue entries are drained.
      // waiter 已经离开，但 BLOCK_DETACHED 仍能在清理旧队列期间挡住重入提交。
      queue_data_->Reset();
      while (queue_info_->Pop(info) == ErrorCode::OK)
      {
        // The waiter has already detached. Finish() will clear the local state
        // without re-posting that waiter.
        // waiter 已经分离。Finish() 会清理本地状态，但不会重新唤醒该 waiter。
        Finish(in_isr, reason, info);
      }
      block_result_ = ErrorCode::OK;
      busy_.store(BusyState::IDLE, std::memory_order_release);
      return;
    }
 
    if (state == BusyState::BLOCK_CLAIMED)
    {
      return;
    }
  }
}

◆ Finish()

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

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

该函数在写入操作完成时更新对应 info 的状态，并调用 UpdateStatus 更新 op。 This function updates the status stored in info and calls UpdateStatus on op when a write operation completes.

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 39 of file write_port.cpp.

{
  if (info.op.type == WriteOperation::OperationType::BLOCK)
  {
    block_result_ = ans;
 
    // Write completion claims the active BLOCK waiter and hands the wakeup to it.
    // 写完成 claim 当前 BLOCK 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;
    }
 
    // The waiter may have timed out and detached before this late completion is
    // reported.
    // waiter 可能已经先超时分离，随后迟到完成才上报。
    if (expected == BusyState::BLOCK_PUBLISHING)
    {
      expected = BusyState::BLOCK_PUBLISHING;
      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 || expected == BusyState::IDLE ||
           expected == BusyState::LOCKED || expected == BusyState::RESETTING);
    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 86 of file write_port.cpp.

86{ 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 88 of file write_port.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 33 of file write_port.cpp.

{
  write_fun_ = fun;
  return *this;
}

◆ 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 25 of file write_port.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 31 of file write_port.cpp.

31{ return write_fun_ != nullptr; }

Field Documentation

◆ block_result_

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

Final status for the current BLOCK write. 当前 BLOCK 写入的最终结果。

Definition at line 71 of file write_port.hpp.

◆ busy_

std::atomic<BusyState> LibXR::WritePort::busy_

Initial value:

{

BusyState::IDLE}

Shared submit/wait handoff state. 共享的提交/等待交接状态。

Definition at line 69 of file write_port.hpp.

69 {

70 BusyState::IDLE};

◆ queue_data_

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

Initial value:

=

nullptr

Payload queue for pending write bytes. 挂起写入字节的数据队列。

Definition at line 67 of file write_port.hpp.

◆ queue_info_

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

Initial value:

=

nullptr

Metadata queue for pending write batches. 挂起写批次的元数据队列。

Definition at line 65 of file write_port.hpp.

◆ write_fun_

WriteFun LibXR::WritePort::write_fun_

Initial value:

=

nullptr

Driver/backend write entry. 底层驱动或后端写入入口。

Definition at line 63 of file write_port.hpp.

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

src/core/rw/write_port.hpp
src/core/rw/write_port.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()

◆ FailAndClearAll()

◆ Finish()

◆ MarkAsRunning()

◆ operator()()

◆ operator=()

◆ Size()

◆ Writable()

Field Documentation

◆ block_result_

◆ busy_

◆ queue_data_

◆ queue_info_

◆ write_fun_