123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498 |
- // Copyright 2015 The Chromium Authors. All rights reserved.
- // Use of this source code is governed by a BSD-style license that can be
- // found in the LICENSE file.
- #include "mojo/core/message_pipe_dispatcher.h"
- #include <limits>
- #include <memory>
- #include "base/logging.h"
- #include "base/memory/ref_counted.h"
- #include "base/trace_event/trace_event.h"
- #include "mojo/core/core.h"
- #include "mojo/core/node_controller.h"
- #include "mojo/core/ports/event.h"
- #include "mojo/core/ports/message_filter.h"
- #include "mojo/core/request_context.h"
- #include "mojo/core/user_message_impl.h"
- #include "mojo/public/cpp/bindings/mojo_buildflags.h"
- namespace mojo {
- namespace core {
- namespace {
- #pragma pack(push, 1)
- struct SerializedState {
- uint64_t pipe_id;
- int8_t endpoint;
- char padding[7];
- };
- static_assert(sizeof(SerializedState) % 8 == 0,
- "Invalid SerializedState size.");
- #pragma pack(pop)
- } // namespace
- // A PortObserver which forwards to a MessagePipeDispatcher. This owns a
- // reference to the MPD to ensure it lives as long as the observed port.
- class MessagePipeDispatcher::PortObserverThunk
- : public NodeController::PortObserver {
- public:
- explicit PortObserverThunk(scoped_refptr<MessagePipeDispatcher> 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<MessagePipeDispatcher> dispatcher_;
- };
- #if DCHECK_IS_ON()
- // A MessageFilter which never matches a message. Used to peek at the size of
- // the next available message on a port, for debug logging only.
- class PeekSizeMessageFilter : public ports::MessageFilter {
- public:
- PeekSizeMessageFilter() = default;
- PeekSizeMessageFilter(const PeekSizeMessageFilter&) = delete;
- PeekSizeMessageFilter& operator=(const PeekSizeMessageFilter&) = delete;
- ~PeekSizeMessageFilter() override = default;
- // ports::MessageFilter:
- bool Match(const ports::UserMessageEvent& message_event) override {
- const auto* message = message_event.GetMessage<UserMessageImpl>();
- if (message->IsSerialized())
- message_size_ = message->user_payload_size();
- return false;
- }
- size_t message_size() const { return message_size_; }
- private:
- size_t message_size_ = 0;
- };
- #endif // DCHECK_IS_ON()
- MessagePipeDispatcher::MessagePipeDispatcher(NodeController* node_controller,
- const ports::PortRef& port,
- uint64_t pipe_id,
- int endpoint)
- : node_controller_(node_controller),
- port_(port),
- pipe_id_(pipe_id),
- endpoint_(endpoint),
- watchers_(this) {
- DVLOG(2) << "Creating new MessagePipeDispatcher for port " << port.name()
- << " [pipe_id=" << pipe_id << "; endpoint=" << endpoint << "]";
- node_controller_->SetPortObserver(
- port_, base::MakeRefCounted<PortObserverThunk>(this));
- }
- bool MessagePipeDispatcher::Fuse(MessagePipeDispatcher* other) {
- node_controller_->SetPortObserver(port_, nullptr);
- node_controller_->SetPortObserver(other->port_, nullptr);
- ports::PortRef port0;
- {
- base::AutoLock lock(signal_lock_);
- port0 = port_;
- port_closed_.Set(true);
- watchers_.NotifyClosed();
- }
- ports::PortRef port1;
- {
- base::AutoLock lock(other->signal_lock_);
- port1 = other->port_;
- other->port_closed_.Set(true);
- other->watchers_.NotifyClosed();
- }
- // Both ports are always closed by this call.
- int rv = node_controller_->MergeLocalPorts(port0, port1);
- return rv == ports::OK;
- }
- Dispatcher::Type MessagePipeDispatcher::GetType() const {
- return Type::MESSAGE_PIPE;
- }
- MojoResult MessagePipeDispatcher::Close() {
- base::AutoLock lock(signal_lock_);
- DVLOG(2) << "Closing message pipe " << pipe_id_ << " endpoint " << endpoint_
- << " [port=" << port_.name() << "]";
- return CloseNoLock();
- }
- MojoResult MessagePipeDispatcher::WriteMessage(
- std::unique_ptr<ports::UserMessageEvent> message) {
- if (port_closed_ || in_transit_)
- return MOJO_RESULT_INVALID_ARGUMENT;
- int rv = node_controller_->SendUserMessage(port_, std::move(message));
- DVLOG(4) << "Sent message on pipe " << pipe_id_ << " endpoint " << endpoint_
- << " [port=" << port_.name() << "; rv=" << rv << "]";
- if (rv != ports::OK) {
- if (rv == ports::ERROR_PORT_UNKNOWN ||
- rv == ports::ERROR_PORT_STATE_UNEXPECTED ||
- rv == ports::ERROR_PORT_CANNOT_SEND_PEER) {
- return MOJO_RESULT_INVALID_ARGUMENT;
- } else if (rv == ports::ERROR_PORT_PEER_CLOSED) {
- return MOJO_RESULT_FAILED_PRECONDITION;
- }
- NOTREACHED();
- return MOJO_RESULT_UNKNOWN;
- }
- // We may need to update anyone watching our signals in case we just exceeded
- // the unread message count quota.
- base::AutoLock lock(signal_lock_);
- watchers_.NotifyState(GetHandleSignalsStateNoLock());
- return MOJO_RESULT_OK;
- }
- MojoResult MessagePipeDispatcher::ReadMessage(
- std::unique_ptr<ports::UserMessageEvent>* message) {
- // We can't read from a port that's closed or in transit!
- if (port_closed_ || in_transit_)
- return MOJO_RESULT_INVALID_ARGUMENT;
- int rv = node_controller_->node()->GetMessage(port_, message, nullptr);
- if (rv != ports::OK && rv != ports::ERROR_PORT_PEER_CLOSED) {
- if (rv == ports::ERROR_PORT_UNKNOWN ||
- rv == ports::ERROR_PORT_STATE_UNEXPECTED)
- return MOJO_RESULT_INVALID_ARGUMENT;
- NOTREACHED();
- return MOJO_RESULT_UNKNOWN;
- }
- if (!*message) {
- // No message was available in queue.
- if (rv == ports::OK)
- return MOJO_RESULT_SHOULD_WAIT;
- // Peer is closed and there are no more messages to read.
- DCHECK_EQ(rv, ports::ERROR_PORT_PEER_CLOSED);
- return MOJO_RESULT_FAILED_PRECONDITION;
- }
- // We may need to update anyone watching our signals in case we just read the
- // last available message.
- base::AutoLock lock(signal_lock_);
- watchers_.NotifyState(GetHandleSignalsStateNoLock());
- return MOJO_RESULT_OK;
- }
- MojoResult MessagePipeDispatcher::SetQuota(MojoQuotaType type, uint64_t limit) {
- absl::optional<uint64_t> new_ack_request_interval;
- {
- base::AutoLock lock(signal_lock_);
- switch (type) {
- case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_LENGTH:
- if (limit == MOJO_QUOTA_LIMIT_NONE)
- receive_queue_length_limit_.reset();
- else
- receive_queue_length_limit_ = limit;
- break;
- case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_MEMORY_SIZE:
- if (limit == MOJO_QUOTA_LIMIT_NONE)
- receive_queue_memory_size_limit_.reset();
- else
- receive_queue_memory_size_limit_ = limit;
- break;
- case MOJO_QUOTA_TYPE_UNREAD_MESSAGE_COUNT:
- if (limit == MOJO_QUOTA_LIMIT_NONE) {
- unread_message_count_limit_.reset();
- new_ack_request_interval = 0;
- } else {
- unread_message_count_limit_ = limit;
- // Setting the acknowledge request interval for the port to half the
- // unread quota limit, means the ack roundtrip has half the window to
- // catch up with sent messages. In other words, if the producer is
- // producing messages at a steady rate of limit/2 packets per message
- // round trip or lower, the quota limit won't be exceeded. This is
- // assuming the consumer is consuming messages at the same rate.
- new_ack_request_interval = (limit + 1) / 2;
- }
- break;
- default:
- return MOJO_RESULT_INVALID_ARGUMENT;
- }
- }
- if (new_ack_request_interval.has_value()) {
- // NOTE: It is not safe to call into SetAcknowledgeRequestInterval while
- // holding a `signal_lock_`, as it may re-enter this object when the peer is
- // in the same process.
- node_controller_->node()->SetAcknowledgeRequestInterval(
- port_, *new_ack_request_interval);
- }
- return MOJO_RESULT_OK;
- }
- MojoResult MessagePipeDispatcher::QueryQuota(MojoQuotaType type,
- uint64_t* limit,
- uint64_t* usage) {
- base::AutoLock lock(signal_lock_);
- ports::PortStatus port_status;
- if (node_controller_->node()->GetStatus(port_, &port_status) != ports::OK) {
- CHECK(in_transit_ || port_transferred_ || port_closed_);
- return MOJO_RESULT_INVALID_ARGUMENT;
- }
- switch (type) {
- case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_LENGTH:
- *limit = receive_queue_length_limit_.value_or(MOJO_QUOTA_LIMIT_NONE);
- *usage = port_status.queued_message_count;
- break;
- case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_MEMORY_SIZE:
- *limit = receive_queue_memory_size_limit_.value_or(MOJO_QUOTA_LIMIT_NONE);
- *usage = port_status.queued_num_bytes;
- break;
- case MOJO_QUOTA_TYPE_UNREAD_MESSAGE_COUNT:
- *limit = unread_message_count_limit_.value_or(MOJO_QUOTA_LIMIT_NONE);
- *usage = port_status.unacknowledged_message_count;
- break;
- default:
- return MOJO_RESULT_INVALID_ARGUMENT;
- }
- return MOJO_RESULT_OK;
- }
- HandleSignalsState MessagePipeDispatcher::GetHandleSignalsState() const {
- base::AutoLock lock(signal_lock_);
- return GetHandleSignalsStateNoLock();
- }
- MojoResult MessagePipeDispatcher::AddWatcherRef(
- const scoped_refptr<WatcherDispatcher>& watcher,
- uintptr_t context) {
- base::AutoLock lock(signal_lock_);
- if (port_closed_ || in_transit_)
- return MOJO_RESULT_INVALID_ARGUMENT;
- return watchers_.Add(watcher, context, GetHandleSignalsStateNoLock());
- }
- MojoResult MessagePipeDispatcher::RemoveWatcherRef(WatcherDispatcher* watcher,
- uintptr_t context) {
- base::AutoLock lock(signal_lock_);
- if (port_closed_ || in_transit_)
- return MOJO_RESULT_INVALID_ARGUMENT;
- return watchers_.Remove(watcher, context);
- }
- void MessagePipeDispatcher::StartSerialize(uint32_t* num_bytes,
- uint32_t* num_ports,
- uint32_t* num_handles) {
- *num_bytes = static_cast<uint32_t>(sizeof(SerializedState));
- *num_ports = 1;
- *num_handles = 0;
- }
- bool MessagePipeDispatcher::EndSerialize(void* destination,
- ports::PortName* ports,
- PlatformHandle* handles) {
- SerializedState* state = static_cast<SerializedState*>(destination);
- state->pipe_id = pipe_id_;
- state->endpoint = static_cast<int8_t>(endpoint_);
- memset(state->padding, 0, sizeof(state->padding));
- ports[0] = port_.name();
- return true;
- }
- bool MessagePipeDispatcher::BeginTransit() {
- base::AutoLock lock(signal_lock_);
- if (in_transit_ || port_closed_)
- return false;
- in_transit_.Set(true);
- return in_transit_;
- }
- void MessagePipeDispatcher::CompleteTransitAndClose() {
- node_controller_->SetPortObserver(port_, nullptr);
- base::AutoLock lock(signal_lock_);
- port_transferred_ = true;
- in_transit_.Set(false);
- CloseNoLock();
- }
- void MessagePipeDispatcher::CancelTransit() {
- base::AutoLock lock(signal_lock_);
- in_transit_.Set(false);
- // Something may have happened while we were waiting for potential transit.
- watchers_.NotifyState(GetHandleSignalsStateNoLock());
- }
- // static
- scoped_refptr<Dispatcher> MessagePipeDispatcher::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 || num_bytes != sizeof(SerializedState)) {
- AssertNotExtractingHandlesFromMessage();
- return nullptr;
- }
- const SerializedState* state = static_cast<const SerializedState*>(data);
- ports::Node* node = Core::Get()->GetNodeController()->node();
- ports::PortRef port;
- if (node->GetPort(ports[0], &port) != ports::OK) {
- AssertNotExtractingHandlesFromMessage();
- return nullptr;
- }
- ports::PortStatus status;
- if (node->GetStatus(port, &status) != ports::OK) {
- AssertNotExtractingHandlesFromMessage();
- return nullptr;
- }
- return new MessagePipeDispatcher(Core::Get()->GetNodeController(), port,
- state->pipe_id, state->endpoint);
- }
- MessagePipeDispatcher::~MessagePipeDispatcher() = default;
- MojoResult MessagePipeDispatcher::CloseNoLock() {
- signal_lock_.AssertAcquired();
- if (port_closed_ || in_transit_)
- return MOJO_RESULT_INVALID_ARGUMENT;
- port_closed_.Set(true);
- watchers_.NotifyClosed();
- if (!port_transferred_) {
- base::AutoUnlock unlock(signal_lock_);
- node_controller_->ClosePort(port_);
- #if BUILDFLAG(MOJO_TRACE_ENABLED)
- TRACE_EVENT_WITH_FLOW0(TRACE_DISABLED_BY_DEFAULT("mojom"),
- "MessagePipe closing", pipe_id_ + endpoint_,
- TRACE_EVENT_FLAG_FLOW_OUT);
- #endif
- }
- return MOJO_RESULT_OK;
- }
- HandleSignalsState MessagePipeDispatcher::GetHandleSignalsStateNoLock() const {
- HandleSignalsState rv;
- ports::PortStatus port_status;
- if (node_controller_->node()->GetStatus(port_, &port_status) != ports::OK) {
- CHECK(in_transit_ || port_transferred_ || port_closed_);
- return HandleSignalsState();
- }
- if (port_status.has_messages) {
- rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
- rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
- }
- if (port_status.receiving_messages)
- rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
- if (!port_status.peer_closed) {
- rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
- rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
- rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
- rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_REMOTE;
- if (port_status.peer_remote)
- rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_REMOTE;
- } else {
- rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
- }
- if (receive_queue_length_limit_ &&
- port_status.queued_message_count > *receive_queue_length_limit_) {
- rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_QUOTA_EXCEEDED;
- } else if (receive_queue_memory_size_limit_ &&
- port_status.queued_num_bytes > *receive_queue_memory_size_limit_) {
- rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_QUOTA_EXCEEDED;
- } else if (unread_message_count_limit_ &&
- port_status.unacknowledged_message_count >
- *unread_message_count_limit_) {
- rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_QUOTA_EXCEEDED;
- }
- rv.satisfiable_signals |=
- MOJO_HANDLE_SIGNAL_PEER_CLOSED | MOJO_HANDLE_SIGNAL_QUOTA_EXCEEDED;
- #if BUILDFLAG(MOJO_TRACE_ENABLED)
- const bool was_peer_closed =
- last_known_satisfied_signals_ & MOJO_HANDLE_SIGNAL_PEER_CLOSED;
- const bool is_peer_closed =
- rv.satisfied_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED;
- if (is_peer_closed && !was_peer_closed) {
- TRACE_EVENT_WITH_FLOW0(
- TRACE_DISABLED_BY_DEFAULT("mojom"), "MessagePipe peer closed",
- pipe_id_ + (1 - endpoint_), TRACE_EVENT_FLAG_FLOW_IN);
- }
- #endif
- last_known_satisfied_signals_ = rv.satisfied_signals;
- return rv;
- }
- void MessagePipeDispatcher::OnPortStatusChanged() {
- DCHECK(RequestContext::current());
- base::AutoLock lock(signal_lock_);
- // We stop observing our port as soon as it's transferred, but this can race
- // with events which are raised right before that happens. This is fine to
- // ignore.
- if (port_transferred_)
- return;
- #if DCHECK_IS_ON()
- ports::PortStatus port_status;
- if (node_controller_->node()->GetStatus(port_, &port_status) == ports::OK) {
- if (port_status.has_messages) {
- std::unique_ptr<ports::UserMessageEvent> unused;
- PeekSizeMessageFilter filter;
- node_controller_->node()->GetMessage(port_, &unused, &filter);
- DVLOG(4) << "New message detected on message pipe " << pipe_id_
- << " endpoint " << endpoint_ << " [port=" << port_.name()
- << "; size=" << filter.message_size() << "]";
- }
- if (port_status.peer_closed) {
- DVLOG(2) << "Peer closure detected on message pipe " << pipe_id_
- << " endpoint " << endpoint_ << " [port=" << port_.name() << "]";
- }
- }
- #endif
- watchers_.NotifyState(GetHandleSignalsStateNoLock());
- }
- } // namespace core
- } // namespace mojo
|