Add sync context dispatch restriction.

This adds a way to restrict on a per-channel basis that incoming messages may only be dispatched when that particular channel is sending a sync message (or in a message loop).

It does so to the PPAPI channels, which may not introduce a sync dependency circle.

BUG=chromiumos:13821
TEST=news.google.com with Pepper Flash (see bug)

Review URL: http://codereview.chromium.org/6810013

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@80892 0039d316-1c4b-4281-b951-d872f2087c98

4 files changed, 194 insertions, 12 deletions

diff --git a/ipc/ipc_sync_channel.cc b/ipc/ipc_sync_channel.cc
index 1c7edfa..3dcab2c 100644
--- a/ipc/ipc_sync_channel.cc
+++ b/ipc/ipc_sync_channel.cc
@@ -1,4 +1,4 @@
-// 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.
 
@@ -68,7 +68,9 @@ class SyncChannel::ReceivedSyncMsgQueue :
     dispatch_event_.Signal();
     if (!was_task_pending) {
       listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
-          this, &ReceivedSyncMsgQueue::DispatchMessagesTask));
+          this,
+          &ReceivedSyncMsgQueue::DispatchMessagesTask,
+          scoped_refptr<SyncContext>(context)));
     }
   }
 
@@ -78,30 +80,36 @@ class SyncChannel::ReceivedSyncMsgQueue :
 
   // 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;
+      }
     }
   }
 
