Made MessagePump a non-thread safe class.

base/message_loop/message_loop.cc

 // Copyright 2013 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 "base/message_loop/message_loop.h"
 
 #include <algorithm>
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
@@ skipping 149 matching lines @@
 #define MESSAGE_PUMP_IO new MessagePumpIOSForIO()
 #elif defined(OS_MACOSX)
 #define MESSAGE_PUMP_UI MessagePumpMac::Create()
 #define MESSAGE_PUMP_IO new MessagePumpLibevent()
 #elif defined(OS_NACL)
 // Currently NaCl doesn't have a UI MessageLoop.
 // TODO(abarth): Figure out if we need this.
 #define MESSAGE_PUMP_UI NULL
 // ipc_channel_nacl.cc uses a worker thread to do socket reads currently, and
 // doesn't require extra support for watching file descriptors.
-#define MESSAGE_PUMP_IO new MessagePumpDefault();
+#define MESSAGE_PUMP_IO new MessagePumpDefault()
 #elif defined(OS_POSIX)  // POSIX but not MACOSX.
 #define MESSAGE_PUMP_UI new MessagePumpForUI()
 #define MESSAGE_PUMP_IO new MessagePumpLibevent()
 #else
 #error Not implemented
 #endif
 
   if (type_ == TYPE_UI) {
     if (message_pump_for_ui_factory_)
-      pump_ = message_pump_for_ui_factory_();
+      pump_.reset(message_pump_for_ui_factory_());
     else
-      pump_ = MESSAGE_PUMP_UI;
+      pump_.reset(MESSAGE_PUMP_UI);
   } else if (type_ == TYPE_IO) {
-    pump_ = MESSAGE_PUMP_IO;
+    pump_.reset(MESSAGE_PUMP_IO);
   } else {
     DCHECK_EQ(TYPE_DEFAULT, type_);
-    pump_ = new MessagePumpDefault();
+    pump_.reset(new MessagePumpDefault());
   }
 }
 
 MessageLoop::~MessageLoop() {
   DCHECK_EQ(this, current());
 
   DCHECK(!run_loop_);
 
   // Clean up any unprocessed tasks, but take care: deleting a task could
   // result in the addition of more tasks (e.g., via DeleteSoon).  We set a
@@ skipping 68 matching lines @@
 void MessageLoop::RemoveDestructionObserver(
     DestructionObserver* destruction_observer) {
   DCHECK_EQ(this, current());
   destruction_observers_.RemoveObserver(destruction_observer);
 }
 
 void MessageLoop::PostTask(
     const tracked_objects::Location& from_here,
     const Closure& task) {
   DCHECK(!task.is_null()) << from_here.ToString();
-  PendingTask pending_task(
-      from_here, task, CalculateDelayedRuntime(TimeDelta()), true);
-  AddToIncomingQueue(&pending_task, false);
+  message_loop_proxy_->AddToIncomingQueue(from_here, task, TimeDelta(), true);
 }
 
 bool MessageLoop::TryPostTask(
     const tracked_objects::Location& from_here,
     const Closure& task) {
   DCHECK(!task.is_null()) << from_here.ToString();
-  PendingTask pending_task(
-      from_here, task, CalculateDelayedRuntime(TimeDelta()), true);
-  return AddToIncomingQueue(&pending_task, true);
+  return message_loop_proxy_->TryAddToIncomingQueue(from_here, task);
 }
 
 void MessageLoop::PostDelayedTask(
     const tracked_objects::Location& from_here,
     const Closure& task,
     TimeDelta delay) {
   DCHECK(!task.is_null()) << from_here.ToString();
-  PendingTask pending_task(
-      from_here, task, CalculateDelayedRuntime(delay), true);
-  AddToIncomingQueue(&pending_task, false);
+  message_loop_proxy_->AddToIncomingQueue(from_here, task, delay, true);
 }
 
 void MessageLoop::PostNonNestableTask(
     const tracked_objects::Location& from_here,
     const Closure& task) {
   DCHECK(!task.is_null()) << from_here.ToString();
-  PendingTask pending_task(
-      from_here, task, CalculateDelayedRuntime(TimeDelta()), false);
-  AddToIncomingQueue(&pending_task, false);
+  message_loop_proxy_->AddToIncomingQueue(from_here, task, TimeDelta(), false);
 }
 
 void MessageLoop::PostNonNestableDelayedTask(
     const tracked_objects::Location& from_here,
     const Closure& task,
     TimeDelta delay) {
   DCHECK(!task.is_null()) << from_here.ToString();
-  PendingTask pending_task(
-      from_here, task, CalculateDelayedRuntime(delay), false);
-  AddToIncomingQueue(&pending_task, false);
+  message_loop_proxy_->AddToIncomingQueue(from_here, task, delay, false);
 }
 
 void MessageLoop::Run() {
   RunLoop run_loop;
   run_loop.Run();
 }
 
 void MessageLoop::RunUntilIdle() {
   RunLoop run_loop;
   run_loop.RunUntilIdle();
@@ skipping 23 matching lines @@
 
 static void QuitCurrentWhenIdle() {
   MessageLoop::current()->QuitWhenIdle();
 }
 
 // static
 Closure MessageLoop::QuitWhenIdleClosure() {
   return Bind(&QuitCurrentWhenIdle);
 }
 
+scoped_refptr<MessageLoopProxy> MessageLoop::message_loop_proxy() {

[rvargas (doing something else), 2013/06/28 03:00:51]

why de-inline this?

[alexeypa (please no reviews), 2013/06/28 17:00:57]

"base/message_loop/message_loop_proxy_impl.h" cannot be included to
message_loop.h because it will create a circular dependency. Without it the
compiler does not know that MessageLoopProxyImpl can be casted to
MessageLoopProxy.

+  return message_loop_proxy_;
+}
+
 void MessageLoop::SetNestableTasksAllowed(bool allowed) {
   if (nestable_tasks_allowed_ != allowed) {
     nestable_tasks_allowed_ = allowed;
     if (!nestable_tasks_allowed_)
       return;
     // Start the native pump if we are not already pumping.
     pump_->ScheduleWork();
   }
 }
 
 bool MessageLoop::NestableTasksAllowed() const {
   return nestable_tasks_allowed_;
 }
 
 bool MessageLoop::IsNested() {
   return run_loop_->run_depth_ > 1;
 }
 
 void MessageLoop::AddTaskObserver(TaskObserver* task_observer) {
   DCHECK_EQ(this, current());
   task_observers_.AddObserver(task_observer);
 }
 
 void MessageLoop::RemoveTaskObserver(TaskObserver* task_observer) {
   DCHECK_EQ(this, current());
   task_observers_.RemoveObserver(task_observer);
 }
 
 void MessageLoop::AssertIdle() const {

[rvargas (doing something else), 2013/06/28 03:00:51]

This should probably be a ForTest method.

[alexeypa (please no reviews), 2013/06/28 17:00:57]

Done.

   // We only check |incoming_queue_|, since we don't want to lock |work_queue_|.
-  AutoLock lock(incoming_queue_lock_);
+  AutoLock lock(message_loop_proxy_->message_loop_lock_);

[rvargas (doing something else), 2013/06/28 03:00:51]

we should do something to avoid grabbing this lock directly.

[alexeypa (please no reviews), 2013/06/28 17:00:57]

Done.

   DCHECK(incoming_queue_.empty());
 }
 
 bool MessageLoop::is_running() const {
   DCHECK_EQ(this, current());
   return run_loop_ != NULL;
 }
 
 //------------------------------------------------------------------------------
 
@@ skipping 112 matching lines @@
 void MessageLoop::ReloadWorkQueue() {
   // We can improve performance of our loading tasks from incoming_queue_ to
   // work_queue_ by waiting until the last minute (work_queue_ is empty) to
   // load.  That reduces the number of locks-per-task significantly when our
   // queues get large.
   if (!work_queue_.empty())
     return;  // Wait till we *really* need to lock and load.
 
   // Acquire all we can from the inter-thread queue with one lock acquisition.
   {
-    AutoLock lock(incoming_queue_lock_);
+    AutoLock lock(message_loop_proxy_->message_loop_lock_);

[rvargas (doing something else), 2013/06/28 03:00:51]

Maybe

message_loop_proxy_->LoadPendingTasks();

MessageLoopProxy::LoadPendingTasks() {
  grab lock
  loop_->LoadPendingTasks();
}

MessageLoop::LoadPendingTasks() {
  DCHECK(work_queue_.empty());
  ...
}
(again, documenting that access to LoadPendingTasks should be serialized by the
caller)

Or at the very least using a public accessor to the lock, but using a lock of
another class doesn't look clean.

[alexeypa (please no reviews), 2013/06/28 17:00:57]

Done.

     if (incoming_queue_.empty())
       return;
     incoming_queue_.Swap(&work_queue_);  // Constant time
     DCHECK(incoming_queue_.empty());
   }
 }
 
 bool MessageLoop::DeletePendingTasks() {
   bool did_work = !work_queue_.empty();
   while (!work_queue_.empty()) {
@@ skipping 55 matching lines @@
       Time::ActivateHighResolutionTimer(false);
       high_resolution_timer_expiration_ = TimeTicks();
     }
   }
 #endif
 
   return delayed_run_time;
 }
 
 // Possibly called on a background thread!
-bool MessageLoop::AddToIncomingQueue(PendingTask* pending_task,
-                                     bool use_try_lock) {
+void MessageLoop::AddToIncomingQueue(const tracked_objects::Location& from_here,
+                                     const Closure& task,
+                                     TimeDelta delay,
+                                     bool nestable) {
   // Warning: Don't try to short-circuit, and handle this thread's tasks more
   // directly, as it could starve handling of foreign threads.  Put every task
   // into this queue.
 
-  scoped_refptr<MessagePump> pump;
-  {
-    if (use_try_lock) {
-      if (!incoming_queue_lock_.Try()) {
-        pending_task->task.Reset();
-        return false;
-      }
-    } else {
-      incoming_queue_lock_.Acquire();
-    }
-    AutoLock locked(incoming_queue_lock_, AutoLock::AlreadyAcquired());
-    // Initialize the sequence number. The sequence number is used for delayed
-    // tasks (to faciliate FIFO sorting when two tasks have the same
-    // delayed_run_time value) and for identifying the task in about:tracing.
-    pending_task->sequence_num = next_sequence_num_++;
+  // This should only be called while the lock is taken.
+  message_loop_proxy_->message_loop_lock_.AssertAcquired();
 
-    TRACE_EVENT_FLOW_BEGIN0("task", "MessageLoop::PostTask",
-        TRACE_ID_MANGLE(GetTaskTraceID(*pending_task, this)));
+  PendingTask pending_task(
+      from_here, task, CalculateDelayedRuntime(delay), nestable);
 
-    bool was_empty = incoming_queue_.empty();
-    incoming_queue_.push(*pending_task);
-    pending_task->task.Reset();
-    if (!was_empty)
-      return true;  // Someone else should have started the sub-pump.
+  // Initialize the sequence number. The sequence number is used for delayed
+  // tasks (to faciliate FIFO sorting when two tasks have the same
+  // delayed_run_time value) and for identifying the task in about:tracing.
+  pending_task.sequence_num = next_sequence_num_++;
 
-    pump = pump_;
-  }
-  // Since the incoming_queue_ may contain a task that destroys this message
-  // loop, we cannot exit incoming_queue_lock_ until we are done with |this|.
-  // We use a stack-based reference to the message pump so that we can call
-  // ScheduleWork outside of incoming_queue_lock_.
+  TRACE_EVENT_FLOW_BEGIN0("task", "MessageLoop::PostTask",
+      TRACE_ID_MANGLE(GetTaskTraceID(pending_task, this)));
 
-  pump->ScheduleWork();
-  return true;
+  bool was_empty = incoming_queue_.empty();
+  incoming_queue_.push(pending_task);
+  pending_task.task.Reset();
+
+  // Wake up the pump.
+  if (was_empty)
+    pump_->ScheduleWork();
 }
 
 //------------------------------------------------------------------------------
 // Method and data for histogramming events and actions taken by each instance
 // on each thread.
 
 void MessageLoop::StartHistogrammer() {
 #if !defined(OS_NACL)  // NaCl build has no metrics code.
   if (enable_histogrammer_ && !message_histogram_
       && StatisticsRecorder::IsActive()) {
@@ skipping 176 matching lines @@
       fd,
       persistent,
       mode,
       controller,
       delegate);
 }
 
 #endif
 
 }  // namespace base