Cleanup IPC::ChannelProxy to use SingleThreadTaskRunner

ipc/ipc_sync_channel.cc

 // 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_sync_channel.h"
 
 #include "base/bind.h"
 #include "base/lazy_instance.h"
 #include "base/location.h"
 #include "base/logging.h"
-#include "base/threading/thread_local.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/synchronization/waitable_event_watcher.h"
+#include "base/thread_task_runner_handle.h"
+#include "base/threading/thread_local.h"
 #include "ipc/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
@@ skipping 38 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));
       message_queue_version_++;
     }
 
     dispatch_event_.Signal();
     if (!was_task_pending) {
-      listener_message_loop_->PostTask(
+      listener_task_runner_->PostTask(
           FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchMessagesTask,
                                 this, 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
@@ skipping 55 matching lines @@
       }
     }
 
     if (--listener_count_ == 0) {
       DCHECK(lazy_tls_ptr_.Pointer()->Get());
       lazy_tls_ptr_.Pointer()->Set(NULL);
     }
   }
 
   WaitableEvent* dispatch_event() { return &dispatch_event_; }
-  base::MessageLoopProxy* listener_message_loop() {
-    return listener_message_loop_;
+  base::SingleThreadTaskRunner* listener_task_runner() {
+    return listener_task_runner_;
   }
 
   // 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) {
@@ skipping 15 matching lines @@
   }
 
  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_(true, false),
-      listener_message_loop_(base::MessageLoopProxy::current()),
+      listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
       task_pending_(false),
       listener_count_(0),
       top_send_done_watcher_(NULL) {
   }
 
   ~ReceivedSyncMsgQueue() {}
 
   // 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::list<QueuedMessage> SyncMessageQueue;
   SyncMessageQueue message_queue_;
   uint32 message_queue_version_;  // Used to signal DispatchMessages to rescan
 
   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_;
-  scoped_refptr<base::MessageLoopProxy> listener_message_loop_;
+  scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
   base::Lock message_lock_;
   bool task_pending_;
   int listener_count_;
 
   // The current send done event watcher for this thread. Used to maintain
   // a local global stack of send done watchers to ensure that nested sync
   // message loops complete correctly.
   base::WaitableEventWatcher* top_send_done_watcher_;
 };
 
 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> >
     SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_ =
         LAZY_INSTANCE_INITIALIZER;
 
 SyncChannel::SyncContext::SyncContext(
     Listener* listener,
-    base::MessageLoopProxy* ipc_thread,
+    base::SingleThreadTaskRunner* ipc_task_runner,
     WaitableEvent* shutdown_event)
-    : ChannelProxy::Context(listener, ipc_thread),
+    : ChannelProxy::Context(listener, ipc_task_runner),
       received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
       shutdown_event_(shutdown_event),
       restrict_dispatch_group_(kRestrictDispatchGroup_None) {
 }
 
 SyncChannel::SyncContext::~SyncContext() {
   while (!deserializers_.empty())
     Pop();
 }
 
@@ skipping 26 matching lines @@
     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(
+  ipc_task_runner()->PostTask(
       FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchReplies,
                             received_sync_msgs_.get()));
 
   return result;
 }
 
 WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
   base::AutoLock auto_lock(deserializers_lock_);
   return deserializers_.back().done_event;
 }
@@ skipping 94 matching lines @@
     DCHECK_EQ(GetSendDoneEvent(), event);
     MessageLoop::current()->QuitNow();
   }
 }
 
 
 SyncChannel::SyncChannel(
     const IPC::ChannelHandle& channel_handle,
     Channel::Mode mode,
     Listener* listener,
-    base::MessageLoopProxy* ipc_message_loop,
+    base::SingleThreadTaskRunner* ipc_task_runner,
     bool create_pipe_now,
     WaitableEvent* shutdown_event)
-    : ChannelProxy(new SyncContext(listener, ipc_message_loop, shutdown_event)),
+    : ChannelProxy(new SyncContext(listener, ipc_task_runner, shutdown_event)),
       sync_messages_with_no_timeout_allowed_(true) {
   ChannelProxy::Init(channel_handle, mode, create_pipe_now);
   StartWatching();
 }
 
 SyncChannel::SyncChannel(
     Listener* listener,
-    base::MessageLoopProxy* ipc_message_loop,
+    base::SingleThreadTaskRunner* ipc_task_runner,
     WaitableEvent* shutdown_event)
-    : ChannelProxy(new SyncContext(listener, ipc_message_loop, shutdown_event)),
+    : ChannelProxy(new SyncContext(listener, ipc_task_runner, shutdown_event)),
       sync_messages_with_no_timeout_allowed_(true) {
   StartWatching();
 }
 
 SyncChannel::~SyncChannel() {
 }
 
 void SyncChannel::SetRestrictDispatchChannelGroup(int group) {
   sync_context()->set_restrict_dispatch_group(group);
 }
@@ skipping 21 matching lines @@
   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(
+    context->ipc_task_runner()->PostDelayedTask(
         FROM_HERE,
         base::Bind(&SyncContext::OnSendTimeout, context.get(), message_id),
         base::TimeDelta::FromMilliseconds(timeout_ms));
   }
 
   // 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();
@@ skipping 75 matching lines @@
   // 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);
 }
 
 }  // namespace IPC