This source file includes following definitions.
- current_num_bytes_
- ProducerCloseImplNoLock
- ProducerWriteDataImplNoLock
- ProducerBeginWriteDataImplNoLock
- ProducerEndWriteDataImplNoLock
- ProducerSatisfiedFlagsNoLock
- ProducerSatisfiableFlagsNoLock
- ConsumerCloseImplNoLock
- ConsumerReadDataImplNoLock
- ConsumerDiscardDataImplNoLock
- ConsumerQueryDataImplNoLock
- ConsumerBeginReadDataImplNoLock
- ConsumerEndReadDataImplNoLock
- ConsumerSatisfiedFlagsNoLock
- ConsumerSatisfiableFlagsNoLock
- EnsureBufferNoLock
- DestroyBufferNoLock
- GetMaxNumBytesToWriteNoLock
- GetMaxNumBytesToReadNoLock
- MarkDataAsConsumedNoLock
#include "mojo/system/local_data_pipe.h"
#include <string.h>
#include <algorithm>
#include "base/logging.h"
#include "mojo/system/constants.h"
namespace mojo {
namespace system {
LocalDataPipe::LocalDataPipe(const MojoCreateDataPipeOptions& options)
: DataPipe(true, true, options),
start_index_(0),
current_num_bytes_(0) {
}
LocalDataPipe::~LocalDataPipe() {
}
void LocalDataPipe::ProducerCloseImplNoLock() {
if (!consumer_open_no_lock() || !current_num_bytes_) {
DCHECK(!consumer_in_two_phase_read_no_lock());
DestroyBufferNoLock();
}
}
MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(const void* elements,
uint32_t* num_bytes,
bool all_or_none) {
DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
DCHECK_GT(*num_bytes, 0u);
DCHECK(consumer_open_no_lock());
size_t num_bytes_to_write = 0;
if (may_discard()) {
if (all_or_none && *num_bytes > capacity_num_bytes())
return MOJO_RESULT_OUT_OF_RANGE;
num_bytes_to_write = std::min(static_cast<size_t>(*num_bytes),
capacity_num_bytes());
if (num_bytes_to_write > capacity_num_bytes() - current_num_bytes_) {
MarkDataAsConsumedNoLock(
num_bytes_to_write - (capacity_num_bytes() - current_num_bytes_));
}
} else {
if (all_or_none && *num_bytes > capacity_num_bytes() - current_num_bytes_) {
return MOJO_RESULT_OUT_OF_RANGE;
}
num_bytes_to_write = std::min(static_cast<size_t>(*num_bytes),
capacity_num_bytes() - current_num_bytes_);
}
if (num_bytes_to_write == 0)
return MOJO_RESULT_SHOULD_WAIT;
size_t num_bytes_to_write_first =
std::min(num_bytes_to_write, GetMaxNumBytesToWriteNoLock());
size_t first_write_index =
(start_index_ + current_num_bytes_) % capacity_num_bytes();
EnsureBufferNoLock();
memcpy(buffer_.get() + first_write_index, elements, num_bytes_to_write_first);
if (num_bytes_to_write_first < num_bytes_to_write) {
memcpy(buffer_.get(),
static_cast<const char*>(elements) + num_bytes_to_write_first,
num_bytes_to_write - num_bytes_to_write_first);
}
current_num_bytes_ += num_bytes_to_write;
DCHECK_LE(current_num_bytes_, capacity_num_bytes());
*num_bytes = static_cast<uint32_t>(num_bytes_to_write);
return MOJO_RESULT_OK;
}
MojoResult LocalDataPipe::ProducerBeginWriteDataImplNoLock(
void** buffer,
uint32_t* buffer_num_bytes,
bool all_or_none) {
DCHECK(consumer_open_no_lock());
size_t write_index =
(start_index_ + current_num_bytes_) % capacity_num_bytes();
size_t max_num_bytes_to_write = GetMaxNumBytesToWriteNoLock();
if (all_or_none && *buffer_num_bytes > max_num_bytes_to_write) {
if (may_discard() &&
*buffer_num_bytes <= capacity_num_bytes() - write_index) {
DCHECK_GE(start_index_, write_index);
DCHECK_GT(*buffer_num_bytes - max_num_bytes_to_write, 0u);
MarkDataAsConsumedNoLock(*buffer_num_bytes - max_num_bytes_to_write);
max_num_bytes_to_write = *buffer_num_bytes;
} else {
return MOJO_RESULT_OUT_OF_RANGE;
}
}
if (max_num_bytes_to_write == 0)
return MOJO_RESULT_SHOULD_WAIT;
EnsureBufferNoLock();
*buffer = buffer_.get() + write_index;
*buffer_num_bytes = static_cast<uint32_t>(max_num_bytes_to_write);
set_producer_two_phase_max_num_bytes_written_no_lock(
static_cast<uint32_t>(max_num_bytes_to_write));
return MOJO_RESULT_OK;
}
MojoResult LocalDataPipe::ProducerEndWriteDataImplNoLock(
uint32_t num_bytes_written) {
DCHECK_LE(num_bytes_written,
producer_two_phase_max_num_bytes_written_no_lock());
current_num_bytes_ += num_bytes_written;
DCHECK_LE(current_num_bytes_, capacity_num_bytes());
set_producer_two_phase_max_num_bytes_written_no_lock(0);
return MOJO_RESULT_OK;
}
MojoWaitFlags LocalDataPipe::ProducerSatisfiedFlagsNoLock() {
MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
if (consumer_open_no_lock() &&
(may_discard() || current_num_bytes_ < capacity_num_bytes()) &&
!producer_in_two_phase_write_no_lock())
rv |= MOJO_WAIT_FLAG_WRITABLE;
return rv;
}
MojoWaitFlags LocalDataPipe::ProducerSatisfiableFlagsNoLock() {
MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
if (consumer_open_no_lock())
rv |= MOJO_WAIT_FLAG_WRITABLE;
return rv;
}
void LocalDataPipe::ConsumerCloseImplNoLock() {
if (!producer_open_no_lock() || !producer_in_two_phase_write_no_lock())
DestroyBufferNoLock();
current_num_bytes_ = 0;
}
MojoResult LocalDataPipe::ConsumerReadDataImplNoLock(void* elements,
uint32_t* num_bytes,
bool all_or_none) {
DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
DCHECK_GT(*num_bytes, 0u);
if (all_or_none && *num_bytes > current_num_bytes_) {
return producer_open_no_lock() ? MOJO_RESULT_OUT_OF_RANGE :
MOJO_RESULT_FAILED_PRECONDITION;
}
size_t num_bytes_to_read =
std::min(static_cast<size_t>(*num_bytes), current_num_bytes_);
if (num_bytes_to_read == 0) {
return producer_open_no_lock() ? MOJO_RESULT_SHOULD_WAIT :
MOJO_RESULT_FAILED_PRECONDITION;
}
size_t num_bytes_to_read_first =
std::min(num_bytes_to_read, GetMaxNumBytesToReadNoLock());
memcpy(elements, buffer_.get() + start_index_, num_bytes_to_read_first);
if (num_bytes_to_read_first < num_bytes_to_read) {
memcpy(static_cast<char*>(elements) + num_bytes_to_read_first,
buffer_.get(),
num_bytes_to_read - num_bytes_to_read_first);
}
MarkDataAsConsumedNoLock(num_bytes_to_read);
*num_bytes = static_cast<uint32_t>(num_bytes_to_read);
return MOJO_RESULT_OK;
}
MojoResult LocalDataPipe::ConsumerDiscardDataImplNoLock(uint32_t* num_bytes,
bool all_or_none) {
DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
DCHECK_GT(*num_bytes, 0u);
if (all_or_none && *num_bytes > current_num_bytes_) {
return producer_open_no_lock() ? MOJO_RESULT_OUT_OF_RANGE :
MOJO_RESULT_FAILED_PRECONDITION;
}
if (current_num_bytes_ == 0) {
return producer_open_no_lock() ? MOJO_RESULT_SHOULD_WAIT :
MOJO_RESULT_FAILED_PRECONDITION;
}
size_t num_bytes_to_discard =
std::min(static_cast<size_t>(*num_bytes), current_num_bytes_);
MarkDataAsConsumedNoLock(num_bytes_to_discard);
*num_bytes = static_cast<uint32_t>(num_bytes_to_discard);
return MOJO_RESULT_OK;
}
MojoResult LocalDataPipe::ConsumerQueryDataImplNoLock(uint32_t* num_bytes) {
*num_bytes = static_cast<uint32_t>(current_num_bytes_);
return MOJO_RESULT_OK;
}
MojoResult LocalDataPipe::ConsumerBeginReadDataImplNoLock(
const void** buffer,
uint32_t* buffer_num_bytes,
bool all_or_none) {
size_t max_num_bytes_to_read = GetMaxNumBytesToReadNoLock();
if (all_or_none && *buffer_num_bytes > max_num_bytes_to_read) {
return producer_open_no_lock() ? MOJO_RESULT_OUT_OF_RANGE :
MOJO_RESULT_FAILED_PRECONDITION;
}
if (max_num_bytes_to_read == 0) {
return producer_open_no_lock() ? MOJO_RESULT_SHOULD_WAIT :
MOJO_RESULT_FAILED_PRECONDITION;
}
*buffer = buffer_.get() + start_index_;
*buffer_num_bytes = static_cast<uint32_t>(max_num_bytes_to_read);
set_consumer_two_phase_max_num_bytes_read_no_lock(
static_cast<uint32_t>(max_num_bytes_to_read));
return MOJO_RESULT_OK;
}
MojoResult LocalDataPipe::ConsumerEndReadDataImplNoLock(
uint32_t num_bytes_read) {
DCHECK_LE(num_bytes_read, consumer_two_phase_max_num_bytes_read_no_lock());
DCHECK_LE(start_index_ + num_bytes_read, capacity_num_bytes());
MarkDataAsConsumedNoLock(num_bytes_read);
set_consumer_two_phase_max_num_bytes_read_no_lock(0);
return MOJO_RESULT_OK;
}
MojoWaitFlags LocalDataPipe::ConsumerSatisfiedFlagsNoLock() {
MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
if (current_num_bytes_ > 0 && !consumer_in_two_phase_read_no_lock())
rv |= MOJO_WAIT_FLAG_READABLE;
return rv;
}
MojoWaitFlags LocalDataPipe::ConsumerSatisfiableFlagsNoLock() {
MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
if (current_num_bytes_ > 0 || producer_open_no_lock())
rv |= MOJO_WAIT_FLAG_READABLE;
return rv;
}
void LocalDataPipe::EnsureBufferNoLock() {
DCHECK(producer_open_no_lock());
if (buffer_.get())
return;
buffer_.reset(static_cast<char*>(
base::AlignedAlloc(capacity_num_bytes(), kDataPipeBufferAlignmentBytes)));
}
void LocalDataPipe::DestroyBufferNoLock() {
#ifndef NDEBUG
if (buffer_.get())
memset(buffer_.get(), 0xcd, capacity_num_bytes());
#endif
buffer_.reset();
}
size_t LocalDataPipe::GetMaxNumBytesToWriteNoLock() {
size_t next_index = start_index_ + current_num_bytes_;
if (next_index >= capacity_num_bytes()) {
next_index %= capacity_num_bytes();
DCHECK_GE(start_index_, next_index);
DCHECK_EQ(start_index_ - next_index,
capacity_num_bytes() - current_num_bytes_);
return start_index_ - next_index;
}
return capacity_num_bytes() - next_index;
}
size_t LocalDataPipe::GetMaxNumBytesToReadNoLock() {
if (start_index_ + current_num_bytes_ > capacity_num_bytes())
return capacity_num_bytes() - start_index_;
return current_num_bytes_;
}
void LocalDataPipe::MarkDataAsConsumedNoLock(size_t num_bytes) {
DCHECK_LE(num_bytes, current_num_bytes_);
start_index_ += num_bytes;
start_index_ %= capacity_num_bytes();
current_num_bytes_ -= num_bytes;
}
}
}