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));
-}
-
 }  // namespace
 
 //------------------------------------------------------------------------------
 
 #if defined(OS_WIN)
 
 // Upon a SEH exception in this thread, it restores the original unhandled
 // exception filter.
 static int SEHFilter(LPTOP_LEVEL_EXCEPTION_FILTER old_filter) {
   ::SetUnhandledExceptionFilter(old_filter);
@@ skipping 19 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();
+  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 29 matching lines @@
 Closure MessageLoop::QuitWhenIdleClosure() {
   return Bind(&QuitCurrentWhenIdle);
 }
 
 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();
+    WakeUpPump();
   }
 }
 
 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 {
-  // 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::IsHighResolutionTimerEnabledForTest() {
+  return incoming_task_queue_->IsHighResolutionTimerEnabledForTest();
+}
+
+bool MessageLoop::IsIdleForTest() {
+  // We only check the imcoming queue|, since we don't want to lock the work
+  // queue.
+  return incoming_task_queue_->IsIdleForTest();
+}
+
+void MessageLoop::LockWaitUnLockForTest(WaitableEvent* caller_wait,
+                                        WaitableEvent* caller_signal) {
+  incoming_task_queue_->LockWaitUnLockForTest(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());
-    // 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();
-    if (!was_empty)
-      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::WakeUpPump() {
+  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