LibXR::WritePort::Stream Class Reference

Collaboration diagram for LibXR::WritePort::Stream:

Public Member Functions
	Stream (LibXR::WritePort *port, LibXR::WriteOperation op)
	构造流写入对象，并尝试锁定端口。
	~Stream ()
	析构时自动提交已累积的数据并释放锁。
ErrorCode	Write (ConstRawData data)
	追加一个原始数据片段到当前流批次。
ErrorCode	Write (std::string_view text)
	追加一个文本片段到当前流批次。
Stream &	operator<< (const ConstRawData &data)
	追加写入数据的语法糖，忽略返回状态并支持链式调用。
ErrorCode	Commit ()
	手动提交已写入的数据到队列，并释放当前锁。
ErrorCode	Acquire ()
	为当前流批次获取一次可写入的端口所有权。
size_t	EmptySize () const
	获取当前批次还可追加的剩余字节数。

Private Member Functions
ErrorCode	SubmitBuffered ()
	将当前已缓存批次提交给 WritePort。
void	Release ()
	将当前批次的端口所有权归还给 WritePort。

Private Attributes
LibXR::WritePort *	port_
	写端口指针 Pointer to the WritePort
LibXR::WriteOperation	op_
	写操作对象 Write operation object
size_t	batch_capacity_ = 0
	当前批次可用的总容量 Total capacity reserved for the current batch
size_t	buffered_size_ = 0
	当前批次已追加到共享 data queue、但尚未发布对应元数据的字节数 Bytes already appended into the shared data queue for the current batch, but whose metadata has not yet been published
bool	owns_port_ = false
	当前 Stream 是否持有该批次的端口所有权 Whether this Stream currently owns the batch

Detailed Description

Definition at line 73 of file write_port.hpp.

Constructor & Destructor Documentation

Stream()

WritePort::Stream::Stream	(	LibXR::WritePort *	port,
		LibXR::WriteOperation	op )

构造流写入对象，并尝试锁定端口。

Constructs a Stream object and tries to acquire WritePort lock.

Parameters

port	指向 WritePort 的指针 Pointer to WritePort.
op	写操作对象（可重用）Write operation object (can be reused).

Definition at line 5 of file write_stream.cpp.

    : port_(port), op_(op)
{
  UNUSED(Acquire());
}

~Stream()

WritePort::Stream::~Stream

(

)

析构时自动提交已累积的数据并释放锁。

Destructor: automatically commits any accumulated data and releases the lock.

Definition at line 13 of file write_stream.cpp.

{
  if (owns_port_ && buffered_size_ > 0)
  {
    UNUSED(SubmitBuffered());
  }
 
  if (owns_port_)
  {
    Release();
  }
}

Member Function Documentation

Acquire()

ErrorCode WritePort::Stream::Acquire ( )

nodiscard

为当前流批次获取一次可写入的端口所有权。

Acquires append ownership for the current stream batch.

Returns

返回获取结果。Returns the acquisition result.

Definition at line 26 of file write_stream.cpp.

{
  if (owns_port_)
  {
    return ErrorCode::OK;
  }
 
  if (port_ == nullptr)
  {
    return ErrorCode::PTR_NULL;
  }
 
  DEV_ASSERT(port_->queue_info_ != nullptr);
  DEV_ASSERT(port_->queue_data_ != nullptr);
 
  if (!port_->Writable())
  {
    return ErrorCode::NOT_SUPPORT;
  }
 
  BusyState expected = BusyState::IDLE;
  if (!port_->busy_.compare_exchange_strong(expected, BusyState::LOCKED,
                                            std::memory_order_acq_rel,
                                            std::memory_order_acquire))
  {
    return ErrorCode::BUSY;
  }
 
  if (port_->queue_info_->EmptySize() < 1)
  {
    port_->busy_.store(BusyState::IDLE, std::memory_order_release);
    return ErrorCode::FULL;
  }
 
  owns_port_ = true;
  batch_capacity_ = port_->queue_data_->EmptySize();
  return ErrorCode::OK;
}

Commit()

ErrorCode WritePort::Stream::Commit

(

)

手动提交已写入的数据到队列，并释放当前锁。

Manually commit accumulated data to the queue, then release the current lock.

调用后会发布当前批次对应的元数据、使这批已追加到共享 data queue 的字节正式成为 一个可消费的写操作，并将 size 计数归零。适合周期性手动 flush。 After calling, the metadata that describes the current batch is published so the bytes already appended into the shared data queue become one consumable write operation, and the size counter is reset. Suitable for periodic manual flush.

Returns

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

Definition at line 147 of file write_stream.cpp.

{
  auto ans = ErrorCode::OK;
 
  if (owns_port_ && buffered_size_ > 0)
  {
    ans = SubmitBuffered();
    if (op_.type == WriteOperation::OperationType::BLOCK)
    {
      return ans;
    }
  }
 
  if (owns_port_)
  {
    Release();
  }
 
  return ans;
}

EmptySize()

size_t LibXR::WritePort::Stream::EmptySize ( ) const

inlinenodiscard

获取当前批次还可追加的剩余字节数。

Returns the remaining appendable bytes in the current batch.

返回值只在 Acquire 成功后有意义；若当前尚未持有流批次所有权，则返回 0。 The return value is meaningful only after Acquire succeeds; it returns zero while this stream does not currently own the batch.

Definition at line 159 of file write_port.hpp.

    {
      return owns_port_ ? (batch_capacity_ - buffered_size_) : 0;
    }

operator<<()

