LibXR::WritePort::Stream Class Reference

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

#include <libxr_rw.hpp>

Collaboration diagram for LibXR::WritePort::Stream:

Public Member Functions
	Stream (LibXR::WritePort *port, LibXR::WriteOperation op)
	构造流写入对象，并尝试锁定端口。
	~Stream ()
	析构时自动提交已累积的数据并释放锁。
Stream &	operator<< (const ConstRawData &data)
	追加写入数据，支持链式调用。
ErrorCode	Commit ()
	手动提交已写入的数据到队列，并释放当前锁。

Private Attributes
LibXR::WritePort *	port_
	写端口指针 Pointer to the WritePort
LibXR::WriteOperation	op_
	写操作对象 Write operation object
size_t	cap_
	当前队列容量 Current queue capacity
size_t	size_ = 0
	当前已写入但未提交的字节数 Bytes written but not yet committed
bool	locked_ = false
	是否持有写锁 Whether write lock is held

Detailed Description

WritePort 的流式写入操作器，支持链式 << 操作和批量提交。

Stream-like writer for WritePort, supporting chainable << operations and batch commit.

构造时会尝试锁定 WritePort，并批量写入，减少碎片化写操作和队列压力。 Automatically acquires the WritePort lock (if possible), enabling batch writes to reduce fragmented write operations and queue pressure.

Definition at line 584 of file libxr_rw.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 498 of file libxr_rw.cpp.

    : port_(port), op_(op)
{
  BusyState expected = BusyState::IDLE;
  if (!port_->busy_.compare_exchange_strong(expected, BusyState::LOCKED,
                                            std::memory_order_acq_rel,
                                            std::memory_order_acquire))
  {
    return;
  }
 
  if (port_->queue_info_->EmptySize() < 1)
  {
    port_->busy_.store(BusyState::IDLE, std::memory_order_release);
    return;
  }
 
  locked_ = true;
  cap_ = port_->queue_data_->EmptySize();
}

~Stream()

WritePort::Stream::~Stream

(

)

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

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

Definition at line 519 of file libxr_rw.cpp.

{
  if (locked_ && size_ > 0)
  {
    auto ans =
        port_->queue_info_->Push(WriteInfoBlock{ConstRawData{nullptr, size_}, op_});
    ASSERT(ans == ErrorCode::OK);
    port_->CommitWrite({nullptr, size_}, op_, true);
    if (op_.type == WriteOperation::OperationType::BLOCK)
    {
      // WritePort now owns the BLOCK wait/finish state machine.
      locked_ = false;
    }
  }
 
  if (locked_)
  {
    port_->busy_.store(BusyState::IDLE, std::memory_order_release);
  }
}

Member Function Documentation

Commit()

ErrorCode WritePort::Stream::Commit

(

)

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

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

调用后已写入数据会立即入队，size 计数归零。适合周期性手动 flush。 After calling, written data is enqueued, size counter reset. Suitable for periodic manual flush.

Returns

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

Definition at line 572 of file libxr_rw.cpp.

{
  auto ans = ErrorCode::OK;
 
  if (locked_ && size_ > 0)
  {
    ans = port_->queue_info_->Push(WriteInfoBlock{ConstRawData{nullptr, size_}, op_});
    ASSERT(ans == ErrorCode::OK);
    ans = port_->CommitWrite({nullptr, size_}, op_, true);
    if (op_.type == WriteOperation::OperationType::BLOCK)
    {
      // WritePort will release busy_ after the BLOCK handoff completes.
      size_ = 0;
      locked_ = false;
      return ans;
    }
 
    ASSERT(ans == ErrorCode::OK);
    size_ = 0;
  }
 
  if (locked_)
  {
    locked_ = false;
    port_->busy_.store(BusyState::IDLE, std::memory_order_release);
  }
 
  return ans;
}

operator<<()

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

(

const ConstRawData &

data

)

追加写入数据，支持链式调用。

Appends data for writing, supporting chain calls.

Parameters

data	要写入的数据 Data to write.

Returns

返回自身引用 Enables chainable call.

Definition at line 540 of file libxr_rw.cpp.

{
  if (!locked_)
  {
    BusyState expected = BusyState::IDLE;
    if (!port_->busy_.compare_exchange_strong(expected, BusyState::LOCKED,
                                              std::memory_order_acq_rel,
                                              std::memory_order_acquire))
    {
      return *this;
    }
 
    if (port_->queue_info_->EmptySize() < 1)
    {
      port_->busy_.store(BusyState::IDLE, std::memory_order_release);
      return *this;
    }
 
    locked_ = true;
    cap_ = port_->queue_data_->EmptySize();
  }
  if (size_ + data.size_ <= cap_)
  {
    auto ans = port_->queue_data_->PushBatch(reinterpret_cast<const uint8_t*>(data.addr_),
                                             data.size_);
    ASSERT(ans == ErrorCode::OK);
    size_ += data.size_;
  }
 
  return *this;
}

Field Documentation

cap_

size_t LibXR::WritePort::Stream::cap_

private

当前队列容量 Current queue capacity

Definition at line 627 of file libxr_rw.hpp.

locked_

bool LibXR::WritePort::Stream::locked_ = false

private

是否持有写锁 Whether write lock is held

Definition at line 629 of file libxr_rw.hpp.

op_

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

private

写操作对象 Write operation object

Definition at line 626 of file libxr_rw.hpp.

port_

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

private

写端口指针 Pointer to the WritePort

Definition at line 625 of file libxr_rw.hpp.

size_

size_t LibXR::WritePort::Stream::size_ = 0

private

当前已写入但未提交的字节数 Bytes written but not yet committed

Definition at line 628 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