// Copyright 2013 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#ifdef UNSAFE_BUFFERS_BUILD
// TODO(crbug.com/351564777): Remove this and convert code to safer constructs.
#pragma allow_unsafe_buffers
#endif
#include "mojo/core/data_pipe_producer_dispatcher.h"
#include <stddef.h>
#include <stdint.h>
#include <utility>
#include "base/functional/bind.h"
#include "base/logging.h"
#include "base/memory/ref_counted.h"
#include "base/numerics/checked_math.h"
#include "base/trace_event/trace_event.h"
#include "mojo/core/configuration.h"
#include "mojo/core/core.h"
#include "mojo/core/data_pipe_control_message.h"
#include "mojo/core/node_controller.h"
#include "mojo/core/platform_handle_utils.h"
#include "mojo/core/request_context.h"
#include "mojo/core/user_message_impl.h"
#include "mojo/public/c/system/data_pipe.h"
namespace mojo {
namespace core {
namespace {
const uint8_t kFlagPeerClosed = 0x01;
#pragma pack(push, 1)
struct SerializedState {
MojoCreateDataPipeOptions options;
uint64_t pipe_id;
uint32_t write_offset;
uint32_t available_capacity;
uint8_t flags;
uint64_t buffer_guid_high;
uint64_t buffer_guid_low;
char padding[7];
};
static_assert(sizeof(SerializedState) % 8 == 0,
"Invalid SerializedState size.");
#pragma pack(pop)
} // namespace
// A PortObserver which forwards to a DataPipeProducerDispatcher. This owns a
// reference to the dispatcher to ensure it lives as long as the observed port.
class DataPipeProducerDispatcher::PortObserverThunk
: public NodeController::PortObserver {
public:
explicit PortObserverThunk(
scoped_refptr<DataPipeProducerDispatcher> dispatcher)
: dispatcher_(dispatcher) {}
PortObserverThunk(const PortObserverThunk&) = delete;
PortObserverThunk& operator=(const PortObserverThunk&) = delete;
private:
~PortObserverThunk() override = default;
// NodeController::PortObserver:
void OnPortStatusChanged() override { dispatcher_->OnPortStatusChanged(); }
scoped_refptr<DataPipeProducerDispatcher> dispatcher_;
};
// static
scoped_refptr<DataPipeProducerDispatcher> DataPipeProducerDispatcher::Create(
NodeController* node_controller,
const ports::PortRef& control_port,
base::UnsafeSharedMemoryRegion shared_ring_buffer,
const MojoCreateDataPipeOptions& options,
uint64_t pipe_id) {
scoped_refptr<DataPipeProducerDispatcher> producer =
new DataPipeProducerDispatcher(node_controller, control_port,
std::move(shared_ring_buffer), options,
pipe_id);
base::AutoLock lock(producer->lock_);
if (!producer->InitializeNoLock())
return nullptr;
return producer;
}
Dispatcher::Type DataPipeProducerDispatcher::GetType() const {
return Type::DATA_PIPE_PRODUCER;
}
MojoResult DataPipeProducerDispatcher::Close() {
base::AutoLock lock(lock_);
DVLOG(1) << "Closing data pipe producer " << pipe_id_;
return CloseNoLock();
}
MojoResult DataPipeProducerDispatcher::WriteData(
const void* elements,
uint32_t* num_bytes,
const MojoWriteDataOptions& options) {
base::AutoLock lock(lock_);
if (!shared_ring_buffer_.IsValid() || in_transit_)
return MOJO_RESULT_INVALID_ARGUMENT;
if (in_two_phase_write_)
return MOJO_RESULT_BUSY;
if (peer_closed_)
return MOJO_RESULT_FAILED_PRECONDITION;
if (*num_bytes % options_.element_num_bytes != 0)
return MOJO_RESULT_INVALID_ARGUMENT;
if (*num_bytes == 0)
return MOJO_RESULT_OK; // Nothing to do.
if ((options.flags & MOJO_WRITE_DATA_FLAG_ALL_OR_NONE) &&
(*num_bytes > available_capacity_)) {
// Don't return "should wait" since you can't wait for a specified amount of
// data.
return MOJO_RESULT_OUT_OF_RANGE;
}
DCHECK_LE(available_capacity_, options_.capacity_num_bytes);
uint32_t num_bytes_to_write = std::min(*num_bytes, available_capacity_);
if (num_bytes_to_write == 0)
return MOJO_RESULT_SHOULD_WAIT;
*num_bytes = num_bytes_to_write;
CHECK(ring_buffer_mapping_.IsValid());
uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_.memory());
CHECK(data);
const uint8_t* source = static_cast<const uint8_t*>(elements);
CHECK(source);
DCHECK_LE(write_offset_, options_.capacity_num_bytes);
uint32_t tail_bytes_to_write =
std::min(options_.capacity_num_bytes - write_offset_, num_bytes_to_write);
uint32_t head_bytes_to_write = num_bytes_to_write - tail_bytes_to_write;
DCHECK_GT(tail_bytes_to_write, 0u);
memcpy(data + write_offset_, source, tail_bytes_to_write);
if (head_bytes_to_write > 0)
memcpy(data, source + tail_bytes_to_write, head_bytes_to_write);
DCHECK_LE(num_bytes_to_write, available_capacity_);
available_capacity_ -= num_bytes_to_write;
write_offset_ =
(write_offset_ + num_bytes_to_write) % options_.capacity_num_bytes;
watchers_.NotifyState(GetHandleSignalsStateNoLock());
base::AutoUnlock unlock(lock_);
NotifyWrite(num_bytes_to_write);
return MOJO_RESULT_OK;
}
MojoResult DataPipeProducerDispatcher::BeginWriteData(
void** buffer,
uint32_t* buffer_num_bytes,
MojoBeginWriteDataFlags flags) {
base::AutoLock lock(lock_);
if (!shared_ring_buffer_.IsValid() || in_transit_)
return MOJO_RESULT_INVALID_ARGUMENT;
if (in_two_phase_write_)
return MOJO_RESULT_BUSY;
if (peer_closed_)
return MOJO_RESULT_FAILED_PRECONDITION;
if (available_capacity_ == 0) {
return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
: MOJO_RESULT_SHOULD_WAIT;
}
in_two_phase_write_ = true;
*buffer_num_bytes = std::min(options_.capacity_num_bytes - write_offset_,
available_capacity_);
DCHECK_GT(*buffer_num_bytes, 0u);
CHECK(ring_buffer_mapping_.IsValid());
uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_.memory());
*buffer = data + write_offset_;
return MOJO_RESULT_OK;
}
MojoResult DataPipeProducerDispatcher::EndWriteData(
uint32_t num_bytes_written) {
base::AutoLock lock(lock_);
if (is_closed_ || in_transit_)
return MOJO_RESULT_INVALID_ARGUMENT;
if (!in_two_phase_write_)
return MOJO_RESULT_FAILED_PRECONDITION;
// Note: Allow successful completion of the two-phase write even if the other
// side has been closed.
MojoResult rv = MOJO_RESULT_OK;
if (num_bytes_written > available_capacity_ ||
num_bytes_written % options_.element_num_bytes != 0 ||
write_offset_ + num_bytes_written > options_.capacity_num_bytes) {
rv = MOJO_RESULT_INVALID_ARGUMENT;
} else {
DCHECK_LE(num_bytes_written + write_offset_, options_.capacity_num_bytes);
available_capacity_ -= num_bytes_written;
write_offset_ =
(write_offset_ + num_bytes_written) % options_.capacity_num_bytes;
base::AutoUnlock unlock(lock_);
NotifyWrite(num_bytes_written);
}
in_two_phase_write_ = false;
// If we're now writable, we *became* writable (since we weren't writable
// during the two-phase write), so notify watchers.
watchers_.NotifyState(GetHandleSignalsStateNoLock());
return rv;
}
HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsState() const {
base::AutoLock lock(lock_);
return GetHandleSignalsStateNoLock();
}
MojoResult DataPipeProducerDispatcher::AddWatcherRef(
const scoped_refptr<WatcherDispatcher>& watcher,
uintptr_t context) {
base::AutoLock lock(lock_);
if (is_closed_ || in_transit_)
return MOJO_RESULT_INVALID_ARGUMENT;
return watchers_.Add(watcher, context, GetHandleSignalsStateNoLock());
}
MojoResult DataPipeProducerDispatcher::RemoveWatcherRef(
WatcherDispatcher* watcher,
uintptr_t context) {
base::AutoLock lock(lock_);
if (is_closed_ || in_transit_)
return MOJO_RESULT_INVALID_ARGUMENT;
return watchers_.Remove(watcher, context);
}
void DataPipeProducerDispatcher::StartSerialize(uint32_t* num_bytes,
uint32_t* num_ports,
uint32_t* num_handles) {
base::AutoLock lock(lock_);
DCHECK(in_transit_);
*num_bytes = sizeof(SerializedState);
*num_ports = 1;
*num_handles = 1;
}
bool DataPipeProducerDispatcher::EndSerialize(
void* destination,
ports::PortName* ports,
PlatformHandle* platform_handles) {
SerializedState* state = static_cast<SerializedState*>(destination);
memcpy(&state->options, &options_, sizeof(MojoCreateDataPipeOptions));
memset(state->padding, 0, sizeof(state->padding));
base::AutoLock lock(lock_);
DCHECK(in_transit_);
state->pipe_id = pipe_id_;
state->write_offset = write_offset_;
state->available_capacity = available_capacity_;
state->flags = peer_closed_ ? kFlagPeerClosed : 0;
auto region_handle =
base::UnsafeSharedMemoryRegion::TakeHandleForSerialization(
std::move(shared_ring_buffer_));
const base::UnguessableToken& guid = region_handle.GetGUID();
state->buffer_guid_high = guid.GetHighForSerialization();
state->buffer_guid_low = guid.GetLowForSerialization();
ports[0] = control_port_.name();
PlatformHandle handle;
PlatformHandle ignored_handle;
ExtractPlatformHandlesFromSharedMemoryRegionHandle(
region_handle.PassPlatformHandle(), &handle, &ignored_handle);
if (!handle.is_valid() || ignored_handle.is_valid())
return false;
platform_handles[0] = std::move(handle);
return true;
}
bool DataPipeProducerDispatcher::BeginTransit() {
base::AutoLock lock(lock_);
if (in_transit_)
return false;
in_transit_ = !in_two_phase_write_;
return in_transit_;
}
void DataPipeProducerDispatcher::CompleteTransitAndClose() {
node_controller_->SetPortObserver(control_port_, nullptr);
base::AutoLock lock(lock_);
DCHECK(in_transit_);
transferred_ = true;
in_transit_ = false;
CloseNoLock();
}
void DataPipeProducerDispatcher::CancelTransit() {
base::AutoLock lock(lock_);
DCHECK(in_transit_);
in_transit_ = false;
HandleSignalsState state = GetHandleSignalsStateNoLock();
watchers_.NotifyState(state);
}
// static
scoped_refptr<DataPipeProducerDispatcher>
DataPipeProducerDispatcher::Deserialize(const void* data,
size_t num_bytes,
const ports::PortName* ports,
size_t num_ports,
PlatformHandle* handles,
size_t num_handles) {
if (num_ports != 1 || num_handles != 1 ||
num_bytes != sizeof(SerializedState)) {
AssertNotExtractingHandlesFromMessage();
return nullptr;
}
const SerializedState* state = static_cast<const SerializedState*>(data);
if (!state->options.capacity_num_bytes || !state->options.element_num_bytes ||
state->options.capacity_num_bytes < state->options.element_num_bytes ||
state->write_offset >= state->options.capacity_num_bytes ||
state->available_capacity > state->options.capacity_num_bytes) {
AssertNotExtractingHandlesFromMessage();
return nullptr;
}
NodeController* node_controller = Core::Get()->GetNodeController();
ports::PortRef port;
if (node_controller->node()->GetPort(ports[0], &port) != ports::OK) {
AssertNotExtractingHandlesFromMessage();
return nullptr;
}
std::optional<base::UnguessableToken> buffer_guid =
base::UnguessableToken::Deserialize(state->buffer_guid_high,
state->buffer_guid_low);
if (!buffer_guid.has_value()) {
AssertNotExtractingHandlesFromMessage();
return nullptr;
}
auto region_handle = CreateSharedMemoryRegionHandleFromPlatformHandles(
std::move(handles[0]), PlatformHandle());
auto region = base::subtle::PlatformSharedMemoryRegion::Take(
std::move(region_handle),
base::subtle::PlatformSharedMemoryRegion::Mode::kUnsafe,
state->options.capacity_num_bytes, buffer_guid.value());
auto ring_buffer =
base::UnsafeSharedMemoryRegion::Deserialize(std::move(region));
if (!ring_buffer.IsValid()) {
DLOG(ERROR) << "Failed to deserialize shared buffer handle.";
AssertNotExtractingHandlesFromMessage();
return nullptr;
}
scoped_refptr<DataPipeProducerDispatcher> dispatcher =
new DataPipeProducerDispatcher(node_controller, port,
std::move(ring_buffer), state->options,
state->pipe_id);
{
base::AutoLock lock(dispatcher->lock_);
dispatcher->write_offset_ = state->write_offset;
dispatcher->available_capacity_ = state->available_capacity;
dispatcher->peer_closed_ = state->flags & kFlagPeerClosed;
if (!dispatcher->InitializeNoLock()) {
AssertNotExtractingHandlesFromMessage();
return nullptr;
}
if (state->options.capacity_num_bytes >
dispatcher->ring_buffer_mapping_.mapped_size()) {
AssertNotExtractingHandlesFromMessage();
return nullptr;
}
dispatcher->UpdateSignalsStateNoLock();
}
return dispatcher;
}
DataPipeProducerDispatcher::DataPipeProducerDispatcher(
NodeController* node_controller,
const ports::PortRef& control_port,
base::UnsafeSharedMemoryRegion shared_ring_buffer,
const MojoCreateDataPipeOptions& options,
uint64_t pipe_id)
: options_(options),
node_controller_(node_controller),
control_port_(control_port),
pipe_id_(pipe_id),
watchers_(this),
shared_ring_buffer_(std::move(shared_ring_buffer)),
available_capacity_(options_.capacity_num_bytes) {}
DataPipeProducerDispatcher::~DataPipeProducerDispatcher() {
DCHECK(is_closed_ && !in_transit_ && !shared_ring_buffer_.IsValid() &&
!ring_buffer_mapping_.IsValid());
}
bool DataPipeProducerDispatcher::InitializeNoLock() {
lock_.AssertAcquired();
if (!shared_ring_buffer_.IsValid())
return false;
DCHECK(!ring_buffer_mapping_.IsValid());
ring_buffer_mapping_ = shared_ring_buffer_.Map();
if (!ring_buffer_mapping_.IsValid()) {
DLOG(ERROR) << "Failed to map shared buffer.";
shared_ring_buffer_ = base::UnsafeSharedMemoryRegion();
return false;
}
base::AutoUnlock unlock(lock_);
node_controller_->SetPortObserver(
control_port_, base::MakeRefCounted<PortObserverThunk>(this));
return true;
}
MojoResult DataPipeProducerDispatcher::CloseNoLock() {
lock_.AssertAcquired();
if (is_closed_ || in_transit_)
return MOJO_RESULT_INVALID_ARGUMENT;
is_closed_ = true;
ring_buffer_mapping_ = base::WritableSharedMemoryMapping();
shared_ring_buffer_ = base::UnsafeSharedMemoryRegion();
watchers_.NotifyClosed();
if (!transferred_) {
base::AutoUnlock unlock(lock_);
node_controller_->ClosePort(control_port_);
}
return MOJO_RESULT_OK;
}
HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsStateNoLock()
const {
lock_.AssertAcquired();
HandleSignalsState rv;
if (!peer_closed_) {
if (!in_two_phase_write_ && shared_ring_buffer_.IsValid() &&
available_capacity_ > 0)
rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
if (peer_remote_)
rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_REMOTE;
rv.satisfiable_signals |=
MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_REMOTE;
} else {
rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
}
rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
return rv;
}
void DataPipeProducerDispatcher::NotifyWrite(uint32_t num_bytes) {
DVLOG(1) << "Data pipe producer " << pipe_id_
<< " notifying peer: " << num_bytes
<< " bytes written. [control_port=" << control_port_.name() << "]";
SendDataPipeControlMessage(node_controller_, control_port_,
DataPipeCommand::DATA_WAS_WRITTEN, num_bytes);
}
void DataPipeProducerDispatcher::OnPortStatusChanged() {
DCHECK(RequestContext::current());
base::AutoLock lock(lock_);
// We stop observing the control port as soon it's transferred, but this can
// race with events which are raised right before that happens. This is fine
// to ignore.
if (transferred_)
return;
DVLOG(1) << "Control port status changed for data pipe producer " << pipe_id_;
UpdateSignalsStateNoLock();
}
void DataPipeProducerDispatcher::UpdateSignalsStateNoLock() {
lock_.AssertAcquired();
const bool was_peer_closed = peer_closed_;
const bool was_peer_remote = peer_remote_;
size_t previous_capacity = available_capacity_;
ports::PortStatus port_status;
int rv = node_controller_->node()->GetStatus(control_port_, &port_status);
peer_remote_ = rv == ports::OK && port_status.peer_remote;
if (rv != ports::OK || !port_status.receiving_messages) {
DVLOG(1) << "Data pipe producer " << pipe_id_ << " is aware of peer closure"
<< " [control_port=" << control_port_.name() << "]";
peer_closed_ = true;
} else if (rv == ports::OK && port_status.has_messages && !in_transit_) {
std::unique_ptr<ports::UserMessageEvent> message_event;
do {
rv = node_controller_->node()->GetMessage(control_port_, &message_event,
nullptr);
if (rv != ports::OK)
peer_closed_ = true;
if (message_event) {
auto* message = message_event->GetMessage<UserMessageImpl>();
if (message->user_payload_size() < sizeof(DataPipeControlMessage)) {
peer_closed_ = true;
break;
}
const DataPipeControlMessage* m =
static_cast<const DataPipeControlMessage*>(message->user_payload());
if (m->command != DataPipeCommand::DATA_WAS_READ) {
DLOG(ERROR) << "Unexpected message from consumer.";
peer_closed_ = true;
break;
}
TRACE_EVENT0("ipc",
"DataPipeProducerDispatcher received DATA_WAS_READ");
uint32_t new_available_capacity;
if (!base::CheckAdd(available_capacity_, m->num_bytes)
.AssignIfValid(&new_available_capacity) ||
new_available_capacity > options_.capacity_num_bytes) {
DLOG(ERROR) << "Consumer claims to have read too many bytes.";
break;
}
DVLOG(1) << "Data pipe producer " << pipe_id_ << " is aware that "
<< m->num_bytes
<< " bytes were read. [control_port=" << control_port_.name()
<< "]";
available_capacity_ = new_available_capacity;
}
} while (message_event);
}
if (peer_closed_ != was_peer_closed ||
available_capacity_ != previous_capacity ||
was_peer_remote != peer_remote_) {
watchers_.NotifyState(GetHandleSignalsStateNoLock());
}
}
} // namespace core
} // namespace mojo
Codebase Browser
chromium