123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621 |
- // Copyright (c) 2012 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_channel_proxy.h"
- #include <stddef.h>
- #include <stdint.h>
- #include <utility>
- #include "base/bind.h"
- #include "base/compiler_specific.h"
- #include "base/containers/contains.h"
- #include "base/location.h"
- #include "base/memory/ptr_util.h"
- #include "base/memory/ref_counted.h"
- #include "base/task/single_thread_task_runner.h"
- #include "base/threading/thread_task_runner_handle.h"
- #include "build/build_config.h"
- #include "ipc/ipc_channel_factory.h"
- #include "ipc/ipc_listener.h"
- #include "ipc/ipc_logging.h"
- #include "ipc/ipc_message_macros.h"
- #include "ipc/message_filter.h"
- #include "ipc/message_filter_router.h"
- #include "mojo/public/cpp/bindings/pending_associated_receiver.h"
- namespace IPC {
- //------------------------------------------------------------------------------
- ChannelProxy::Context::Context(
- Listener* listener,
- const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
- const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner)
- : default_listener_task_runner_(listener_task_runner),
- listener_(listener),
- ipc_task_runner_(ipc_task_runner),
- channel_connected_called_(false),
- message_filter_router_(new MessageFilterRouter()),
- peer_pid_(base::kNullProcessId) {
- DCHECK(ipc_task_runner_.get());
- // The Listener thread where Messages are handled must be a separate thread
- // to avoid oversubscribing the IO thread. If you trigger this error, you
- // need to either:
- // 1) Create the ChannelProxy on a different thread, or
- // 2) Just use Channel
- // We make an exception for NULL listeners.
- DCHECK(!listener ||
- (ipc_task_runner_.get() != default_listener_task_runner_.get()));
- }
- ChannelProxy::Context::~Context() = default;
- void ChannelProxy::Context::ClearIPCTaskRunner() {
- ipc_task_runner_.reset();
- }
- void ChannelProxy::Context::CreateChannel(
- std::unique_ptr<ChannelFactory> factory) {
- base::AutoLock channel_lock(channel_lifetime_lock_);
- DCHECK(!channel_);
- DCHECK_EQ(factory->GetIPCTaskRunner(), ipc_task_runner_);
- channel_ = factory->BuildChannel(this);
- Channel::AssociatedInterfaceSupport* support =
- channel_->GetAssociatedInterfaceSupport();
- if (support) {
- thread_safe_channel_ = support->CreateThreadSafeChannel();
- base::AutoLock filter_lock(pending_filters_lock_);
- for (auto& entry : pending_io_thread_interfaces_)
- support->AddGenericAssociatedInterface(entry.first, entry.second);
- pending_io_thread_interfaces_.clear();
- }
- }
- bool ChannelProxy::Context::TryFilters(const Message& message) {
- DCHECK(message_filter_router_);
- #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
- Logging* logger = Logging::GetInstance();
- if (logger->Enabled())
- logger->OnPreDispatchMessage(message);
- #endif
- if (message_filter_router_->TryFilters(message)) {
- if (message.dispatch_error()) {
- GetTaskRunner(message.routing_id())
- ->PostTask(FROM_HERE, base::BindOnce(&Context::OnDispatchBadMessage,
- this, message));
- }
- #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
- if (logger->Enabled())
- logger->OnPostDispatchMessage(message);
- #endif
- return true;
- }
- return false;
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::PauseChannel() {
- DCHECK(channel_);
- channel_->Pause();
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::UnpauseChannel(bool flush) {
- DCHECK(channel_);
- channel_->Unpause(flush);
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::FlushChannel() {
- DCHECK(channel_);
- channel_->Flush();
- }
- // Called on the IPC::Channel thread
- bool ChannelProxy::Context::OnMessageReceived(const Message& message) {
- // First give a chance to the filters to process this message.
- if (!TryFilters(message))
- OnMessageReceivedNoFilter(message);
- return true;
- }
- // Called on the IPC::Channel thread
- bool ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) {
- GetTaskRunner(message.routing_id())
- ->PostTask(FROM_HERE,
- base::BindOnce(&Context::OnDispatchMessage, this, message));
- return true;
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnChannelConnected(int32_t peer_pid) {
- // We cache off the peer_pid so it can be safely accessed from both threads.
- {
- base::AutoLock l(peer_pid_lock_);
- peer_pid_ = peer_pid;
- }
- // Add any pending filters. This avoids a race condition where someone
- // creates a ChannelProxy, calls AddFilter, and then right after starts the
- // peer process. The IO thread could receive a message before the task to add
- // the filter is run on the IO thread.
- OnAddFilter();
- // See above comment about using default_listener_task_runner_ here.
- default_listener_task_runner_->PostTask(
- FROM_HERE, base::BindOnce(&Context::OnDispatchConnected, this));
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnChannelError() {
- for (size_t i = 0; i < filters_.size(); ++i)
- filters_[i]->OnChannelError();
- // See above comment about using default_listener_task_runner_ here.
- default_listener_task_runner_->PostTask(
- FROM_HERE, base::BindOnce(&Context::OnDispatchError, this));
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnAssociatedInterfaceRequest(
- const std::string& interface_name,
- mojo::ScopedInterfaceEndpointHandle handle) {
- default_listener_task_runner_->PostTask(
- FROM_HERE, base::BindOnce(&Context::OnDispatchAssociatedInterfaceRequest,
- this, interface_name, std::move(handle)));
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnChannelOpened() {
- DCHECK(channel_);
- // Assume a reference to ourselves on behalf of this thread. This reference
- // will be released when we are closed.
- AddRef();
- if (!channel_->Connect()) {
- OnChannelError();
- return;
- }
- for (size_t i = 0; i < filters_.size(); ++i)
- filters_[i]->OnFilterAdded(channel_.get());
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnChannelClosed() {
- // It's okay for IPC::ChannelProxy::Close to be called more than once, which
- // would result in this branch being taken.
- if (!channel_)
- return;
- for (auto& filter : pending_filters_) {
- filter->OnChannelClosing();
- filter->OnFilterRemoved();
- }
- for (auto& filter : filters_) {
- filter->OnChannelClosing();
- filter->OnFilterRemoved();
- }
- // We don't need the filters anymore.
- message_filter_router_->Clear();
- filters_.clear();
- // We don't need the lock, because at this point, the listener thread can't
- // access it any more.
- pending_filters_.clear();
- ClearChannel();
- // Balance with the reference taken during startup. This may result in
- // self-destruction.
- Release();
- }
- void ChannelProxy::Context::Clear() {
- listener_ = nullptr;
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnSendMessage(std::unique_ptr<Message> message) {
- if (quota_checker_)
- quota_checker_->AfterMessagesDequeued(1);
- if (!channel_) {
- OnChannelClosed();
- return;
- }
- if (!channel_->Send(message.release()))
- OnChannelError();
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnAddFilter() {
- // Our OnChannelConnected method has not yet been called, so we can't be
- // sure that channel_ is valid yet. When OnChannelConnected *is* called,
- // it invokes OnAddFilter, so any pending filter(s) will be added at that
- // time.
- // No lock necessary for |peer_pid_| because it is only modified on this
- // thread.
- if (peer_pid_ == base::kNullProcessId)
- return;
- std::vector<scoped_refptr<MessageFilter> > new_filters;
- {
- base::AutoLock auto_lock(pending_filters_lock_);
- new_filters.swap(pending_filters_);
- }
- for (size_t i = 0; i < new_filters.size(); ++i) {
- filters_.push_back(new_filters[i]);
- message_filter_router_->AddFilter(new_filters[i].get());
- // The channel has already been created and connected, so we need to
- // inform the filters right now.
- new_filters[i]->OnFilterAdded(channel_.get());
- new_filters[i]->OnChannelConnected(peer_pid_);
- }
- }
- // Called on the IPC::Channel thread
- void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
- // No lock necessary for |peer_pid_| because it is only modified on this
- // thread.
- if (peer_pid_ == base::kNullProcessId) {
- // The channel is not yet connected, so any filters are still pending.
- base::AutoLock auto_lock(pending_filters_lock_);
- for (size_t i = 0; i < pending_filters_.size(); ++i) {
- if (pending_filters_[i].get() == filter) {
- filter->OnFilterRemoved();
- pending_filters_.erase(pending_filters_.begin() + i);
- return;
- }
- }
- return;
- }
- if (!channel_)
- return; // The filters have already been deleted.
- message_filter_router_->RemoveFilter(filter);
- for (size_t i = 0; i < filters_.size(); ++i) {
- if (filters_[i].get() == filter) {
- filter->OnFilterRemoved();
- filters_.erase(filters_.begin() + i);
- return;
- }
- }
- NOTREACHED() << "filter to be removed not found";
- }
- // Called on the listener's thread
- void ChannelProxy::Context::AddFilter(MessageFilter* filter) {
- base::AutoLock auto_lock(pending_filters_lock_);
- pending_filters_.push_back(base::WrapRefCounted(filter));
- ipc_task_runner_->PostTask(FROM_HERE,
- base::BindOnce(&Context::OnAddFilter, this));
- }
- // Called on the listener's thread
- void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
- if (!listener_)
- return;
- OnDispatchConnected();
- #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
- Logging* logger = Logging::GetInstance();
- if (message.type() == IPC_LOGGING_ID) {
- logger->OnReceivedLoggingMessage(message);
- return;
- }
- if (logger->Enabled())
- logger->OnPreDispatchMessage(message);
- #endif
- listener_->OnMessageReceived(message);
- if (message.dispatch_error())
- listener_->OnBadMessageReceived(message);
- #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
- if (logger->Enabled())
- logger->OnPostDispatchMessage(message);
- #endif
- }
- // Called on the listener's thread.
- void ChannelProxy::Context::AddListenerTaskRunner(
- int32_t routing_id,
- scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
- DCHECK(default_listener_task_runner_->BelongsToCurrentThread());
- DCHECK(task_runner);
- base::AutoLock lock(listener_thread_task_runners_lock_);
- if (!base::Contains(listener_thread_task_runners_, routing_id))
- listener_thread_task_runners_.insert({routing_id, std::move(task_runner)});
- }
- // Called on the listener's thread.
- void ChannelProxy::Context::RemoveListenerTaskRunner(int32_t routing_id) {
- DCHECK(default_listener_task_runner_->BelongsToCurrentThread());
- base::AutoLock lock(listener_thread_task_runners_lock_);
- listener_thread_task_runners_.erase(routing_id);
- }
- // Called on the IPC::Channel thread.
- scoped_refptr<base::SingleThreadTaskRunner>
- ChannelProxy::Context::GetTaskRunner(int32_t routing_id) {
- DCHECK(ipc_task_runner_->BelongsToCurrentThread());
- if (routing_id == MSG_ROUTING_NONE)
- return default_listener_task_runner_;
- base::AutoLock lock(listener_thread_task_runners_lock_);
- auto task_runner = listener_thread_task_runners_.find(routing_id);
- if (task_runner == listener_thread_task_runners_.end())
- return default_listener_task_runner_;
- DCHECK(task_runner->second);
- return task_runner->second;
- }
- // Called on the listener's thread
- void ChannelProxy::Context::OnDispatchConnected() {
- if (channel_connected_called_)
- return;
- base::ProcessId peer_pid;
- {
- base::AutoLock l(peer_pid_lock_);
- peer_pid = peer_pid_;
- }
- channel_connected_called_ = true;
- if (listener_)
- listener_->OnChannelConnected(peer_pid);
- }
- // Called on the listener's thread
- void ChannelProxy::Context::OnDispatchError() {
- if (listener_)
- listener_->OnChannelError();
- }
- // Called on the listener's thread
- void ChannelProxy::Context::OnDispatchBadMessage(const Message& message) {
- if (listener_)
- listener_->OnBadMessageReceived(message);
- }
- // Called on the listener's thread
- void ChannelProxy::Context::OnDispatchAssociatedInterfaceRequest(
- const std::string& interface_name,
- mojo::ScopedInterfaceEndpointHandle handle) {
- if (listener_)
- listener_->OnAssociatedInterfaceRequest(interface_name, std::move(handle));
- }
- void ChannelProxy::Context::ClearChannel() {
- base::AutoLock l(channel_lifetime_lock_);
- channel_.reset();
- }
- void ChannelProxy::Context::AddGenericAssociatedInterfaceForIOThread(
- const std::string& name,
- const GenericAssociatedInterfaceFactory& factory) {
- base::AutoLock channel_lock(channel_lifetime_lock_);
- if (!channel_) {
- base::AutoLock filter_lock(pending_filters_lock_);
- pending_io_thread_interfaces_.emplace_back(name, factory);
- return;
- }
- Channel::AssociatedInterfaceSupport* support =
- channel_->GetAssociatedInterfaceSupport();
- if (support)
- support->AddGenericAssociatedInterface(name, factory);
- }
- void ChannelProxy::Context::Send(Message* message) {
- if (quota_checker_)
- quota_checker_->BeforeMessagesEnqueued(1);
- ipc_task_runner()->PostTask(
- FROM_HERE, base::BindOnce(&ChannelProxy::Context::OnSendMessage, this,
- base::WrapUnique(message)));
- }
- //-----------------------------------------------------------------------------
- // static
- std::unique_ptr<ChannelProxy> ChannelProxy::Create(
- const IPC::ChannelHandle& channel_handle,
- Channel::Mode mode,
- Listener* listener,
- const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
- const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner) {
- std::unique_ptr<ChannelProxy> channel(
- new ChannelProxy(listener, ipc_task_runner, listener_task_runner));
- channel->Init(channel_handle, mode, true);
- return channel;
- }
- // static
- std::unique_ptr<ChannelProxy> ChannelProxy::Create(
- std::unique_ptr<ChannelFactory> factory,
- Listener* listener,
- const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
- const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner) {
- std::unique_ptr<ChannelProxy> channel(
- new ChannelProxy(listener, ipc_task_runner, listener_task_runner));
- channel->Init(std::move(factory), true);
- return channel;
- }
- ChannelProxy::ChannelProxy(Context* context)
- : context_(context), did_init_(false) {
- #if defined(ENABLE_IPC_FUZZER)
- outgoing_message_filter_ = nullptr;
- #endif
- }
- ChannelProxy::ChannelProxy(
- Listener* listener,
- const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
- const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner)
- : context_(new Context(listener, ipc_task_runner, listener_task_runner)),
- did_init_(false) {
- #if defined(ENABLE_IPC_FUZZER)
- outgoing_message_filter_ = nullptr;
- #endif
- }
- ChannelProxy::~ChannelProxy() {
- DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
- Close();
- }
- void ChannelProxy::Init(const IPC::ChannelHandle& channel_handle,
- Channel::Mode mode,
- bool create_pipe_now) {
- #if BUILDFLAG(IS_POSIX) || BUILDFLAG(IS_FUCHSIA)
- // When we are creating a server on POSIX, we need its file descriptor
- // to be created immediately so that it can be accessed and passed
- // to other processes. Forcing it to be created immediately avoids
- // race conditions that may otherwise arise.
- if (mode & Channel::MODE_SERVER_FLAG) {
- create_pipe_now = true;
- }
- #endif // BUILDFLAG(IS_POSIX) || BUILDFLAG(IS_FUCHSIA)
- Init(
- ChannelFactory::Create(channel_handle, mode, context_->ipc_task_runner()),
- create_pipe_now);
- }
- void ChannelProxy::Init(std::unique_ptr<ChannelFactory> factory,
- bool create_pipe_now) {
- DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
- DCHECK(!did_init_);
- DCHECK(!context_->quota_checker_);
- context_->quota_checker_ = factory->GetQuotaChecker();
- if (create_pipe_now) {
- // Create the channel immediately. This effectively sets up the
- // low-level pipe so that the client can connect. Without creating
- // the pipe immediately, it is possible for a listener to attempt
- // to connect and get an error since the pipe doesn't exist yet.
- context_->CreateChannel(std::move(factory));
- } else {
- context_->ipc_task_runner()->PostTask(
- FROM_HERE,
- base::BindOnce(&Context::CreateChannel, context_, std::move(factory)));
- }
- // complete initialization on the background thread
- context_->ipc_task_runner()->PostTask(
- FROM_HERE, base::BindOnce(&Context::OnChannelOpened, context_));
- did_init_ = true;
- OnChannelInit();
- }
- void ChannelProxy::Pause() {
- context_->ipc_task_runner()->PostTask(
- FROM_HERE, base::BindOnce(&Context::PauseChannel, context_));
- }
- void ChannelProxy::Unpause(bool flush) {
- context_->ipc_task_runner()->PostTask(
- FROM_HERE, base::BindOnce(&Context::UnpauseChannel, context_, flush));
- }
- void ChannelProxy::Flush() {
- context_->ipc_task_runner()->PostTask(
- FROM_HERE, base::BindOnce(&Context::FlushChannel, context_));
- }
- void ChannelProxy::Close() {
- DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
- // Clear the backpointer to the listener so that any pending calls to
- // Context::OnDispatchMessage or OnDispatchError will be ignored. It is
- // possible that the channel could be closed while it is receiving messages!
- context_->Clear();
- if (context_->ipc_task_runner()) {
- context_->ipc_task_runner()->PostTask(
- FROM_HERE, base::BindOnce(&Context::OnChannelClosed, context_));
- }
- }
- bool ChannelProxy::Send(Message* message) {
- DCHECK(!message->is_sync()) << "Need to use IPC::SyncChannel";
- SendInternal(message);
- return true;
- }
- void ChannelProxy::SendInternal(Message* message) {
- DCHECK(did_init_);
- // TODO(alexeypa): add DCHECK(CalledOnValidThread()) here. Currently there are
- // tests that call Send() from a wrong thread. See http://crbug.com/163523.
- #ifdef ENABLE_IPC_FUZZER
- // In IPC fuzzing builds, it is possible to define a filter to apply to
- // outgoing messages. It will either rewrite the message and return a new
- // one, freeing the original, or return the message unchanged.
- if (outgoing_message_filter())
- message = outgoing_message_filter()->Rewrite(message);
- #endif
- #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
- Logging::GetInstance()->OnSendMessage(message);
- #endif
- context_->Send(message);
- }
- void ChannelProxy::AddFilter(MessageFilter* filter) {
- DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
- context_->AddFilter(filter);
- }
- void ChannelProxy::RemoveFilter(MessageFilter* filter) {
- DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
- context_->ipc_task_runner()->PostTask(
- FROM_HERE, base::BindOnce(&Context::OnRemoveFilter, context_,
- base::RetainedRef(filter)));
- }
- void ChannelProxy::AddGenericAssociatedInterfaceForIOThread(
- const std::string& name,
- const GenericAssociatedInterfaceFactory& factory) {
- context()->AddGenericAssociatedInterfaceForIOThread(name, factory);
- }
- void ChannelProxy::GetRemoteAssociatedInterface(
- mojo::GenericPendingAssociatedReceiver receiver) {
- DCHECK(did_init_);
- context()->thread_safe_channel().GetAssociatedInterface(std::move(receiver));
- }
- void ChannelProxy::ClearIPCTaskRunner() {
- DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
- context()->ClearIPCTaskRunner();
- }
- void ChannelProxy::OnChannelInit() {
- }
- //-----------------------------------------------------------------------------
- } // namespace IPC
|