12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226 |
- // Copyright 2014 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 "ipc/ipc_mojo_bootstrap.h"
- #include <inttypes.h>
- #include <stdint.h>
- #include <map>
- #include <memory>
- #include <set>
- #include <utility>
- #include <vector>
- #include "base/bind.h"
- #include "base/callback.h"
- #include "base/check_op.h"
- #include "base/containers/contains.h"
- #include "base/containers/queue.h"
- #include "base/memory/ptr_util.h"
- #include "base/memory/raw_ptr.h"
- #include "base/no_destructor.h"
- #include "base/strings/stringprintf.h"
- #include "base/synchronization/lock.h"
- #include "base/task/common/task_annotator.h"
- #include "base/task/sequenced_task_runner.h"
- #include "base/task/single_thread_task_runner.h"
- #include "base/threading/thread_checker.h"
- #include "base/threading/thread_local.h"
- #include "base/threading/thread_task_runner_handle.h"
- #include "base/trace_event/memory_allocator_dump.h"
- #include "base/trace_event/memory_dump_manager.h"
- #include "base/trace_event/memory_dump_provider.h"
- #include "base/trace_event/typed_macros.h"
- #include "ipc/ipc_channel.h"
- #include "mojo/public/cpp/bindings/associated_group.h"
- #include "mojo/public/cpp/bindings/associated_group_controller.h"
- #include "mojo/public/cpp/bindings/connector.h"
- #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
- #include "mojo/public/cpp/bindings/interface_endpoint_controller.h"
- #include "mojo/public/cpp/bindings/interface_id.h"
- #include "mojo/public/cpp/bindings/message.h"
- #include "mojo/public/cpp/bindings/message_header_validator.h"
- #include "mojo/public/cpp/bindings/mojo_buildflags.h"
- #include "mojo/public/cpp/bindings/pipe_control_message_handler.h"
- #include "mojo/public/cpp/bindings/pipe_control_message_handler_delegate.h"
- #include "mojo/public/cpp/bindings/pipe_control_message_proxy.h"
- #include "mojo/public/cpp/bindings/sequence_local_sync_event_watcher.h"
- #include "mojo/public/cpp/bindings/tracing_helpers.h"
- namespace IPC {
- namespace {
- class ChannelAssociatedGroupController;
- base::ThreadLocalBoolean& GetOffSequenceBindingAllowedFlag() {
- static base::NoDestructor<base::ThreadLocalBoolean> flag;
- return *flag;
- }
- bool CanBindOffSequence() {
- return GetOffSequenceBindingAllowedFlag().Get();
- }
- // Used to track some internal Channel state in pursuit of message leaks.
- //
- // TODO(https://crbug.com/813045): Remove this.
- class ControllerMemoryDumpProvider
- : public base::trace_event::MemoryDumpProvider {
- public:
- ControllerMemoryDumpProvider() {
- base::trace_event::MemoryDumpManager::GetInstance()->RegisterDumpProvider(
- this, "IPCChannel", nullptr);
- }
- ControllerMemoryDumpProvider(const ControllerMemoryDumpProvider&) = delete;
- ControllerMemoryDumpProvider& operator=(const ControllerMemoryDumpProvider&) =
- delete;
- ~ControllerMemoryDumpProvider() override {
- base::trace_event::MemoryDumpManager::GetInstance()->UnregisterDumpProvider(
- this);
- }
- void AddController(ChannelAssociatedGroupController* controller) {
- base::AutoLock lock(lock_);
- controllers_.insert(controller);
- }
- void RemoveController(ChannelAssociatedGroupController* controller) {
- base::AutoLock lock(lock_);
- controllers_.erase(controller);
- }
- // base::trace_event::MemoryDumpProvider:
- bool OnMemoryDump(const base::trace_event::MemoryDumpArgs& args,
- base::trace_event::ProcessMemoryDump* pmd) override;
- private:
- base::Lock lock_;
- std::set<ChannelAssociatedGroupController*> controllers_;
- };
- ControllerMemoryDumpProvider& GetMemoryDumpProvider() {
- static base::NoDestructor<ControllerMemoryDumpProvider> provider;
- return *provider;
- }
- // Messages are grouped by this info when recording memory metrics.
- struct MessageMemoryDumpInfo {
- MessageMemoryDumpInfo(const mojo::Message& message)
- : id(message.name()), profiler_tag(message.heap_profiler_tag()) {}
- MessageMemoryDumpInfo() = default;
- bool operator==(const MessageMemoryDumpInfo& other) const {
- return other.id == id && other.profiler_tag == profiler_tag;
- }
- uint32_t id = 0;
- const char* profiler_tag = nullptr;
- };
- struct MessageMemoryDumpInfoHash {
- size_t operator()(const MessageMemoryDumpInfo& info) const {
- return base::HashInts(
- info.id, info.profiler_tag ? base::FastHash(info.profiler_tag) : 0);
- }
- };
- class ChannelAssociatedGroupController
- : public mojo::AssociatedGroupController,
- public mojo::MessageReceiver,
- public mojo::PipeControlMessageHandlerDelegate {
- public:
- ChannelAssociatedGroupController(
- bool set_interface_id_namespace_bit,
- const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
- const scoped_refptr<base::SingleThreadTaskRunner>& proxy_task_runner,
- const scoped_refptr<mojo::internal::MessageQuotaChecker>& quota_checker)
- : task_runner_(ipc_task_runner),
- proxy_task_runner_(proxy_task_runner),
- quota_checker_(quota_checker),
- set_interface_id_namespace_bit_(set_interface_id_namespace_bit),
- dispatcher_(this),
- control_message_handler_(this),
- control_message_proxy_thunk_(this),
- control_message_proxy_(&control_message_proxy_thunk_) {
- thread_checker_.DetachFromThread();
- control_message_handler_.SetDescription(
- "IPC::mojom::Bootstrap [primary] PipeControlMessageHandler");
- dispatcher_.SetValidator(std::make_unique<mojo::MessageHeaderValidator>(
- "IPC::mojom::Bootstrap [primary] MessageHeaderValidator"));
- GetMemoryDumpProvider().AddController(this);
- }
- ChannelAssociatedGroupController(const ChannelAssociatedGroupController&) =
- delete;
- ChannelAssociatedGroupController& operator=(
- const ChannelAssociatedGroupController&) = delete;
- size_t GetQueuedMessageCount() {
- base::AutoLock lock(outgoing_messages_lock_);
- return outgoing_messages_.size();
- }
- void GetTopQueuedMessageMemoryDumpInfo(MessageMemoryDumpInfo* info,
- size_t* count) {
- std::unordered_map<MessageMemoryDumpInfo, size_t, MessageMemoryDumpInfoHash>
- counts;
- std::pair<MessageMemoryDumpInfo, size_t> top_message_info_and_count = {
- MessageMemoryDumpInfo(), 0};
- base::AutoLock lock(outgoing_messages_lock_);
- for (const auto& message : outgoing_messages_) {
- auto it_and_inserted = counts.emplace(MessageMemoryDumpInfo(message), 0);
- it_and_inserted.first->second++;
- if (it_and_inserted.first->second > top_message_info_and_count.second)
- top_message_info_and_count = *it_and_inserted.first;
- }
- *info = top_message_info_and_count.first;
- *count = top_message_info_and_count.second;
- }
- void Pause() {
- DCHECK(!paused_);
- paused_ = true;
- }
- void Unpause() {
- DCHECK(paused_);
- paused_ = false;
- }
- void FlushOutgoingMessages() {
- std::vector<mojo::Message> outgoing_messages;
- {
- base::AutoLock lock(outgoing_messages_lock_);
- std::swap(outgoing_messages, outgoing_messages_);
- }
- if (quota_checker_ && outgoing_messages.size())
- quota_checker_->AfterMessagesDequeued(outgoing_messages.size());
- for (auto& message : outgoing_messages)
- SendMessage(&message);
- }
- void Bind(mojo::ScopedMessagePipeHandle handle,
- mojo::PendingAssociatedRemote<mojom::Channel>* sender,
- mojo::PendingAssociatedReceiver<mojom::Channel>* receiver) {
- connector_ = std::make_unique<mojo::Connector>(
- std::move(handle), mojo::Connector::SINGLE_THREADED_SEND,
- "IPC Channel");
- connector_->set_incoming_receiver(&dispatcher_);
- connector_->set_connection_error_handler(
- base::BindOnce(&ChannelAssociatedGroupController::OnPipeError,
- base::Unretained(this)));
- connector_->set_enforce_errors_from_incoming_receiver(false);
- if (quota_checker_)
- connector_->SetMessageQuotaChecker(quota_checker_);
- // Don't let the Connector do any sort of queuing on our behalf. Individual
- // messages bound for the IPC::ChannelProxy thread (i.e. that vast majority
- // of messages received by this Connector) are already individually
- // scheduled for dispatch by ChannelProxy, so Connector's normal mode of
- // operation would only introduce a redundant scheduling step for most
- // messages.
- connector_->set_force_immediate_dispatch(true);
- mojo::InterfaceId sender_id, receiver_id;
- if (set_interface_id_namespace_bit_) {
- sender_id = 1 | mojo::kInterfaceIdNamespaceMask;
- receiver_id = 1;
- } else {
- sender_id = 1;
- receiver_id = 1 | mojo::kInterfaceIdNamespaceMask;
- }
- {
- base::AutoLock locker(lock_);
- Endpoint* sender_endpoint = new Endpoint(this, sender_id);
- Endpoint* receiver_endpoint = new Endpoint(this, receiver_id);
- endpoints_.insert({ sender_id, sender_endpoint });
- endpoints_.insert({ receiver_id, receiver_endpoint });
- sender_endpoint->set_handle_created();
- receiver_endpoint->set_handle_created();
- }
- mojo::ScopedInterfaceEndpointHandle sender_handle =
- CreateScopedInterfaceEndpointHandle(sender_id);
- mojo::ScopedInterfaceEndpointHandle receiver_handle =
- CreateScopedInterfaceEndpointHandle(receiver_id);
- *sender = mojo::PendingAssociatedRemote<mojom::Channel>(
- std::move(sender_handle), 0);
- *receiver = mojo::PendingAssociatedReceiver<mojom::Channel>(
- std::move(receiver_handle));
- }
- void StartReceiving() { connector_->StartReceiving(task_runner_); }
- void ShutDown() {
- DCHECK(thread_checker_.CalledOnValidThread());
- shut_down_ = true;
- if (connector_)
- connector_->CloseMessagePipe();
- OnPipeError();
- connector_.reset();
- base::AutoLock lock(outgoing_messages_lock_);
- if (quota_checker_ && outgoing_messages_.size())
- quota_checker_->AfterMessagesDequeued(outgoing_messages_.size());
- outgoing_messages_.clear();
- }
- // mojo::AssociatedGroupController:
- mojo::InterfaceId AssociateInterface(
- mojo::ScopedInterfaceEndpointHandle handle_to_send) override {
- if (!handle_to_send.pending_association())
- return mojo::kInvalidInterfaceId;
- uint32_t id = 0;
- {
- base::AutoLock locker(lock_);
- do {
- if (next_interface_id_ >= mojo::kInterfaceIdNamespaceMask)
- next_interface_id_ = 2;
- id = next_interface_id_++;
- if (set_interface_id_namespace_bit_)
- id |= mojo::kInterfaceIdNamespaceMask;
- } while (base::Contains(endpoints_, id));
- Endpoint* endpoint = new Endpoint(this, id);
- if (encountered_error_)
- endpoint->set_peer_closed();
- endpoint->set_handle_created();
- endpoints_.insert({id, endpoint});
- }
- if (!NotifyAssociation(&handle_to_send, id)) {
- // The peer handle of |handle_to_send|, which is supposed to join this
- // associated group, has been closed.
- {
- base::AutoLock locker(lock_);
- Endpoint* endpoint = FindEndpoint(id);
- if (endpoint)
- MarkClosedAndMaybeRemove(endpoint);
- }
- control_message_proxy_.NotifyPeerEndpointClosed(
- id, handle_to_send.disconnect_reason());
- }
- return id;
- }
- mojo::ScopedInterfaceEndpointHandle CreateLocalEndpointHandle(
- mojo::InterfaceId id) override {
- if (!mojo::IsValidInterfaceId(id))
- return mojo::ScopedInterfaceEndpointHandle();
- // Unless it is the primary ID, |id| is from the remote side and therefore
- // its namespace bit is supposed to be different than the value that this
- // router would use.
- if (!mojo::IsPrimaryInterfaceId(id) &&
- set_interface_id_namespace_bit_ ==
- mojo::HasInterfaceIdNamespaceBitSet(id)) {
- return mojo::ScopedInterfaceEndpointHandle();
- }
- base::AutoLock locker(lock_);
- bool inserted = false;
- Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
- if (inserted) {
- DCHECK(!endpoint->handle_created());
- if (encountered_error_)
- endpoint->set_peer_closed();
- } else {
- if (endpoint->handle_created())
- return mojo::ScopedInterfaceEndpointHandle();
- }
- endpoint->set_handle_created();
- return CreateScopedInterfaceEndpointHandle(id);
- }
- void CloseEndpointHandle(
- mojo::InterfaceId id,
- const absl::optional<mojo::DisconnectReason>& reason) override {
- if (!mojo::IsValidInterfaceId(id))
- return;
- {
- base::AutoLock locker(lock_);
- DCHECK(base::Contains(endpoints_, id));
- Endpoint* endpoint = endpoints_[id].get();
- DCHECK(!endpoint->client());
- DCHECK(!endpoint->closed());
- MarkClosedAndMaybeRemove(endpoint);
- }
- if (!mojo::IsPrimaryInterfaceId(id) || reason)
- control_message_proxy_.NotifyPeerEndpointClosed(id, reason);
- }
- mojo::InterfaceEndpointController* AttachEndpointClient(
- const mojo::ScopedInterfaceEndpointHandle& handle,
- mojo::InterfaceEndpointClient* client,
- scoped_refptr<base::SequencedTaskRunner> runner) override {
- const mojo::InterfaceId id = handle.id();
- DCHECK(mojo::IsValidInterfaceId(id));
- DCHECK(client);
- base::AutoLock locker(lock_);
- DCHECK(base::Contains(endpoints_, id));
- Endpoint* endpoint = endpoints_[id].get();
- endpoint->AttachClient(client, std::move(runner));
- if (endpoint->peer_closed())
- NotifyEndpointOfError(endpoint, true /* force_async */);
- return endpoint;
- }
- void DetachEndpointClient(
- const mojo::ScopedInterfaceEndpointHandle& handle) override {
- const mojo::InterfaceId id = handle.id();
- DCHECK(mojo::IsValidInterfaceId(id));
- base::AutoLock locker(lock_);
- DCHECK(base::Contains(endpoints_, id));
- Endpoint* endpoint = endpoints_[id].get();
- endpoint->DetachClient();
- }
- void RaiseError() override {
- // We ignore errors on channel endpoints, leaving the pipe open. There are
- // good reasons for this:
- //
- // * We should never close a channel endpoint in either process as long as
- // the child process is still alive. The child's endpoint should only be
- // closed implicitly by process death, and the browser's endpoint should
- // only be closed after the child process is confirmed to be dead. Crash
- // reporting logic in Chrome relies on this behavior in order to do the
- // right thing.
- //
- // * There are two interesting conditions under which RaiseError() can be
- // implicitly reached: an incoming message fails validation, or the
- // local endpoint drops a response callback without calling it.
- //
- // * In the validation case, we also report the message as bad, and this
- // will imminently trigger the common bad-IPC path in the browser,
- // causing the browser to kill the offending renderer.
- //
- // * In the dropped response callback case, the net result of ignoring the
- // issue is generally innocuous. While indicative of programmer error,
- // it's not a severe failure and is already covered by separate DCHECKs.
- //
- // See https://crbug.com/861607 for additional discussion.
- }
- bool PrefersSerializedMessages() override { return true; }
- private:
- class Endpoint;
- class ControlMessageProxyThunk;
- friend class Endpoint;
- friend class ControlMessageProxyThunk;
- // MessageWrapper objects are always destroyed under the controller's lock. On
- // destruction, if the message it wrappers contains
- // ScopedInterfaceEndpointHandles (which cannot be destructed under the
- // controller's lock), the wrapper unlocks to clean them up.
- class MessageWrapper {
- public:
- MessageWrapper() = default;
- MessageWrapper(ChannelAssociatedGroupController* controller,
- mojo::Message message)
- : controller_(controller), value_(std::move(message)) {}
- MessageWrapper(MessageWrapper&& other)
- : controller_(other.controller_), value_(std::move(other.value_)) {}
- MessageWrapper(const MessageWrapper&) = delete;
- MessageWrapper& operator=(const MessageWrapper&) = delete;
- ~MessageWrapper() {
- if (value_.associated_endpoint_handles()->empty())
- return;
- controller_->lock_.AssertAcquired();
- {
- base::AutoUnlock unlocker(controller_->lock_);
- value_.mutable_associated_endpoint_handles()->clear();
- }
- }
- MessageWrapper& operator=(MessageWrapper&& other) {
- controller_ = other.controller_;
- value_ = std::move(other.value_);
- return *this;
- }
- mojo::Message& value() { return value_; }
- private:
- raw_ptr<ChannelAssociatedGroupController> controller_ = nullptr;
- mojo::Message value_;
- };
- class Endpoint : public base::RefCountedThreadSafe<Endpoint>,
- public mojo::InterfaceEndpointController {
- public:
- Endpoint(ChannelAssociatedGroupController* controller, mojo::InterfaceId id)
- : controller_(controller), id_(id) {}
- Endpoint(const Endpoint&) = delete;
- Endpoint& operator=(const Endpoint&) = delete;
- mojo::InterfaceId id() const { return id_; }
- bool closed() const {
- controller_->lock_.AssertAcquired();
- return closed_;
- }
- void set_closed() {
- controller_->lock_.AssertAcquired();
- closed_ = true;
- }
- bool peer_closed() const {
- controller_->lock_.AssertAcquired();
- return peer_closed_;
- }
- void set_peer_closed() {
- controller_->lock_.AssertAcquired();
- peer_closed_ = true;
- }
- bool handle_created() const {
- controller_->lock_.AssertAcquired();
- return handle_created_;
- }
- void set_handle_created() {
- controller_->lock_.AssertAcquired();
- handle_created_ = true;
- }
- const absl::optional<mojo::DisconnectReason>& disconnect_reason() const {
- return disconnect_reason_;
- }
- void set_disconnect_reason(
- const absl::optional<mojo::DisconnectReason>& disconnect_reason) {
- disconnect_reason_ = disconnect_reason;
- }
- base::SequencedTaskRunner* task_runner() const {
- return task_runner_.get();
- }
- bool was_bound_off_sequence() const { return was_bound_off_sequence_; }
- mojo::InterfaceEndpointClient* client() const {
- controller_->lock_.AssertAcquired();
- return client_;
- }
- void AttachClient(mojo::InterfaceEndpointClient* client,
- scoped_refptr<base::SequencedTaskRunner> runner) {
- controller_->lock_.AssertAcquired();
- DCHECK(!client_);
- DCHECK(!closed_);
- task_runner_ = std::move(runner);
- client_ = client;
- if (CanBindOffSequence())
- was_bound_off_sequence_ = true;
- }
- void DetachClient() {
- controller_->lock_.AssertAcquired();
- DCHECK(client_);
- DCHECK(!closed_);
- task_runner_ = nullptr;
- client_ = nullptr;
- sync_watcher_.reset();
- }
- uint32_t EnqueueSyncMessage(MessageWrapper message) {
- controller_->lock_.AssertAcquired();
- uint32_t id = GenerateSyncMessageId();
- sync_messages_.emplace(id, std::move(message));
- SignalSyncMessageEvent();
- return id;
- }
- void SignalSyncMessageEvent() {
- controller_->lock_.AssertAcquired();
- if (sync_watcher_)
- sync_watcher_->SignalEvent();
- }
- MessageWrapper PopSyncMessage(uint32_t id) {
- controller_->lock_.AssertAcquired();
- if (sync_messages_.empty() || sync_messages_.front().first != id)
- return MessageWrapper();
- MessageWrapper message = std::move(sync_messages_.front().second);
- sync_messages_.pop();
- return message;
- }
- // mojo::InterfaceEndpointController:
- bool SendMessage(mojo::Message* message) override {
- DCHECK(task_runner_->RunsTasksInCurrentSequence());
- message->set_interface_id(id_);
- return controller_->SendMessage(message);
- }
- void AllowWokenUpBySyncWatchOnSameThread() override {
- DCHECK(task_runner_->RunsTasksInCurrentSequence());
- EnsureSyncWatcherExists();
- sync_watcher_->AllowWokenUpBySyncWatchOnSameSequence();
- }
- bool SyncWatch(const bool& should_stop) override {
- DCHECK(task_runner_->RunsTasksInCurrentSequence());
- // It's not legal to make sync calls from the primary endpoint's thread,
- // and in fact they must only happen from the proxy task runner.
- DCHECK(!controller_->task_runner_->BelongsToCurrentThread());
- DCHECK(controller_->proxy_task_runner_->BelongsToCurrentThread());
- EnsureSyncWatcherExists();
- return sync_watcher_->SyncWatch(&should_stop);
- }
- bool SyncWatchExclusive(uint64_t request_id) override {
- // We don't support exclusive waits on Channel-associated interfaces.
- NOTREACHED();
- return false;
- }
- void RegisterExternalSyncWaiter(uint64_t request_id) override {}
- private:
- friend class base::RefCountedThreadSafe<Endpoint>;
- ~Endpoint() override {
- controller_->lock_.AssertAcquired();
- DCHECK(!client_);
- DCHECK(closed_);
- DCHECK(peer_closed_);
- DCHECK(!sync_watcher_);
- }
- void OnSyncMessageEventReady() {
- DCHECK(task_runner_->RunsTasksInCurrentSequence());
- scoped_refptr<Endpoint> keepalive(this);
- scoped_refptr<AssociatedGroupController> controller_keepalive(
- controller_.get());
- base::AutoLock locker(controller_->lock_);
- bool more_to_process = false;
- if (!sync_messages_.empty()) {
- MessageWrapper message_wrapper =
- std::move(sync_messages_.front().second);
- sync_messages_.pop();
- bool dispatch_succeeded;
- mojo::InterfaceEndpointClient* client = client_;
- {
- base::AutoUnlock unlocker(controller_->lock_);
- dispatch_succeeded =
- client->HandleIncomingMessage(&message_wrapper.value());
- }
- if (!sync_messages_.empty())
- more_to_process = true;
- if (!dispatch_succeeded)
- controller_->RaiseError();
- }
- if (!more_to_process)
- sync_watcher_->ResetEvent();
- // If there are no queued sync messages and the peer has closed, there
- // there won't be incoming sync messages in the future. If any
- // SyncWatch() calls are on the stack for this endpoint, resetting the
- // watcher will allow them to exit as the stack undwinds.
- if (!more_to_process && peer_closed_)
- sync_watcher_.reset();
- }
- void EnsureSyncWatcherExists() {
- DCHECK(task_runner_->RunsTasksInCurrentSequence());
- if (sync_watcher_)
- return;
- base::AutoLock locker(controller_->lock_);
- sync_watcher_ = std::make_unique<mojo::SequenceLocalSyncEventWatcher>(
- base::BindRepeating(&Endpoint::OnSyncMessageEventReady,
- base::Unretained(this)));
- if (peer_closed_ || !sync_messages_.empty())
- SignalSyncMessageEvent();
- }
- uint32_t GenerateSyncMessageId() {
- // Overflow is fine.
- uint32_t id = next_sync_message_id_++;
- DCHECK(sync_messages_.empty() || sync_messages_.front().first != id);
- return id;
- }
- const raw_ptr<ChannelAssociatedGroupController> controller_;
- const mojo::InterfaceId id_;
- bool closed_ = false;
- bool peer_closed_ = false;
- bool handle_created_ = false;
- bool was_bound_off_sequence_ = false;
- absl::optional<mojo::DisconnectReason> disconnect_reason_;
- raw_ptr<mojo::InterfaceEndpointClient> client_ = nullptr;
- scoped_refptr<base::SequencedTaskRunner> task_runner_;
- std::unique_ptr<mojo::SequenceLocalSyncEventWatcher> sync_watcher_;
- base::queue<std::pair<uint32_t, MessageWrapper>> sync_messages_;
- uint32_t next_sync_message_id_ = 0;
- };
- class ControlMessageProxyThunk : public MessageReceiver {
- public:
- explicit ControlMessageProxyThunk(
- ChannelAssociatedGroupController* controller)
- : controller_(controller) {}
- ControlMessageProxyThunk(const ControlMessageProxyThunk&) = delete;
- ControlMessageProxyThunk& operator=(const ControlMessageProxyThunk&) =
- delete;
- private:
- // MessageReceiver:
- bool Accept(mojo::Message* message) override {
- return controller_->SendMessage(message);
- }
- raw_ptr<ChannelAssociatedGroupController> controller_;
- };
- ~ChannelAssociatedGroupController() override {
- DCHECK(!connector_);
- base::AutoLock locker(lock_);
- for (auto iter = endpoints_.begin(); iter != endpoints_.end();) {
- Endpoint* endpoint = iter->second.get();
- ++iter;
- if (!endpoint->closed()) {
- // This happens when a NotifyPeerEndpointClosed message been received,
- // but the interface ID hasn't been used to create local endpoint
- // handle.
- DCHECK(!endpoint->client());
- DCHECK(endpoint->peer_closed());
- MarkClosedAndMaybeRemove(endpoint);
- } else {
- MarkPeerClosedAndMaybeRemove(endpoint);
- }
- }
- DCHECK(endpoints_.empty());
- GetMemoryDumpProvider().RemoveController(this);
- }
- bool SendMessage(mojo::Message* message) {
- DCHECK(message->heap_profiler_tag());
- if (task_runner_->BelongsToCurrentThread()) {
- DCHECK(thread_checker_.CalledOnValidThread());
- if (!connector_ || paused_) {
- if (!shut_down_) {
- base::AutoLock lock(outgoing_messages_lock_);
- if (quota_checker_)
- quota_checker_->BeforeMessagesEnqueued(1);
- outgoing_messages_.emplace_back(std::move(*message));
- }
- return true;
- }
- return connector_->Accept(message);
- } else {
- // We always post tasks to the primary endpoint thread when called from
- // other threads in order to simulate IPC::ChannelProxy::Send behavior.
- task_runner_->PostTask(
- FROM_HERE,
- base::BindOnce(
- &ChannelAssociatedGroupController::SendMessageOnPrimaryThread,
- this, std::move(*message)));
- return true;
- }
- }
- void SendMessageOnPrimaryThread(mojo::Message message) {
- DCHECK(thread_checker_.CalledOnValidThread());
- if (!SendMessage(&message))
- RaiseError();
- }
- void OnPipeError() {
- DCHECK(thread_checker_.CalledOnValidThread());
- // We keep |this| alive here because it's possible for the notifications
- // below to release all other references.
- scoped_refptr<ChannelAssociatedGroupController> keepalive(this);
- base::AutoLock locker(lock_);
- encountered_error_ = true;
- std::vector<scoped_refptr<Endpoint>> endpoints_to_notify;
- for (auto iter = endpoints_.begin(); iter != endpoints_.end();) {
- Endpoint* endpoint = iter->second.get();
- ++iter;
- if (endpoint->client())
- endpoints_to_notify.push_back(endpoint);
- MarkPeerClosedAndMaybeRemove(endpoint);
- }
- for (auto& endpoint : endpoints_to_notify) {
- // Because a notification may in turn detach any endpoint, we have to
- // check each client again here.
- if (endpoint->client())
- NotifyEndpointOfError(endpoint.get(), false /* force_async */);
- }
- }
- void NotifyEndpointOfError(Endpoint* endpoint, bool force_async) {
- lock_.AssertAcquired();
- DCHECK(endpoint->task_runner() && endpoint->client());
- if (endpoint->task_runner()->RunsTasksInCurrentSequence() && !force_async) {
- mojo::InterfaceEndpointClient* client = endpoint->client();
- absl::optional<mojo::DisconnectReason> reason(
- endpoint->disconnect_reason());
- base::AutoUnlock unlocker(lock_);
- client->NotifyError(reason);
- } else {
- endpoint->task_runner()->PostTask(
- FROM_HERE,
- base::BindOnce(&ChannelAssociatedGroupController::
- NotifyEndpointOfErrorOnEndpointThread,
- this, endpoint->id(), base::Unretained(endpoint)));
- }
- }
- void NotifyEndpointOfErrorOnEndpointThread(mojo::InterfaceId id,
- Endpoint* endpoint) {
- base::AutoLock locker(lock_);
- auto iter = endpoints_.find(id);
- if (iter == endpoints_.end() || iter->second.get() != endpoint)
- return;
- if (!endpoint->client())
- return;
- DCHECK(endpoint->task_runner()->RunsTasksInCurrentSequence());
- NotifyEndpointOfError(endpoint, false /* force_async */);
- }
- void MarkClosedAndMaybeRemove(Endpoint* endpoint) {
- lock_.AssertAcquired();
- endpoint->set_closed();
- if (endpoint->closed() && endpoint->peer_closed())
- endpoints_.erase(endpoint->id());
- }
- void MarkPeerClosedAndMaybeRemove(Endpoint* endpoint) {
- lock_.AssertAcquired();
- endpoint->set_peer_closed();
- endpoint->SignalSyncMessageEvent();
- if (endpoint->closed() && endpoint->peer_closed())
- endpoints_.erase(endpoint->id());
- }
- Endpoint* FindOrInsertEndpoint(mojo::InterfaceId id, bool* inserted) {
- lock_.AssertAcquired();
- DCHECK(!inserted || !*inserted);
- Endpoint* endpoint = FindEndpoint(id);
- if (!endpoint) {
- endpoint = new Endpoint(this, id);
- endpoints_.insert({id, endpoint});
- if (inserted)
- *inserted = true;
- }
- return endpoint;
- }
- Endpoint* FindEndpoint(mojo::InterfaceId id) {
- lock_.AssertAcquired();
- auto iter = endpoints_.find(id);
- return iter != endpoints_.end() ? iter->second.get() : nullptr;
- }
- // mojo::MessageReceiver:
- bool Accept(mojo::Message* message) override {
- DCHECK(thread_checker_.CalledOnValidThread());
- if (!message->DeserializeAssociatedEndpointHandles(this))
- return false;
- if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message))
- return control_message_handler_.Accept(message);
- mojo::InterfaceId id = message->interface_id();
- if (!mojo::IsValidInterfaceId(id))
- return false;
- base::ReleasableAutoLock locker(&lock_);
- Endpoint* endpoint = FindEndpoint(id);
- if (!endpoint)
- return true;
- mojo::InterfaceEndpointClient* client = endpoint->client();
- if (!client || !endpoint->task_runner()->RunsTasksInCurrentSequence()) {
- // The ChannelProxy for this channel is bound to `proxy_task_runner_` and
- // by default legacy IPCs must dispatch to either the IO thread or the
- // proxy task runner. We generally impose the same constraint on
- // associated interface endpoints so that FIFO can be guaranteed across
- // all interfaces without stalling any of them to wait for a pending
- // endpoint to be bound.
- //
- // This allows us to assume that if an endpoint is not yet bound when we
- // receive a message targeting it, it *will* be bound on the proxy task
- // runner by the time a newly posted task runs there. Hence we simply post
- // a hopeful dispatch task to that task runner.
- //
- // As it turns out, there are even some instances of endpoints binding to
- // alternative (non-IO-thread, non-proxy) task runners, but still
- // ultimately relying on the fact that we schedule their messages on the
- // proxy task runner. So even if the endpoint is already bound, we
- // default to scheduling it on the proxy task runner as long as it's not
- // bound specifically to the IO task runner.
- // TODO(rockot): Try to sort out these cases and maybe eliminate them.
- //
- // Finally, it's also possible that an endpoint was bound to an
- // alternative task runner and it really does want its messages to
- // dispatch there. In that case `was_bound_off_sequence()` will be true to
- // signal that we should really use that task runner.
- const scoped_refptr<base::SequencedTaskRunner> task_runner =
- client && endpoint->was_bound_off_sequence()
- ? endpoint->task_runner()
- : proxy_task_runner_.get();
- if (message->has_flag(mojo::Message::kFlagIsSync)) {
- MessageWrapper message_wrapper(this, std::move(*message));
- // Sync messages may need to be handled by the endpoint if it's blocking
- // on a sync reply. We pass ownership of the message to the endpoint's
- // sync message queue. If the endpoint was blocking, it will dequeue the
- // message and dispatch it. Otherwise the posted |AcceptSyncMessage()|
- // call will dequeue the message and dispatch it.
- uint32_t message_id =
- endpoint->EnqueueSyncMessage(std::move(message_wrapper));
- task_runner->PostTask(
- FROM_HERE,
- base::BindOnce(&ChannelAssociatedGroupController::AcceptSyncMessage,
- this, id, message_id));
- return true;
- }
- // If |task_runner| has been torn down already, this PostTask will fail
- // and destroy |message|. That operation may need to in turn destroy
- // in-transit associated endpoints and thus acquire |lock_|. We no longer
- // need the lock to be held now, so we can release it before the PostTask.
- {
- // Grab interface name from |client| before releasing the lock to ensure
- // that |client| is safe to access.
- base::TaskAnnotator::ScopedSetIpcHash scoped_set_ipc_hash(
- client ? client->interface_name() : "unknown interface");
- locker.Release();
- task_runner->PostTask(
- FROM_HERE,
- base::BindOnce(
- &ChannelAssociatedGroupController::AcceptOnEndpointThread, this,
- std::move(*message)));
- }
- return true;
- }
- locker.Release();
- // It's safe to access |client| here without holding a lock, because this
- // code runs on a proxy thread and |client| can't be destroyed from any
- // thread.
- return client->HandleIncomingMessage(message);
- }
- void AcceptOnEndpointThread(mojo::Message message) {
- TRACE_EVENT0(TRACE_DISABLED_BY_DEFAULT("mojom"),
- "ChannelAssociatedGroupController::AcceptOnEndpointThread");
- mojo::InterfaceId id = message.interface_id();
- DCHECK(mojo::IsValidInterfaceId(id) && !mojo::IsPrimaryInterfaceId(id));
- base::AutoLock locker(lock_);
- Endpoint* endpoint = FindEndpoint(id);
- if (!endpoint)
- return;
- mojo::InterfaceEndpointClient* client = endpoint->client();
- if (!client)
- return;
- if (!endpoint->task_runner()->RunsTasksInCurrentSequence() &&
- !proxy_task_runner_->RunsTasksInCurrentSequence()) {
- return;
- }
- // TODO(altimin): This event is temporarily kept as a debug fallback. Remove
- // it once the new implementation proves to be stable.
- TRACE_EVENT(
- TRACE_DISABLED_BY_DEFAULT("mojom"),
- // Using client->interface_name() is safe here because this is a static
- // string defined for each mojo interface.
- perfetto::StaticString(client->interface_name()),
- [&](perfetto::EventContext& ctx) {
- static const uint8_t* toplevel_flow_enabled =
- TRACE_EVENT_API_GET_CATEGORY_GROUP_ENABLED("toplevel.flow");
- if (!*toplevel_flow_enabled)
- return;
- perfetto::Flow::Global(message.GetTraceId())(ctx);
- });
- // Sync messages should never make their way to this method.
- DCHECK(!message.has_flag(mojo::Message::kFlagIsSync));
- bool result = false;
- {
- base::AutoUnlock unlocker(lock_);
- result = client->HandleIncomingMessage(&message);
- }
- if (!result)
- RaiseError();
- }
- void AcceptSyncMessage(mojo::InterfaceId interface_id, uint32_t message_id) {
- TRACE_EVENT0(TRACE_DISABLED_BY_DEFAULT("mojom"),
- "ChannelAssociatedGroupController::AcceptSyncMessage");
- base::AutoLock locker(lock_);
- Endpoint* endpoint = FindEndpoint(interface_id);
- if (!endpoint)
- return;
- // Careful, if the endpoint is detached its members are cleared. Check for
- // that before dereferencing.
- mojo::InterfaceEndpointClient* client = endpoint->client();
- if (!client)
- return;
- if (!endpoint->task_runner()->RunsTasksInCurrentSequence() &&
- !proxy_task_runner_->RunsTasksInCurrentSequence()) {
- return;
- }
- // Using client->interface_name() is safe here because this is a static
- // string defined for each mojo interface.
- TRACE_EVENT0("mojom", client->interface_name());
- MessageWrapper message_wrapper = endpoint->PopSyncMessage(message_id);
- // The message must have already been dequeued by the endpoint waking up
- // from a sync wait. Nothing to do.
- if (message_wrapper.value().IsNull())
- return;
- bool result = false;
- {
- base::AutoUnlock unlocker(lock_);
- result = client->HandleIncomingMessage(&message_wrapper.value());
- }
- if (!result)
- RaiseError();
- }
- // mojo::PipeControlMessageHandlerDelegate:
- bool OnPeerAssociatedEndpointClosed(
- mojo::InterfaceId id,
- const absl::optional<mojo::DisconnectReason>& reason) override {
- DCHECK(thread_checker_.CalledOnValidThread());
- scoped_refptr<ChannelAssociatedGroupController> keepalive(this);
- base::AutoLock locker(lock_);
- scoped_refptr<Endpoint> endpoint = FindOrInsertEndpoint(id, nullptr);
- if (reason)
- endpoint->set_disconnect_reason(reason);
- if (!endpoint->peer_closed()) {
- if (endpoint->client())
- NotifyEndpointOfError(endpoint.get(), false /* force_async */);
- MarkPeerClosedAndMaybeRemove(endpoint.get());
- }
- return true;
- }
- bool WaitForFlushToComplete(
- mojo::ScopedMessagePipeHandle flush_pipe) override {
- // We don't support async flushing on the IPC Channel pipe.
- return false;
- }
- // Checked in places which must be run on the primary endpoint's thread.
- base::ThreadChecker thread_checker_;
- scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
- const scoped_refptr<base::SingleThreadTaskRunner> proxy_task_runner_;
- const scoped_refptr<mojo::internal::MessageQuotaChecker> quota_checker_;
- const bool set_interface_id_namespace_bit_;
- bool paused_ = false;
- std::unique_ptr<mojo::Connector> connector_;
- mojo::MessageDispatcher dispatcher_;
- mojo::PipeControlMessageHandler control_message_handler_;
- ControlMessageProxyThunk control_message_proxy_thunk_;
- // NOTE: It is unsafe to call into this object while holding |lock_|.
- mojo::PipeControlMessageProxy control_message_proxy_;
- // Guards access to |outgoing_messages_| only. Used to support memory dumps
- // which may be triggered from any thread.
- base::Lock outgoing_messages_lock_;
- // Outgoing messages that were sent before this controller was bound to a
- // real message pipe.
- std::vector<mojo::Message> outgoing_messages_;
- // Guards the fields below for thread-safe access.
- base::Lock lock_;
- bool encountered_error_ = false;
- bool shut_down_ = false;
- // ID #1 is reserved for the mojom::Channel interface.
- uint32_t next_interface_id_ = 2;
- std::map<uint32_t, scoped_refptr<Endpoint>> endpoints_;
- };
- bool ControllerMemoryDumpProvider::OnMemoryDump(
- const base::trace_event::MemoryDumpArgs& args,
- base::trace_event::ProcessMemoryDump* pmd) {
- base::AutoLock lock(lock_);
- for (auto* controller : controllers_) {
- base::trace_event::MemoryAllocatorDump* dump = pmd->CreateAllocatorDump(
- base::StringPrintf("mojo/queued_ipc_channel_message/0x%" PRIxPTR,
- reinterpret_cast<uintptr_t>(controller)));
- dump->AddScalar(base::trace_event::MemoryAllocatorDump::kNameObjectCount,
- base::trace_event::MemoryAllocatorDump::kUnitsObjects,
- controller->GetQueuedMessageCount());
- MessageMemoryDumpInfo info;
- size_t count = 0;
- controller->GetTopQueuedMessageMemoryDumpInfo(&info, &count);
- dump->AddScalar("top_message_name", "id", info.id);
- dump->AddScalar("top_message_count",
- base::trace_event::MemoryAllocatorDump::kUnitsObjects,
- count);
- if (info.profiler_tag) {
- // TODO(ssid): Memory dumps currently do not support adding string
- // arguments in background dumps. So, add this value as a trace event for
- // now.
- TRACE_EVENT2(base::trace_event::MemoryDumpManager::kTraceCategory,
- "ControllerMemoryDumpProvider::OnMemoryDump",
- "top_queued_message_tag", info.profiler_tag,
- "count", count);
- }
- }
- return true;
- }
- class MojoBootstrapImpl : public MojoBootstrap {
- public:
- MojoBootstrapImpl(
- mojo::ScopedMessagePipeHandle handle,
- const scoped_refptr<ChannelAssociatedGroupController> controller)
- : controller_(controller),
- associated_group_(controller),
- handle_(std::move(handle)) {}
- MojoBootstrapImpl(const MojoBootstrapImpl&) = delete;
- MojoBootstrapImpl& operator=(const MojoBootstrapImpl&) = delete;
- ~MojoBootstrapImpl() override {
- controller_->ShutDown();
- }
- private:
- void Connect(
- mojo::PendingAssociatedRemote<mojom::Channel>* sender,
- mojo::PendingAssociatedReceiver<mojom::Channel>* receiver) override {
- controller_->Bind(std::move(handle_), sender, receiver);
- }
- void StartReceiving() override { controller_->StartReceiving(); }
- void Pause() override {
- controller_->Pause();
- }
- void Unpause() override {
- controller_->Unpause();
- }
- void Flush() override {
- controller_->FlushOutgoingMessages();
- }
- mojo::AssociatedGroup* GetAssociatedGroup() override {
- return &associated_group_;
- }
- scoped_refptr<ChannelAssociatedGroupController> controller_;
- mojo::AssociatedGroup associated_group_;
- mojo::ScopedMessagePipeHandle handle_;
- };
- } // namespace
- ScopedAllowOffSequenceChannelAssociatedBindings::
- ScopedAllowOffSequenceChannelAssociatedBindings()
- : outer_flag_(GetOffSequenceBindingAllowedFlag().Get()) {
- GetOffSequenceBindingAllowedFlag().Set(true);
- }
- ScopedAllowOffSequenceChannelAssociatedBindings::
- ~ScopedAllowOffSequenceChannelAssociatedBindings() {
- GetOffSequenceBindingAllowedFlag().Set(outer_flag_);
- }
- // static
- std::unique_ptr<MojoBootstrap> MojoBootstrap::Create(
- mojo::ScopedMessagePipeHandle handle,
- Channel::Mode mode,
- const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
- const scoped_refptr<base::SingleThreadTaskRunner>& proxy_task_runner,
- const scoped_refptr<mojo::internal::MessageQuotaChecker>& quota_checker) {
- return std::make_unique<MojoBootstrapImpl>(
- std::move(handle), base::MakeRefCounted<ChannelAssociatedGroupController>(
- mode == Channel::MODE_SERVER, ipc_task_runner,
- proxy_task_runner, quota_checker));
- }
- } // namespace IPC
|