The Tor Browser: comparison ipc/chromium/src/chrome/common/ipc_sync

--1:000000000000
+:a164a51d80c7
+// Copyright (c) 2006-2008 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 "chrome/common/ipc_sync_channel.h"
+#include "base/lazy_instance.h"
+#include "base/logging.h"
+#include "base/thread_local.h"
+#include "base/message_loop.h"
+#include "base/waitable_event.h"
+#include "base/waitable_event_watcher.h"
+#include "chrome/common/ipc_sync_message.h"
+using base::TimeDelta;
+using base::TimeTicks;
+using base::WaitableEvent;
+namespace IPC {
+// 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;
+}
+~ReceivedSyncMsgQueue() {
+}
+// Called on IPC thread when a synchronous message or reply arrives.
+void QueueMessage(const Message& msg, SyncChannel::SyncContext* context) {
+bool was_task_pending;
+{
+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));
+}
+dispatch_event_.Signal();
+if (!was_task_pending) {
+listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
+this, &ReceivedSyncMsgQueue::DispatchMessagesTask));
+}
+}
+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() {
+{
+AutoLock auto_lock(message_lock_);
+task_pending_ = false;
+}
+DispatchMessages();
+}
+void DispatchMessages() {
+while (true) {
+Message* message;
+scoped_refptr<SyncChannel::SyncContext> context;
+{
+AutoLock auto_lock(message_lock_);
+if (message_queue_.empty())
+break;
+message = message_queue_.front().message;
+context = message_queue_.front().context;
+message_queue_.pop_front();
+}
+context->OnDispatchMessage(*message);
+delete message;
+}
+}
+// SyncChannel calls this in its destructor.
+void RemoveContext(SyncContext* context) {
+AutoLock auto_lock(message_lock_);
+SyncMessageQueue::iterator iter = message_queue_.begin();
+while (iter != message_queue_.end()) {
+if (iter->context == context) {
+delete iter->message;
+iter = message_queue_.erase(iter);
+} else {
+iter++;
+}
+}
+if (--listener_count_ == 0) {
+DCHECK(lazy_tls_ptr_.Pointer()->Get());
+lazy_tls_ptr_.Pointer()->Set(NULL);
+}
+}
+WaitableEvent* dispatch_event() { return &dispatch_event_; }
+MessageLoop* listener_message_loop() { return listener_message_loop_; }
+// Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
+static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> >
+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:
+// See the comment in SyncChannel::SyncChannel for why this event is created
+// as manual reset.
+ReceivedSyncMsgQueue() :
+dispatch_event_(true, false),
+listener_message_loop_(MessageLoop::current()),
+task_pending_(false),
+listener_count_(0) {
+}
+// Holds information about a queued synchronous message or reply.
+struct QueuedMessage {
+QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
+Message* message;
+scoped_refptr<SyncChannel::SyncContext> context;
+};
+typedef std::deque<QueuedMessage> SyncMessageQueue;
+SyncMessageQueue message_queue_;
+std::vector<QueuedMessage> received_replies_;
+// Set when we got a synchronous message that we must respond to as the
+// sender needs its reply before it can reply to our original synchronous
+// message.
+WaitableEvent dispatch_event_;
+MessageLoop* listener_message_loop_;
+Lock message_lock_;
+bool task_pending_;
+int listener_count_;
+};
+base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> >
+SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_(base::LINKER_INITIALIZED);
+SyncChannel::SyncContext::SyncContext(
+Channel::Listener* listener,
+MessageFilter* filter,
+MessageLoop* ipc_thread,
+WaitableEvent* shutdown_event)
+: ChannelProxy::Context(listener, filter, ipc_thread),
+received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
+shutdown_event_(shutdown_event) {
+}
+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 a handle that's set when
+// the reply has arrived.
+void SyncChannel::SyncContext::Push(SyncMessage* sync_msg) {
+// 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.
+PendingSyncMsg pending(SyncMessage::GetMessageId(*sync_msg),
+sync_msg->GetReplyDeserializer(),
+new WaitableEvent(true, false));
+AutoLock auto_lock(deserializers_lock_);
+deserializers_.push_back(pending);
+}
+bool SyncChannel::SyncContext::Pop() {
+bool result;
+{
+AutoLock auto_lock(deserializers_lock_);
+PendingSyncMsg msg = deserializers_.back();
+delete msg.deserializer;
+delete msg.done_event;
+msg.done_event = NULL;
+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_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
+received_sync_msgs_.get(), &ReceivedSyncMsgQueue::DispatchReplies));
+return result;
+}
+WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
+AutoLock auto_lock(deserializers_lock_);
+return deserializers_.back().done_event;
+}
+WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
+return received_sync_msgs_->dispatch_event();
+}
+void SyncChannel::SyncContext::DispatchMessages() {
+received_sync_msgs_->DispatchMessages();
+}
+bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
+AutoLock auto_lock(deserializers_lock_);
+if (deserializers_.empty() ||
+!SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
+return false;
+}
+if (!msg->is_reply_error()) {
+deserializers_.back().send_result = deserializers_.back().deserializer->
+SerializeOutputParameters(*msg);
+}
+deserializers_.back().done_event->Signal();
+return true;
+}
+void SyncChannel::SyncContext::Clear() {
+CancelPendingSends();
+received_sync_msgs_->RemoveContext(this);
+Context::Clear();
+}
+void SyncChannel::SyncContext::OnMessageReceived(const Message& msg) {
+// Give the filters a chance at processing this message.
+if (TryFilters(msg))
+return;
+if (TryToUnblockListener(&msg))
+return;
+if (msg.should_unblock()) {
+received_sync_msgs_->QueueMessage(msg, this);
+return;
+}
+if (msg.is_reply()) {
+received_sync_msgs_->QueueReply(msg, this);
+return;
+}
+return Context::OnMessageReceivedNoFilter(msg);
+}
+void SyncChannel::SyncContext::OnChannelError() {
+CancelPendingSends();
+shutdown_watcher_.StopWatching();
+Context::OnChannelError();
+}
+void SyncChannel::SyncContext::OnChannelOpened() {
+shutdown_watcher_.StartWatching(shutdown_event_, this);
+Context::OnChannelOpened();
+}
+void SyncChannel::SyncContext::OnChannelClosed() {
+shutdown_watcher_.StopWatching();
+Context::OnChannelClosed();
+}
+void SyncChannel::SyncContext::OnSendTimeout(int message_id) {
+AutoLock auto_lock(deserializers_lock_);
+PendingSyncMessageQueue::iterator iter;
+for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
+if (iter->id == message_id) {
+iter->done_event->Signal();
+break;
+}
+}
+}
+void SyncChannel::SyncContext::CancelPendingSends() {
+AutoLock auto_lock(deserializers_lock_);
+PendingSyncMessageQueue::iterator iter;
+for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++)
+iter->done_event->Signal();
+}
+void SyncChannel::SyncContext::OnWaitableEventSignaled(WaitableEvent* event) {
+DCHECK(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();
+}
+SyncChannel::SyncChannel(
+const std::wstring& channel_id, Channel::Mode mode,
+Channel::Listener* listener, MessageFilter* filter,
+MessageLoop* ipc_message_loop, bool create_pipe_now,
+WaitableEvent* shutdown_event)
+: ChannelProxy(
+channel_id, mode, ipc_message_loop,
+new SyncContext(listener, filter, ipc_message_loop, shutdown_event),
+create_pipe_now),
+sync_messages_with_no_timeout_allowed_(true) {
+// Ideally we only want to watch this object when running a nested message
+// loop.  However, we don't know when it exits if there's another nested
+// message loop running under it or not, so we wouldn't know whether to
+// stop or keep watching.  So we always watch it, and create the event as
+// manual reset since the object watcher might otherwise reset the event
+// when we're doing a WaitMany.
+dispatch_watcher_.StartWatching(sync_context()->GetDispatchEvent(), this);
+}
+SyncChannel::~SyncChannel() {
+}
+bool SyncChannel::Send(Message* message) {
+return SendWithTimeout(message, base::kNoTimeout);
+}
+bool SyncChannel::SendWithTimeout(Message* message, int timeout_ms) {
+if (!message->is_sync()) {
+ChannelProxy::Send(message);
+return true;
+}
+// *this* might get deleted in WaitForReply.
+scoped_refptr<SyncContext> context(sync_context());
+if (context->shutdown_event()->IsSignaled()) {
+delete message;
+return false;
+}
+DCHECK(sync_messages_with_no_timeout_allowed_ ||
+timeout_ms != base::kNoTimeout);
+SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
+context->Push(sync_msg);
+int message_id = SyncMessage::GetMessageId(*sync_msg);
+WaitableEvent* pump_messages_event = sync_msg->pump_messages_event();
+ChannelProxy::Send(message);
+if (timeout_ms != base::kNoTimeout) {
+// We use the sync message id so that when a message times out, we don't
+// confuse it with another send that is either above/below this Send in
+// the call stack.
+context->ipc_message_loop()->PostDelayedTask(FROM_HERE,
+NewRunnableMethod(context.get(),
+&SyncContext::OnSendTimeout, message_id), timeout_ms);
+}
+// Wait for reply, or for any other incoming synchronous messages.
+WaitForReply(pump_messages_event);
+return context->Pop();
+}
+void SyncChannel::WaitForReply(WaitableEvent* pump_messages_event) {
+while (true) {
+WaitableEvent* objects[] = {
+sync_context()->GetDispatchEvent(),
+sync_context()->GetSendDoneEvent(),
+pump_messages_event
+};
+unsigned count = pump_messages_event ? 3: 2;
+unsigned result = WaitableEvent::WaitMany(objects, count);
+if (result == 0 /* dispatch event */) {
+// We're waiting for a reply, but we received a blocking synchronous
+// call.  We must process it or otherwise a deadlock might occur.
+sync_context()->GetDispatchEvent()->Reset();
+sync_context()->DispatchMessages();
+continue;
+}
+if (result == 2 /* pump_messages_event */)
+WaitForReplyWithNestedMessageLoop();  // Start a nested message loop.
+break;
+}
+}
+void SyncChannel::WaitForReplyWithNestedMessageLoop() {
+WaitableEvent* old_done_event = send_done_watcher_.GetWatchedEvent();
+send_done_watcher_.StopWatching();
+send_done_watcher_.StartWatching(sync_context()->GetSendDoneEvent(), this);
+bool old_state = MessageLoop::current()->NestableTasksAllowed();
+MessageLoop::current()->SetNestableTasksAllowed(true);
+MessageLoop::current()->Run();
+MessageLoop::current()->SetNestableTasksAllowed(old_state);
+if (old_done_event)
+send_done_watcher_.StartWatching(old_done_event, this);
+}
+void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) {
+WaitableEvent* dispatch_event = sync_context()->GetDispatchEvent();
+if (event == dispatch_event) {
+// The call to DispatchMessages might delete this object, so reregister
+// the object watcher first.
+dispatch_event->Reset();
+dispatch_watcher_.StartWatching(dispatch_event, this);
+sync_context()->DispatchMessages();
+} else {
+// We got the reply, timed out or the process shutdown.
+DCHECK(event == sync_context()->GetSendDoneEvent());
+MessageLoop::current()->Quit();
+}
+}
+}  // namespace IPC

The Tor Browser / file comparison

comparison: ipc/chromium/src/chrome/common/ipc_sync_channel.cc

ipc/chromium/src/chrome/common/ipc_sync_channel.cc