WritePort::Stream & WritePort::Stream::operator<<

(

const ConstRawData &

data

)

追加写入数据的语法糖，忽略返回状态并支持链式调用。

Syntax sugar for appending data; ignores the status and supports chaining.

Parameters

data	要写入的数据 Data to write.

Returns

返回自身引用 Enables chainable call.

Definition at line 131 of file write_stream.cpp.

{
  if (Acquire() != ErrorCode::OK)
  {
    return *this;
  }
 
  if (EmptySize() < data.size_)
  {
    return *this;
  }
 
  UNUSED(Write(data));
  return *this;
}

Release()

void WritePort::Stream::Release ( )

private

将当前批次的端口所有权归还给 WritePort。

Releases the current batch ownership back to WritePort.

Definition at line 122 of file write_stream.cpp.

{
  if (owns_port_)
  {
    owns_port_ = false;
    port_->busy_.store(BusyState::IDLE, std::memory_order_release);
  }
}

SubmitBuffered()

ErrorCode WritePort::Stream::SubmitBuffered ( )

nodiscardprivate

将当前已缓存批次提交给 WritePort。

Submits the currently buffered batch to WritePort.

非 BLOCK 路径下，提交后当前 Stream 仍负责释放端口所有权；BLOCK 路径下， 所有权会交给 WritePort 的等待状态机继续管理。 On non-BLOCK paths, this Stream still releases the port ownership after submission; on BLOCK paths, ownership is handed off to WritePort's wait-state machine.

Definition at line 92 of file write_stream.cpp.

{
  ASSERT(owns_port_);
  ASSERT(buffered_size_ > 0);
 
  if (op_.type == WriteOperation::OperationType::BLOCK)
  {
    // Publish the wait state before the queued metadata can be consumed.
    // 元数据可能被消费前，先发布等待状态。
    op_.MarkAsRunning();
    port_->busy_.store(BusyState::BLOCK_PUBLISHING, std::memory_order_release);
  }
 
  auto ans = port_->queue_info_->Push(
      WriteInfoBlock{ConstRawData{nullptr, buffered_size_}, op_});
  ASSERT(ans == ErrorCode::OK);
 
  ans = port_->CommitWrite({nullptr, buffered_size_}, op_, true);
  buffered_size_ = 0;
 
  if (op_.type == WriteOperation::OperationType::BLOCK)
  {
    // WritePort now owns the BLOCK wait/finish state machine.
    // BLOCK 等待/完成状态机此后由 WritePort 接管。
    owns_port_ = false;
  }
 
  return ans;
}

Write() [1/2]

ErrorCode WritePort::Stream::Write ( ConstRawData data )

nodiscard

追加一个原始数据片段到当前流批次。

Appends one raw-data chunk to the current stream batch.

该接口是 Stream 的底层语义写入入口。它负责拿到当前批次的写入所有权，并尝试将 整个片段原子地追加到当前批次对应的共享 data queue 尾部；Commit() 负责随后发布 这批字节对应的元数据。若空间不足则返回 FULL，并保持该片段完全未写入。 This is the low-level semantic write entrypoint of Stream. It acquires the current batch ownership and then attempts to append the whole chunk atomically into the shared data-queue tail; Commit() later publishes the metadata that describes this batch. If space is insufficient it returns FULL and leaves that chunk entirely unwritten.

Parameters

data	要写入的数据片段 Raw-data chunk to append.

Returns

返回操作结果。Returns the write result.

Definition at line 65 of file write_stream.cpp.

{
  if (data.size_ == 0)
  {
    return ErrorCode::OK;
  }
 
  if (data.addr_ == nullptr)
  {
    return ErrorCode::PTR_NULL;
  }
 
  auto lock_result = Acquire();
  if (lock_result != ErrorCode::OK)
  {
    return lock_result;
  }
 
  auto ans = port_->queue_data_->PushBatch(reinterpret_cast<const uint8_t*>(data.addr_),
                                           data.size_);
  if (ans == ErrorCode::OK)
  {
    buffered_size_ += data.size_;
  }
  return ans;
}

Write() [2/2]

ErrorCode LibXR::WritePort::Stream::Write ( std::string_view text )

inlinenodiscard

追加一个文本片段到当前流批次。

Appends one text chunk to the current stream batch.

Parameters

text	要写入的文本片段 Text chunk to append.

Returns

返回操作结果。Returns the write result.

Definition at line 115 of file write_port.hpp.

    {
      return Write(ConstRawData{text.data(), text.size()});
    }

Field Documentation

batch_capacity_

size_t LibXR::WritePort::Stream::batch_capacity_ = 0

private

当前批次可用的总容量 Total capacity reserved for the current batch

Definition at line 184 of file write_port.hpp.

buffered_size_

size_t LibXR::WritePort::Stream::buffered_size_ = 0

private

当前批次已追加到共享 data queue、但尚未发布对应元数据的字节数 Bytes already appended into the shared data queue for the current batch, but whose metadata has not yet been published

Definition at line 185 of file write_port.hpp.

op_

LibXR::WriteOperation LibXR::WritePort::Stream::op_

private

写操作对象 Write operation object

Definition at line 183 of file write_port.hpp.

owns_port_

bool LibXR::WritePort::Stream::owns_port_ = false

private

当前 Stream 是否持有该批次的端口所有权 Whether this Stream currently owns the batch

Definition at line 186 of file write_port.hpp.

port_

LibXR::WritePort* LibXR::WritePort::Stream::port_

private

写端口指针 Pointer to the WritePort

Definition at line 182 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_stream.cpp