Add sync context dispatch restriction

ipc/ipc_sync_channel.cc

-// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
+// Copyright (c) 2011 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_sync_channel.h"
 
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/threading/thread_local.h"
 #include "base/synchronization/waitable_event.h"
@@ skipping 49 matching lines @@
       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));
+          this,
+          &ReceivedSyncMsgQueue::DispatchMessagesTask,
+          scoped_refptr<SyncContext>(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() {
+  void DispatchMessagesTask(SyncContext* context) {
     {
       base::AutoLock auto_lock(message_lock_);
       task_pending_ = false;
     }
-    DispatchMessages();
+    context->DispatchMessages();
   }
 
-  void DispatchMessages() {
+  void DispatchMessages(SyncContext* dispatching_context) {
+    SyncMessageQueue delayed_queue;
     while (true) {
       Message* message;
       scoped_refptr<SyncChannel::SyncContext> context;
       {
         base::AutoLock auto_lock(message_lock_);
-        if (message_queue_.empty())
+        if (message_queue_.empty()) {
+          message_queue_ = delayed_queue;
           break;
+        }
 
         message = message_queue_.front().message;
         context = message_queue_.front().context;
         message_queue_.pop_front();
       }
-
-      context->OnDispatchMessage(*message);
-      delete message;
+      if (context->restrict_dispatch() && context != dispatching_context) {
+        delayed_queue.push_back(QueuedMessage(message, context));
+      } else {
+        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 == context) {
@@ skipping 82 matching lines @@
 
 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> >
     SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_(base::LINKER_INITIALIZED);
 
 SyncChannel::SyncContext::SyncContext(
     Channel::Listener* listener,
     MessageLoop* ipc_thread,
     WaitableEvent* shutdown_event)
     : ChannelProxy::Context(listener, ipc_thread),
       received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
-      shutdown_event_(shutdown_event) {
+      shutdown_event_(shutdown_event),
+      restrict_dispatch_(false) {
 }
 
 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.
@@ skipping 35 matching lines @@
 WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
   base::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();
+  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()) {
@@ skipping 97 matching lines @@
   // 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() {
 }
 
+void SyncChannel::SetRestrictDispatchToSameChannel(bool value) {
+  sync_context()->set_restrict_dispatch(value);
+}
+
 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;
   }
 
@@ skipping 24 matching lines @@
 
   // Wait for reply, or for any other incoming synchronous messages.
   // *this* might get deleted, so only call static functions at this point.
   WaitForReply(context, pump_messages_event);
 
   return context->Pop();
 }
 
 void SyncChannel::WaitForReply(
     SyncContext* context, WaitableEvent* pump_messages_event) {
+  context->DispatchMessages();
   while (true) {
     WaitableEvent* objects[] = {
       context->GetDispatchEvent(),
       context->GetSendDoneEvent(),
       pump_messages_event
     };
 
     unsigned count = pump_messages_event ? 3: 2;
     size_t result = WaitableEvent::WaitMany(objects, count);
     if (result == 0 /* dispatch event */) {
@@ skipping 50 matching lines @@
 void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) {
   DCHECK(event == sync_context()->GetDispatchEvent());
   // The call to DispatchMessages might delete this object, so reregister
   // the object watcher first.
   event->Reset();
   dispatch_watcher_.StartWatching(event, this);
   sync_context()->DispatchMessages();
 }
 
 }  // namespace IPC