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"
 #include "base/debug/alias.h"
 #include "base/debug/trace_event.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
-#include "base/message_loop/message_loop_proxy_impl.h"
 #include "base/message_loop/message_pump_default.h"
 #include "base/metrics/histogram.h"
 #include "base/metrics/statistics_recorder.h"
 #include "base/run_loop.h"
 #include "base/third_party/dynamic_annotations/dynamic_annotations.h"
 #include "base/thread_task_runner_handle.h"
 #include "base/threading/thread_local.h"
 #include "base/time/time.h"
 #include "base/tracked_objects.h"
 
@@ skipping 55 matching lines @@
   VALUE_TO_NUMBER_AND_NAME(kTaskRunEvent)
   VALUE_TO_NUMBER_AND_NAME(kTimerEvent)
 
   {-1, NULL}  // The list must be null terminated, per API to histogram.
 };
 
 bool enable_histogrammer_ = false;
 
 MessageLoop::MessagePumpFactory* message_pump_for_ui_factory_ = NULL;
 
-// Create a process-wide unique ID to represent this task in trace events. This
-// will be mangled with a Process ID hash to reduce the likelyhood of colliding
-// with MessageLoop pointers on other processes.
-uint64 GetTaskTraceID(const PendingTask& task, MessageLoop* loop) {
-  return (static_cast<uint64>(task.sequence_num) << 32) |
-         static_cast<uint64>(reinterpret_cast<intptr_t>(loop));
-}
-
 // Returns true if MessagePump::ScheduleWork() must be called one
 // time for every task that is added to the MessageLoop incoming queue.
 bool AlwaysNotifyPump(MessageLoop::Type type) {
 #if defined(OS_ANDROID)
   return type == MessageLoop::TYPE_UI;
 #else
   return false;
 #endif
 }
 
@@ skipping 29 matching lines @@
 MessageLoop::TaskObserver::~TaskObserver() {
 }
 
 MessageLoop::DestructionObserver::~DestructionObserver() {
 }
 
 //------------------------------------------------------------------------------
 
 MessageLoop::MessageLoop(Type type)
     : type_(type),
+      exception_restoration_(false),
       nestable_tasks_allowed_(true),
-      exception_restoration_(false),
-      message_histogram_(NULL),
-      run_loop_(NULL),
 #if defined(OS_WIN)
       os_modal_loop_(false),
 #endif  // OS_WIN
-      next_sequence_num_(0) {
+      message_histogram_(NULL),
+      run_loop_(NULL) {
   DCHECK(!current()) << "should only have one message loop per thread";
   lazy_tls_ptr.Pointer()->Set(this);
 
-  message_loop_proxy_ = new MessageLoopProxyImpl();
+  incoming_task_queue_ = new internal::IncomingTaskQueue(this);
+  message_loop_proxy_ =
+      new internal::MessageLoopProxyImpl(incoming_task_queue_);
   thread_task_runner_handle_.reset(
       new ThreadTaskRunnerHandle(message_loop_proxy_));
 
 // TODO(rvargas): Get rid of the OS guards.
 #if defined(OS_WIN)
 #define MESSAGE_PUMP_UI new MessagePumpForUI()
 #define MESSAGE_PUMP_IO new MessagePumpForIO()
 #elif defined(OS_IOS)
 #define MESSAGE_PUMP_UI MessagePumpMac::Create()
 #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 11 matching lines @@
       break;
   }
   DCHECK(!did_work);
 
   // Let interested parties have one last shot at accessing this.
   FOR_EACH_OBSERVER(DestructionObserver, destruction_observers_,
                     WillDestroyCurrentMessageLoop());
 
   thread_task_runner_handle_.reset();
 
-  // Tell the message_loop_proxy that we are dying.
-  static_cast<MessageLoopProxyImpl*>(message_loop_proxy_.get())->
-      WillDestroyCurrentMessageLoop();
+  // Tell the incoming queue that we are dying.
+  incoming_task_queue_->WillDestroyCurrentMessageLoop();
+  incoming_task_queue_ = NULL;
   message_loop_proxy_ = NULL;
 
   // OK, now make it so that no one can find us.
   lazy_tls_ptr.Pointer()->Set(NULL);
-
-#if defined(OS_WIN)
-  // If we left the high-resolution timer activated, deactivate it now.
-  // Doing this is not-critical, it is mainly to make sure we track
-  // the high resolution timer activations properly in our unit tests.
-  if (!high_resolution_timer_expiration_.is_null()) {
-    Time::ActivateHighResolutionTimer(false);
-    high_resolution_timer_expiration_ = TimeTicks();
-  }
-#endif
 }
 
 // static
 MessageLoop* MessageLoop::current() {
   // TODO(darin): sadly, we cannot enable this yet since people call us even
   // when they have no intention of using us.
   // DCHECK(loop) << "Ouch, did you forget to initialize me?";
   return lazy_tls_ptr.Pointer()->Get();
 }
 