@@ -204,7 +212,8 @@ SyncChannel::SyncContext::SyncContext(
     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() {
@@ -260,7 +269,7 @@ WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
 }
 
 void SyncChannel::SyncContext::DispatchMessages() {
-  received_sync_msgs_->DispatchMessages();
+  received_sync_msgs_->DispatchMessages(this);
 }
 
 bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
@@ -378,6 +387,10 @@ SyncChannel::SyncChannel(
 SyncChannel::~SyncChannel() {
 }
 
+void SyncChannel::SetRestrictDispatchToSameChannel(bool value) {
+  sync_context()->set_restrict_dispatch(value);
+}
+
 bool SyncChannel::Send(Message* message) {
   return SendWithTimeout(message, base::kNoTimeout);
 }
@@ -422,6 +435,7 @@ bool SyncChannel::SendWithTimeout(Message* message, int timeout_ms) {
 
 void SyncChannel::WaitForReply(
     SyncContext* context, WaitableEvent* pump_messages_event) {
+  context->DispatchMessages();
   while (true) {
     WaitableEvent* objects[] = {
       context->GetDispatchEvent(),
diff --git a/ipc/ipc_sync_channel.h b/ipc/ipc_sync_channel.h
index 0386b8f..e7ff051 100644
--- a/ipc/ipc_sync_channel.h
+++ b/ipc/ipc_sync_channel.h
@@ -51,6 +51,17 @@ class SyncChannel : public ChannelProxy,
     sync_messages_with_no_timeout_allowed_ = value;
   }
 
+  // Sets this channel to only dispatch its incoming unblocking messages when it
+  // is itself blocked on sending a sync message, not when other channels are.
+  //
+  // Normally, any unblocking message coming from any channel can be dispatched
+  // when any (possibly other) channel is blocked on sending a message. This is
+  // needed in some cases to unblock certain loops (e.g. necessary when some
+  // processes share a window hierarchy), but may cause re-entrancy issues in
+  // some cases where such loops are not possible. This flags allows the tagging
+  // of some particular channels to not re-enter in such cases.
+  void SetRestrictDispatchToSameChannel(bool value);
+
  protected:
   class ReceivedSyncMsgQueue;
   friend class ReceivedSyncMsgQueue;
@@ -98,6 +109,9 @@ class SyncChannel : public ChannelProxy,
       return received_sync_msgs_;
     }
 
+    void set_restrict_dispatch(bool value) { restrict_dispatch_ = value; }
+    bool restrict_dispatch() const { return restrict_dispatch_; }
+
    private:
     ~SyncContext();
     // ChannelProxy methods that we override.
@@ -125,6 +139,7 @@ class SyncChannel : public ChannelProxy,
 
     base::WaitableEvent* shutdown_event_;
     base::WaitableEventWatcher shutdown_watcher_;
+    bool restrict_dispatch_;
   };
 
  private:
diff --git a/ipc/ipc_sync_channel_unittest.cc b/ipc/ipc_sync_channel_unittest.cc
index 8e207bf..6952aac 100644
--- a/ipc/ipc_sync_channel_unittest.cc
+++ b/ipc/ipc_sync_channel_unittest.cc
@@ -158,6 +158,8 @@ class Worker : public Channel::Listener, public Message::Sender {
     return overrided_thread_ ? overrided_thread_ : &listener_thread_;
   }
 
+  const base::Thread& ipc_thread() const { return ipc_thread_; }
+
  private:
   // Called on the listener thread to create the sync channel.
   void OnStart() {
@@ -1164,4 +1166,152 @@ TEST_F(IPCSyncChannelTest, SendAfterClose) {
   EXPECT_FALSE(server.send_result());
 }
 
+//-----------------------------------------------------------------------------
+
+namespace {
+
+class RestrictedDispatchServer : public Worker {
+ public:
+  RestrictedDispatchServer(WaitableEvent* sent_ping_event)
+      : Worker("restricted_channel", Channel::MODE_SERVER),
+        sent_ping_event_(sent_ping_event) { }
+
+  void OnDoPing(int ping) {
+    // Send an asynchronous message that unblocks the caller.
+    IPC::Message* msg = new SyncChannelTestMsg_Ping(ping);
+    msg->set_unblock(true);
+    Send(msg);
+    // Signal the event after the message has been sent on the channel, on the
+    // IPC thread.
+    ipc_thread().message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &RestrictedDispatchServer::OnPingSent));
+  }
+
+  base::Thread* ListenerThread() { return Worker::ListenerThread(); }
+
+ private:
+  bool OnMessageReceived(const Message& message) {
+    IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchServer, message)
+     IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
+     IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
+    IPC_END_MESSAGE_MAP()
+    return true;
+  }
+
+  void OnPingSent() {
+    sent_ping_event_->Signal();
+  }
+
+  void OnNoArgs() { }
+  WaitableEvent* sent_ping_event_;
+};
+
+class NonRestrictedDispatchServer : public Worker {
+ public:
+  NonRestrictedDispatchServer()
+      : Worker("non_restricted_channel", Channel::MODE_SERVER) {}
+
+ private:
+  bool OnMessageReceived(const Message& message) {
+    IPC_BEGIN_MESSAGE_MAP(NonRestrictedDispatchServer, message)
+     IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
+     IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
+    IPC_END_MESSAGE_MAP()
+    return true;
+  }
 
+  void OnNoArgs() { }
+};
+
+class RestrictedDispatchClient : public Worker {
+ public:
+  RestrictedDispatchClient(WaitableEvent* sent_ping_event,
+                           RestrictedDispatchServer* server,
+                           int* success)
+      : Worker("restricted_channel", Channel::MODE_CLIENT),
+        ping_(0),
+        server_(server),
+        success_(success),
+        sent_ping_event_(sent_ping_event) {}
+
+  void Run() {
+    // Incoming messages from our channel should only be dispatched when we
+    // send a message on that same channel.
+    channel()->SetRestrictDispatchToSameChannel(true);
+
+    server_->ListenerThread()->message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(server_, &RestrictedDispatchServer::OnDoPing, 1));
+    sent_ping_event_->Wait();
+    Send(new SyncChannelTestMsg_NoArgs);
+    if (ping_ == 1)
+      ++*success_;
+    else
+      LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
+
+    scoped_ptr<SyncChannel> non_restricted_channel(new SyncChannel(
+        "non_restricted_channel", Channel::MODE_CLIENT, this,
+        ipc_thread().message_loop(), true, shutdown_event()));
+
+    server_->ListenerThread()->message_loop()->PostTask(FROM_HERE,
+        NewRunnableMethod(server_, &RestrictedDispatchServer::OnDoPing, 2));
+    sent_ping_event_->Wait();
+    // Check that the incoming message is *not* dispatched when sending on the
+    // non restricted channel.
+    // TODO(piman): there is a possibility of a false positive race condition
+    // here, if the message that was posted on the server-side end of the pipe
+    // is not visible yet on the client side, but I don't know how to solve this
+    // without hooking into the internals of SyncChannel. I haven't seen it in
+    // practice (i.e. not setting SetRestrictDispatchToSameChannel does cause
+    // the following to fail).
+    non_restricted_channel->Send(new SyncChannelTestMsg_NoArgs);
+    if (ping_ == 1)
+      ++*success_;
+    else
+      LOG(ERROR) << "Send dispatched message from restricted channel";
+
+    Send(new SyncChannelTestMsg_NoArgs);
+    if (ping_ == 2)
+      ++*success_;
+    else
+      LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
+
+    non_restricted_channel->Send(new SyncChannelTestMsg_Done);
+    non_restricted_channel.reset();
+    Send(new SyncChannelTestMsg_Done);
+    Done();
+  }
+
+ private:
+  bool OnMessageReceived(const Message& message) {
+    IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchClient, message)
+     IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Ping, OnPing)
+    IPC_END_MESSAGE_MAP()
+    return true;
+  }
+
+  void OnPing(int ping) {
+    ping_ = ping;
+  }
+
+  int ping_;
+  RestrictedDispatchServer* server_;
+  int* success_;
+  WaitableEvent* sent_ping_event_;
+};
+
+}  // namespace
+
+TEST_F(IPCSyncChannelTest, RestrictedDispatch) {
+  WaitableEvent sent_ping_event(false, false);
+
+  RestrictedDispatchServer* server =
+      new RestrictedDispatchServer(&sent_ping_event);
+  int success = 0;
+  std::vector<Worker*> workers;
+  workers.push_back(new NonRestrictedDispatchServer);
+  workers.push_back(server);
+  workers.push_back(
+      new RestrictedDispatchClient(&sent_ping_event, server, &success));
+  RunTest(workers);
+  EXPECT_EQ(3, success);
+}
diff --git a/ipc/ipc_sync_message_unittest.h b/ipc/ipc_sync_message_unittest.h
index d5c6905..c77d0f8 100644
--- a/ipc/ipc_sync_message_unittest.h
+++ b/ipc/ipc_sync_message_unittest.h
@@ -1,4 +1,4 @@
-// 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.
 
@@ -109,3 +109,6 @@ IPC_SYNC_MESSAGE_ROUTED3_3(Msg_R_3_3, int, std::string, bool, std::string,
 // true, out3 is "3_4", out4 is false
 IPC_SYNC_MESSAGE_ROUTED3_4(Msg_R_3_4, bool, int, std::string, int, bool,
                            std::string, bool)
+
+IPC_MESSAGE_CONTROL1(SyncChannelTestMsg_Ping, int)
+IPC_SYNC_MESSAGE_CONTROL0_0(SyncChannelTestMsg_Done)