[go: nahoru, domu]

blob: cc4fa1eb5d9ae129fdda5be456eef7f28121644b [file] [log] [blame]
// Copyright 2012 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "ipc/ipc_sync_channel.h"
#include <stddef.h>
#include <stdint.h>
#include <utility>
#include "base/functional/bind.h"
#include "base/lazy_instance.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/memory/ptr_util.h"
#include "base/memory/raw_ptr.h"
#include "base/run_loop.h"
#include "base/synchronization/waitable_event.h"
#include "base/task/sequenced_task_runner.h"
#include "base/task/single_thread_task_runner.h"
#include "base/threading/thread_local.h"
#include "base/trace_event/trace_event.h"
#include "build/build_config.h"
#include "ipc/ipc_channel_factory.h"
#include "ipc/ipc_logging.h"
#include "ipc/ipc_message_macros.h"
#include "ipc/ipc_sync_message.h"
#include "mojo/public/cpp/bindings/sync_event_watcher.h"
#if !BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
#include "ipc/trace_ipc_message.h"
#endif
using base::WaitableEvent;
namespace IPC {
namespace {
// A generic callback used when watching handles synchronously. Sets |*signal|
// to true.
void OnEventReady(bool* signal) {
*signal = true;
}
} // namespace
// When we're blocked in a Send(), we need to process incoming synchronous
// messages right away because it could be blocking our reply (either
// directly from the same object we're calling, or indirectly through one or
// more other channels). That means that in SyncContext's OnMessageReceived,
// we need to process sync message right away if we're blocked. However a
// simple check isn't sufficient, because the listener thread can be in the
// process of calling Send.
// To work around this, when SyncChannel filters a sync message, it sets
// an event that the listener thread waits on during its Send() call. This
// allows us to dispatch incoming sync messages when blocked. The race
// condition is handled because if Send is in the process of being called, it
// will check the event. In case the listener thread isn't sending a message,
// we queue a task on the listener thread to dispatch the received messages.
// The messages are stored in this queue object that's shared among all
// SyncChannel objects on the same thread (since one object can receive a
// sync message while another one is blocked).
class SyncChannel::ReceivedSyncMsgQueue :
public base::RefCountedThreadSafe<ReceivedSyncMsgQueue> {
public:
// Returns the ReceivedSyncMsgQueue instance for this thread, creating one
// if necessary. Call RemoveContext on the same thread when done.
static ReceivedSyncMsgQueue* AddContext() {
// We want one ReceivedSyncMsgQueue per listener thread (i.e. since multiple
// SyncChannel objects can block the same thread).
ReceivedSyncMsgQueue* rv = lazy_tls_ptr_.Pointer()->Get();
if (!rv) {
rv = new ReceivedSyncMsgQueue();
ReceivedSyncMsgQueue::lazy_tls_ptr_.Pointer()->Set(rv);
}
rv->listener_count_++;
return rv;
}
// Prevents messages from being dispatched immediately when the dispatch event
// is signaled. Instead, |*dispatch_flag| will be set.
void BlockDispatch(bool* dispatch_flag) { dispatch_flag_ = dispatch_flag; }
// Allows messages to be dispatched immediately when the dispatch event is
// signaled.
void UnblockDispatch() { dispatch_flag_ = nullptr; }
// Called on IPC thread when a synchronous message or reply arrives.
void QueueMessage(const Message& msg, SyncChannel::SyncContext* context) {
bool was_task_pending;
{
base::AutoLock auto_lock(message_lock_);
was_task_pending = task_pending_;
task_pending_ = true;
// We set the event in case the listener thread is blocked (or is about
// to). In case it's not, the PostTask dispatches the messages.
message_queue_.push_back(QueuedMessage(new Message(msg), context));
message_queue_version_++;
}
dispatch_event_.Signal();
if (!was_task_pending) {
listener_task_runner_->PostTask(
FROM_HERE, base::BindOnce(&ReceivedSyncMsgQueue::DispatchMessagesTask,
this, base::RetainedRef(context)));
}
}
void QueueReply(const Message &msg, SyncChannel::SyncContext* context) {
received_replies_.push_back(QueuedMessage(new Message(msg), context));
}
// Called on the listener's thread to process any queues synchronous
// messages.
void DispatchMessagesTask(SyncContext* context) {
{
base::AutoLock auto_lock(message_lock_);
task_pending_ = false;
}
context->DispatchMessages();
}
// Dispatches any queued incoming sync messages. If |dispatching_context| is
// not null, messages which target a restricted dispatch channel will only be
// dispatched if |dispatching_context| belongs to the same restricted dispatch
// group as that channel. If |dispatching_context| is null, all queued
// messages are dispatched.
void DispatchMessages(SyncContext* dispatching_context) {
bool first_time = true;
uint32_t expected_version = 0;
SyncMessageQueue::iterator it;
while (true) {
Message* message = nullptr;
scoped_refptr<SyncChannel::SyncContext> context;
{
base::AutoLock auto_lock(message_lock_);
if (first_time || message_queue_version_ != expected_version) {
it = message_queue_.begin();
first_time = false;
}
for (; it != message_queue_.end(); it++) {
int message_group = it->context->restrict_dispatch_group();
if (message_group == kRestrictDispatchGroup_None ||
(dispatching_context &&
message_group ==
dispatching_context->restrict_dispatch_group())) {
message = it->message;
context = it->context;
it = message_queue_.erase(it);
message_queue_version_++;
expected_version = message_queue_version_;
break;
}
}
}
if (message == nullptr)
break;
context->OnDispatchMessage(*message);
delete message;
}
}
// SyncChannel calls this in its destructor.
void RemoveContext(SyncContext* context) {
base::AutoLock auto_lock(message_lock_);
SyncMessageQueue::iterator iter = message_queue_.begin();
while (iter != message_queue_.end()) {
if (iter->context.get() == context) {
delete iter->message;
iter = message_queue_.erase(iter);
message_queue_version_++;
} else {
iter++;
}
}
if (--listener_count_ == 0) {
DCHECK(lazy_tls_ptr_.Pointer()->Get());
lazy_tls_ptr_.Pointer()->Set(nullptr);
sync_dispatch_watcher_.reset();
}
}
base::WaitableEvent* dispatch_event() { return &dispatch_event_; }
base::SingleThreadTaskRunner* listener_task_runner() {
return listener_task_runner_.get();
}
// Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue>>::
DestructorAtExit lazy_tls_ptr_;
// Called on the ipc thread to check if we can unblock any current Send()
// calls based on a queued reply.
void DispatchReplies() {
for (size_t i = 0; i < received_replies_.size(); ++i) {
Message* message = received_replies_[i].message;
if (received_replies_[i].context->TryToUnblockListener(message)) {
delete message;
received_replies_.erase(received_replies_.begin() + i);
return;
}
}
}
private:
friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>;
// See the comment in SyncChannel::SyncChannel for why this event is created
// as manual reset.
ReceivedSyncMsgQueue()
: message_queue_version_(0),
dispatch_event_(base::WaitableEvent::ResetPolicy::MANUAL,
base::WaitableEvent::InitialState::NOT_SIGNALED),
listener_task_runner_(
base::SingleThreadTaskRunner::GetCurrentDefault()),
sync_dispatch_watcher_(std::make_unique<mojo::SyncEventWatcher>(
&dispatch_event_,
base::BindRepeating(&ReceivedSyncMsgQueue::OnDispatchEventReady,
base::Unretained(this)))) {
sync_dispatch_watcher_->AllowWokenUpBySyncWatchOnSameThread();
}
~ReceivedSyncMsgQueue() = default;
void OnDispatchEventReady() {
if (dispatch_flag_) {
*dispatch_flag_ = true;
return;
}
// We were woken up during a sync wait, but no specific SyncChannel is
// currently waiting. i.e., some other Mojo interface on this thread is
// waiting for a response. Since we don't support anything analogous to
// restricted dispatch on Mojo interfaces, in this case it's safe to
// dispatch sync messages for any context.
DispatchMessages(nullptr);
}
// Holds information about a queued synchronous message or reply.
struct QueuedMessage {
QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
raw_ptr<Message> message;
scoped_refptr<SyncChannel::SyncContext> context;
};
typedef std::list<QueuedMessage> SyncMessageQueue;
SyncMessageQueue message_queue_;
// Used to signal DispatchMessages to rescan
uint32_t message_queue_version_ = 0;
std::vector<QueuedMessage> received_replies_;
// Signaled when we get a synchronous message that we must respond to, as the
// sender needs its reply before it can reply to our original synchronous
// message.
base::WaitableEvent dispatch_event_;
scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
base::Lock message_lock_;
bool task_pending_ = false;
int listener_count_ = 0;
// If not null, the address of a flag to set when the dispatch event signals,
// in lieu of actually dispatching messages. This is used by
// SyncChannel::WaitForReply to restrict the scope of queued messages we're
// allowed to process while it's waiting.
raw_ptr<bool> dispatch_flag_ = nullptr;
// Watches |dispatch_event_| during all sync handle watches on this thread.
std::unique_ptr<mojo::SyncEventWatcher> sync_dispatch_watcher_;
};
base::LazyInstance<base::ThreadLocalPointer<
SyncChannel::ReceivedSyncMsgQueue>>::DestructorAtExit
SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_ =
LAZY_INSTANCE_INITIALIZER;
SyncChannel::SyncContext::SyncContext(
Listener* listener,
const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
WaitableEvent* shutdown_event)
: ChannelProxy::Context(listener, ipc_task_runner, listener_task_runner),
received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
shutdown_event_(shutdown_event),
restrict_dispatch_group_(kRestrictDispatchGroup_None) {}
void SyncChannel::SyncContext::OnSendDoneEventSignaled(
base::RunLoop* nested_loop,
base::WaitableEvent* event) {
DCHECK_EQ(GetSendDoneEvent(), event);
nested_loop->Quit();
}
SyncChannel::SyncContext::~SyncContext() {
while (!deserializers_.empty())
Pop();
}
// Adds information about an outgoing sync message to the context so that
// we know how to deserialize the reply. Returns |true| if the message was added
// to the context or |false| if it was rejected (e.g. due to shutdown.)
bool SyncChannel::SyncContext::Push(SyncMessage* sync_msg) {
// Create the tracking information for this message. This object is stored
// by value since all members are pointers that are cheap to copy. These
// pointers are cleaned up in the Pop() function.
//
// The event is created as manual reset because in between Signal and
// OnObjectSignalled, another Send can happen which would stop the watcher
// from being called. The event would get watched later, when the nested
// Send completes, so the event will need to remain set.
base::AutoLock auto_lock(deserializers_lock_);
if (reject_new_deserializers_)
return false;
PendingSyncMsg pending(
SyncMessage::GetMessageId(*sync_msg), sync_msg->GetReplyDeserializer(),
new base::WaitableEvent(base::WaitableEvent::ResetPolicy::MANUAL,
base::WaitableEvent::InitialState::NOT_SIGNALED));
deserializers_.push_back(pending);
return true;
}
bool SyncChannel::SyncContext::Pop() {
bool result;
{
base::AutoLock auto_lock(deserializers_lock_);
PendingSyncMsg msg = deserializers_.back();
delete msg.deserializer;
delete msg.done_event;
msg.done_event = nullptr;
deserializers_.pop_back();
result = msg.send_result;
}
// We got a reply to a synchronous Send() call that's blocking the listener
// thread. However, further down the call stack there could be another
// blocking Send() call, whose reply we received after we made this last
// Send() call. So check if we have any queued replies available that
// can now unblock the listener thread.
ipc_task_runner()->PostTask(
FROM_HERE, base::BindOnce(&ReceivedSyncMsgQueue::DispatchReplies,
received_sync_msgs_));
return result;
}
base::WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
base::AutoLock auto_lock(deserializers_lock_);
return deserializers_.back().done_event;
}
base::WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
return received_sync_msgs_->dispatch_event();
}
void SyncChannel::SyncContext::DispatchMessages() {
received_sync_msgs_->DispatchMessages(this);
}
bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
base::AutoLock auto_lock(deserializers_lock_);
if (deserializers_.empty() ||
!SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
return false;
}
if (!msg->is_reply_error()) {
bool send_result = deserializers_.back().deserializer->
SerializeOutputParameters(*msg);
deserializers_.back().send_result = send_result;
DVLOG_IF(1, !send_result) << "Couldn't deserialize reply message";
} else {
DVLOG(1) << "Received error reply";
}
base::WaitableEvent* done_event = deserializers_.back().done_event;
TRACE_EVENT_WITH_FLOW0("toplevel.flow",
"SyncChannel::SyncContext::TryToUnblockListener",
done_event, TRACE_EVENT_FLAG_FLOW_OUT);
done_event->Signal();
return true;
}
void SyncChannel::SyncContext::Clear() {
CancelPendingSends();
received_sync_msgs_->RemoveContext(this);
Context::Clear();
}
bool SyncChannel::SyncContext::OnMessageReceived(const Message& msg) {
// Give the filters a chance at processing this message.
if (TryFilters(msg))
return true;
if (TryToUnblockListener(&msg))
return true;
if (msg.is_reply()) {
received_sync_msgs_->QueueReply(msg, this);
return true;
}
if (msg.should_unblock()) {
received_sync_msgs_->QueueMessage(msg, this);
return true;
}
return Context::OnMessageReceivedNoFilter(msg);
}
void SyncChannel::SyncContext::OnChannelError() {
CancelPendingSends();
shutdown_watcher_.StopWatching();
Context::OnChannelError();
}
void SyncChannel::SyncContext::OnChannelOpened() {
if (shutdown_event_) {
shutdown_watcher_.StartWatching(
shutdown_event_,
base::BindOnce(&SyncChannel::SyncContext::OnShutdownEventSignaled,
base::Unretained(this)),
base::SequencedTaskRunner::GetCurrentDefault());
}
Context::OnChannelOpened();
}
void SyncChannel::SyncContext::OnChannelClosed() {
CancelPendingSends();
shutdown_watcher_.StopWatching();
Context::OnChannelClosed();
}
void SyncChannel::SyncContext::CancelPendingSends() {
base::AutoLock auto_lock(deserializers_lock_);
reject_new_deserializers_ = true;
PendingSyncMessageQueue::iterator iter;
DVLOG(1) << "Canceling pending sends";
for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
TRACE_EVENT_WITH_FLOW0("toplevel.flow",
"SyncChannel::SyncContext::CancelPendingSends",
iter->done_event, TRACE_EVENT_FLAG_FLOW_OUT);
iter->done_event->Signal();
}
}
void SyncChannel::SyncContext::OnShutdownEventSignaled(WaitableEvent* event) {
DCHECK_EQ(event, shutdown_event_);
// Process shut down before we can get a reply to a synchronous message.
// Cancel pending Send calls, which will end up setting the send done event.
CancelPendingSends();
}
// static
std::unique_ptr<SyncChannel> SyncChannel::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,
bool create_pipe_now,
base::WaitableEvent* shutdown_event) {
// TODO(tobiasjs): The shutdown_event object is passed to a refcounted
// Context object, and as a result it is not easy to ensure that it
// outlives the Context. There should be some way to either reset
// the shutdown_event when it is destroyed, or allow the Context to
// control the lifetime of shutdown_event.
std::unique_ptr<SyncChannel> channel =
Create(listener, ipc_task_runner, listener_task_runner, shutdown_event);
channel->Init(channel_handle, mode, create_pipe_now);
return channel;
}
// static
std::unique_ptr<SyncChannel> SyncChannel::Create(
Listener* listener,
const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
WaitableEvent* shutdown_event) {
return base::WrapUnique(new SyncChannel(
listener, ipc_task_runner, listener_task_runner, shutdown_event));
}
SyncChannel::SyncChannel(
Listener* listener,
const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
WaitableEvent* shutdown_event)
: ChannelProxy(new SyncContext(listener,
ipc_task_runner,
listener_task_runner,
shutdown_event)),
sync_handle_registry_(mojo::SyncHandleRegistry::current()) {
// The current (listener) thread must be distinct from the IPC thread, or else
// sending synchronous messages will deadlock.
DCHECK_NE(ipc_task_runner.get(),
base::SingleThreadTaskRunner::GetCurrentDefault().get());
StartWatching();
}
void SyncChannel::AddListenerTaskRunner(
int32_t routing_id,
scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
context()->AddListenerTaskRunner(routing_id, std::move(task_runner));
}
void SyncChannel::RemoveListenerTaskRunner(int32_t routing_id) {
context()->RemoveListenerTaskRunner(routing_id);
}
SyncChannel::~SyncChannel() = default;
void SyncChannel::SetRestrictDispatchChannelGroup(int group) {
sync_context()->set_restrict_dispatch_group(group);
}
scoped_refptr<SyncMessageFilter> SyncChannel::CreateSyncMessageFilter() {
scoped_refptr<SyncMessageFilter> filter = new SyncMessageFilter(
sync_context()->shutdown_event());
AddFilter(filter.get());
if (!did_init())
pre_init_sync_message_filters_.push_back(filter);
return filter;
}
bool SyncChannel::Send(Message* message) {
#if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
std::string name;
Logging::GetInstance()->GetMessageText(
message->type(), &name, message, nullptr);
TRACE_EVENT1("ipc", "SyncChannel::Send", "name", name);
#else
TRACE_IPC_MESSAGE_SEND("ipc", "SyncChannel::Send", message);
#endif
if (!message->is_sync()) {
ChannelProxy::SendInternal(message);
return true;
}
SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
// *this* might get deleted in WaitForReply.
scoped_refptr<SyncContext> context(sync_context());
if (!context->Push(sync_msg)) {
DVLOG(1) << "Channel is shutting down. Dropping sync message.";
delete message;
return false;
}
ChannelProxy::SendInternal(message);
// Wait for reply, or for any other incoming synchronous messages.
// |this| might get deleted, so only call static functions at this point.
scoped_refptr<mojo::SyncHandleRegistry> registry = sync_handle_registry_;
WaitForReply(registry.get(), context.get());
TRACE_EVENT_WITH_FLOW0("toplevel.flow", "SyncChannel::Send",
context->GetSendDoneEvent(), TRACE_EVENT_FLAG_FLOW_IN);
return context->Pop();
}
void SyncChannel::WaitForReply(mojo::SyncHandleRegistry* registry,
SyncContext* context) {
context->DispatchMessages();
while (true) {
bool dispatch = false;
{
bool send_done = false;
mojo::SyncHandleRegistry::EventCallbackSubscription
send_done_subscription = registry->RegisterEvent(
context->GetSendDoneEvent(),
base::BindRepeating(&OnEventReady, &send_done));
const bool* stop_flags[] = {&dispatch, &send_done};
context->received_sync_msgs()->BlockDispatch(&dispatch);
registry->Wait(stop_flags, 2);
context->received_sync_msgs()->UnblockDispatch();
}
if (dispatch) {
// We're waiting for a reply, but we received a blocking synchronous call.
// We must process it to avoid potential deadlocks.
context->GetDispatchEvent()->Reset();
context->DispatchMessages();
continue;
}
break;
}
}
void SyncChannel::OnDispatchEventSignaled(base::WaitableEvent* event) {
DCHECK_EQ(sync_context()->GetDispatchEvent(), event);
sync_context()->GetDispatchEvent()->Reset();
StartWatching();
// NOTE: May delete |this|.
sync_context()->DispatchMessages();
}
void SyncChannel::StartWatching() {
// |dispatch_watcher_| watches the event asynchronously, only dispatching
// messages once the listener thread is unblocked and pumping its task queue.
// The ReceivedSyncMsgQueue also watches this event and may dispatch
// immediately if woken up by a message which it's allowed to dispatch.
dispatch_watcher_.StartWatching(
sync_context()->GetDispatchEvent(),
base::BindOnce(&SyncChannel::OnDispatchEventSignaled,
base::Unretained(this)),
sync_context()->listener_task_runner());
}
void SyncChannel::OnChannelInit() {
pre_init_sync_message_filters_.clear();
}
} // namespace IPC