@@ skipping 20 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);
+  incoming_task_queue_->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 incoming_task_queue_->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);
+  incoming_task_queue_->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);
+  incoming_task_queue_->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);
+  incoming_task_queue_->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 51 matching lines @@
 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 {
-  // We only check |incoming_queue_|, since we don't want to lock |work_queue_|.
-  AutoLock lock(incoming_queue_lock_);
-  DCHECK(incoming_queue_.empty());
-}
-
 bool MessageLoop::is_running() const {
   DCHECK_EQ(this, current());
   return run_loop_ != NULL;
 }
 
+bool MessageLoop::IsHighResolutionTimerEnabledForTesting() {
+  return incoming_task_queue_->IsHighResolutionTimerEnabledForTesting();
+}
+
+bool MessageLoop::IsIdleForTesting() {
+  // We only check the imcoming queue|, since we don't want to lock the work
+  // queue.
+  return incoming_task_queue_->IsIdleForTesting();
+}
+
+void MessageLoop::LockWaitUnLockForTesting(WaitableEvent* caller_wait,
+                                           WaitableEvent* caller_signal) {
+  incoming_task_queue_->LockWaitUnLockForTesting(caller_wait, caller_signal);
+}
+
 //------------------------------------------------------------------------------
 
 // Runs the loop in two different SEH modes:
 // enable_SEH_restoration_ = false : any unhandled exception goes to the last
 // one that calls SetUnhandledExceptionFilter().
 // enable_SEH_restoration_ = true : any unhandled exception goes to the filter
 // that was existed before the loop was run.
 void MessageLoop::RunHandler() {
 #if defined(OS_WIN)
   if (exception_restoration_) {
@@ skipping 44 matching lines @@
 
   RunTask(pending_task);
   return true;
 }
 
 void MessageLoop::RunTask(const PendingTask& pending_task) {
   tracked_objects::TrackedTime start_time =
       tracked_objects::ThreadData::NowForStartOfRun(pending_task.birth_tally);
 
   TRACE_EVENT_FLOW_END1("task", "MessageLoop::PostTask",
-      TRACE_ID_MANGLE(GetTaskTraceID(pending_task, this)),
+      TRACE_ID_MANGLE(GetTaskTraceID(pending_task)),
       "queue_duration",
       (start_time - pending_task.EffectiveTimePosted()).InMilliseconds());
   TRACE_EVENT2("task", "MessageLoop::RunTask",
                "src_file", pending_task.posted_from.file_name(),
                "src_func", pending_task.posted_from.function_name());
 
   DCHECK(nestable_tasks_allowed_);
   // Execute the task and assume the worst: It is probably not reentrant.
   nestable_tasks_allowed_ = false;
 
@@ skipping 32 matching lines @@
   // and the task isn't nestable.
   deferred_non_nestable_work_queue_.push(pending_task);
   return false;
 }
 
 void MessageLoop::AddToDelayedWorkQueue(const PendingTask& pending_task) {
   // Move to the delayed work queue.
   delayed_work_queue_.push(pending_task);
 }
 
-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_);
-    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()) {
     PendingTask pending_task = work_queue_.front();
     work_queue_.pop();
     if (!pending_task.delayed_run_time.is_null()) {
       // We want to delete delayed tasks in the same order in which they would
       // normally be deleted in case of any funny dependencies between delayed
       // tasks.
       AddToDelayedWorkQueue(pending_task);
     }
   }
   did_work |= !deferred_non_nestable_work_queue_.empty();
   while (!deferred_non_nestable_work_queue_.empty()) {
     deferred_non_nestable_work_queue_.pop();
   }
   did_work |= !delayed_work_queue_.empty();
 
   // Historically, we always delete the task regardless of valgrind status. It's
   // not completely clear why we want to leak them in the loops above.  This
   // code is replicating legacy behavior, and should not be considered
   // absolutely "correct" behavior.  See TODO above about deleting all tasks
   // when it's safe.
   while (!delayed_work_queue_.empty()) {
     delayed_work_queue_.pop();
   }
   return did_work;
 }
 
-TimeTicks MessageLoop::CalculateDelayedRuntime(TimeDelta delay) {
-  TimeTicks delayed_run_time;
-  if (delay > TimeDelta()) {
-    delayed_run_time = TimeTicks::Now() + delay;
-
-#if defined(OS_WIN)
-    if (high_resolution_timer_expiration_.is_null()) {
-      // Windows timers are granular to 15.6ms.  If we only set high-res
-      // timers for those under 15.6ms, then a 18ms timer ticks at ~32ms,
-      // which as a percentage is pretty inaccurate.  So enable high
-      // res timers for any timer which is within 2x of the granularity.
-      // This is a tradeoff between accuracy and power management.
-      bool needs_high_res_timers = delay.InMilliseconds() <
-          (2 * Time::kMinLowResolutionThresholdMs);
-      if (needs_high_res_timers) {
-        if (Time::ActivateHighResolutionTimer(true)) {
-          high_resolution_timer_expiration_ = TimeTicks::Now() +
-              TimeDelta::FromMilliseconds(kHighResolutionTimerModeLeaseTimeMs);
-        }
-      }
-    }
-#endif
-  } else {
-    DCHECK_EQ(delay.InMilliseconds(), 0) << "delay should not be negative";
-  }
-
-#if defined(OS_WIN)
-  if (!high_resolution_timer_expiration_.is_null()) {
-    if (TimeTicks::Now() > high_resolution_timer_expiration_) {
-      Time::ActivateHighResolutionTimer(false);
-      high_resolution_timer_expiration_ = TimeTicks();
-    }
-  }
-#endif
-
-  return delayed_run_time;
+uint64 MessageLoop::GetTaskTraceID(const PendingTask& task) {
+  return (static_cast<uint64>(task.sequence_num) << 32) |
+         static_cast<uint64>(reinterpret_cast<intptr_t>(this));
 }
 
-// Possibly called on a background thread!
-bool MessageLoop::AddToIncomingQueue(PendingTask* pending_task,
-                                     bool use_try_lock) {
-  // 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.
+void MessageLoop::ReloadWorkQueue() {
+  // We can improve performance of our loading tasks from the 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())
+    incoming_task_queue_->ReloadWorkQueue(&work_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());

[jar1, 2013/07/29 23:30:50]

It is very important when using locks to hold them for the minimal length of
time, and also to try to avoid calling any other functions that may "take time"
or worse "grab other locks."  

Specifically in this case, care was taken to release the lock on line 646 before
calling ScheduleWork().

In addition, this code was careful to do as much work outside the lock as
possible, such as the construction of the task, which was done in the caller.

[alexeypa (please no reviews), 2013/07/30 00:31:13]

Yes, I know. Unfortunately that caused a race down in Win32 code when one thread
was destroying the message loop, while the other was executing ScheduleWork().  

-    // 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_++;
-
-    TRACE_EVENT_FLOW_BEGIN0("task", "MessageLoop::PostTask",
-        TRACE_ID_MANGLE(GetTaskTraceID(*pending_task, this)));
-
-    bool was_empty = incoming_queue_.empty();
-    incoming_queue_.push(*pending_task);
-    pending_task->task.Reset();
-    // The Android UI message loop needs to get notified each time
-    // a task is added to the incoming queue.
-    if (!was_empty && !AlwaysNotifyPump(type_))
-      return true;  // Someone else should have started the sub-pump.
-
-    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_.
-
-  pump->ScheduleWork();
-  return true;
+void MessageLoop::ScheduleWork(bool was_empty) {

[jar1, 2013/07/29 23:30:50]

This is (I think) where you forward the call while holding the lock on the
queue.

+  // The Android UI message loop needs to get notified each time
+  // a task is added to the incoming queue.
+  if (was_empty || AlwaysNotifyPump(type_))
+    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