Add support for base::Closure in the MessageLoop.

base/message_loop.cc

 // 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.
 
 #include "base/message_loop.h"
 
-#if defined(OS_POSIX) && !defined(OS_MACOSX)
-#include <gdk/gdk.h>
-#include <gdk/gdkx.h>
-#endif
-
 #include <algorithm>
 
+#include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/message_pump_default.h"
 #include "base/metrics/histogram.h"
+#include "base/scoped_ptr.h"
 #include "base/third_party/dynamic_annotations/dynamic_annotations.h"
 #include "base/threading/thread_local.h"
+#include "base/tracked_objects.h"
 
 #if defined(OS_MACOSX)
 #include "base/message_pump_mac.h"
 #endif
 #if defined(OS_POSIX)
 #include "base/message_pump_libevent.h"
 #endif
 #if defined(OS_POSIX) && !defined(OS_MACOSX)
+#include <gdk/gdk.h>
+#include <gdk/gdkx.h>
 #include "base/message_pump_glib.h"
 #endif
 #if defined(TOUCH_UI)
 #include "base/message_pump_glib_x.h"
 #endif
 
 using base::TimeDelta;
 using base::TimeTicks;
 
 namespace {
@@ skipping 35 matching lines @@
 
   // A few events we handle (kindred to messages), and used to profile actions.
   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;
 
+// TODO(ajwong): This is one use case for a Owned() tag that behaves like a
+// "Unique" pointer.  If we had that, and Tasks were always safe to delete on
+// MessageLoop shutdown, this class could just be a function.
+class TaskClosureAdapter : public base::RefCounted<TaskClosureAdapter> {
+ public:
+  // |should_leak_task| points to a flag variable that can be used to determine
+  // if this class should leak the Task on destruction.  This is important
+  // at MessageLoop shutdown since not all tasks can be safely deleted without
+  // running.  See MessageLoop::DeletePendingClosures() for the exact behavior
+  // of when a Task should be deleted. It is subtle.
+  TaskClosureAdapter(Task* task, bool* should_leak_task)
+      : task_(task),
+        should_leak_task_(should_leak_task) {
+  }
+
+  void Run() {
+    task_->Run();
+    delete task_;
+    task_ = NULL;
+  }
+
+ private:
+  friend class base::RefCounted<TaskClosureAdapter>;
+
+  ~TaskClosureAdapter() {
+    if (!*should_leak_task_) {
+      delete task_;
+    }
+  }
+
+  Task* task_;
+  bool* should_leak_task_;
+};
+
 }  // 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);
   return EXCEPTION_CONTINUE_SEARCH;
 }
 
 // Retrieves a pointer to the current unhandled exception filter. There
 // is no standalone getter method.
 static LPTOP_LEVEL_EXCEPTION_FILTER GetTopSEHFilter() {
   LPTOP_LEVEL_EXCEPTION_FILTER top_filter = NULL;
   top_filter = ::SetUnhandledExceptionFilter(0);
   ::SetUnhandledExceptionFilter(top_filter);
   return top_filter;
 }
 
 #endif  // defined(OS_WIN)
 
 //------------------------------------------------------------------------------
 
-MessageLoop::TaskObserver::TaskObserver() {
+MessageLoop::ClosureObserver::ClosureObserver() {
 }
 
-MessageLoop::TaskObserver::~TaskObserver() {
+MessageLoop::ClosureObserver::~ClosureObserver() {
 }
 
 MessageLoop::DestructionObserver::~DestructionObserver() {
 }
 
 //------------------------------------------------------------------------------
 
 MessageLoop::MessageLoop(Type type)
     : type_(type),
       nestable_tasks_allowed_(true),
       exception_restoration_(false),
       message_histogram_(NULL),
       state_(NULL),
+      should_leak_tasks_(true),
 #ifdef OS_WIN
       os_modal_loop_(false),
 #endif  // OS_WIN
       next_sequence_num_(0) {
   DCHECK(!current()) << "should only have one message loop per thread";
   lazy_tls_ptr.Pointer()->Set(this);
 
 // TODO(rvargas): Get rid of the OS guards.
 #if defined(OS_WIN)
 #define MESSAGE_PUMP_UI new base::MessagePumpForUI()
@@ skipping 32 matching lines @@
   DCHECK(!state_);
 
   // 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
   // limit on the number of times we will allow a deleted task to generate more
   // tasks.  Normally, we should only pass through this loop once or twice.  If
   // we end up hitting the loop limit, then it is probably due to one task that
   // is being stubborn.  Inspect the queues to see who is left.
   bool did_work;
   for (int i = 0; i < 100; ++i) {
-    DeletePendingTasks();
+    DeletePendingClosures();
     ReloadWorkQueue();
     // If we end up with empty queues, then break out of the loop.
-    did_work = DeletePendingTasks();
+    did_work = DeletePendingClosures();
     if (!did_work)
       break;
   }
   DCHECK(!did_work);
 
   // Let interested parties have one last shot at accessing this.
   FOR_EACH_OBSERVER(DestructionObserver, destruction_observers_,
                     WillDestroyCurrentMessageLoop());
 
   // OK, now make it so that no one can find us.
@@ 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, Task* task) {
-  PostTask_Helper(from_here, task, 0, true);
+  PendingClosure pending_closure(
+      base::Bind(&TaskClosureAdapter::Run,
+                 new TaskClosureAdapter(task, &should_leak_tasks_)),
+      from_here,
+      CalculateDelayedRuntime(0), true);
+  AddToIncomingQueue(&pending_closure);
 }
 
 void MessageLoop::PostDelayedTask(
     const tracked_objects::Location& from_here, Task* task, int64 delay_ms) {
-  PostTask_Helper(from_here, task, delay_ms, true);
+  PendingClosure pending_closure(
+      base::Bind(&TaskClosureAdapter::Run,
+                 new TaskClosureAdapter(task, &should_leak_tasks_)),
+      from_here,
+      CalculateDelayedRuntime(delay_ms), true);
+  AddToIncomingQueue(&pending_closure);
 }
 
 void MessageLoop::PostNonNestableTask(
     const tracked_objects::Location& from_here, Task* task) {
-  PostTask_Helper(from_here, task, 0, false);
+  PendingClosure pending_closure(
+      base::Bind(&TaskClosureAdapter::Run,
+                 new TaskClosureAdapter(task, &should_leak_tasks_)),
+      from_here,
+      CalculateDelayedRuntime(0), false);
+  AddToIncomingQueue(&pending_closure);
 }
 
 void MessageLoop::PostNonNestableDelayedTask(
     const tracked_objects::Location& from_here, Task* task, int64 delay_ms) {
-  PostTask_Helper(from_here, task, delay_ms, false);
+  PendingClosure pending_closure(
+      base::Bind(&TaskClosureAdapter::Run,
+                 new TaskClosureAdapter(task, &should_leak_tasks_)),
+      from_here,
+      CalculateDelayedRuntime(delay_ms), false);
+  AddToIncomingQueue(&pending_closure);
+}
+
+void MessageLoop::PostClosure(
+    const tracked_objects::Location& from_here, const base::Closure& closure) {
+  DCHECK(!closure.is_null());
+  PendingClosure pending_closure(closure, from_here,
+                                 CalculateDelayedRuntime(0), true);
+  AddToIncomingQueue(&pending_closure);
+}
+
+void MessageLoop::PostDelayedClosure(
+    const tracked_objects::Location& from_here, const base::Closure& closure,
+    int64 delay_ms) {
+  DCHECK(!closure.is_null());
+  PendingClosure pending_closure(closure, from_here,
+                                 CalculateDelayedRuntime(delay_ms), true);
+  AddToIncomingQueue(&pending_closure);
+}
+
+void MessageLoop::PostNonNestableClosure(
+    const tracked_objects::Location& from_here, const base::Closure& closure) {
+  DCHECK(!closure.is_null());
+  PendingClosure pending_closure(closure, from_here,
+                                 CalculateDelayedRuntime(0), false);
+  AddToIncomingQueue(&pending_closure);
+}
+
+void MessageLoop::PostNonNestableDelayedClosure(
+    const tracked_objects::Location& from_here, const base::Closure& closure,
+    int64 delay_ms) {
+  DCHECK(!closure.is_null());
+  PendingClosure pending_closure(closure, from_here,
+                                 CalculateDelayedRuntime(delay_ms), false);
+  AddToIncomingQueue(&pending_closure);
 }
 
 void MessageLoop::Run() {
   AutoRunState save_state(this);
   RunHandler();
 }
 
 void MessageLoop::RunAllPending() {
   AutoRunState save_state(this);
   state_->quit_received = true;  // Means run until we would otherwise block.
@@ skipping 29 matching lines @@
 }
 
 bool MessageLoop::NestableTasksAllowed() const {
   return nestable_tasks_allowed_;
 }
 
 bool MessageLoop::IsNested() {
   return state_->run_depth > 1;
 }
 
-void MessageLoop::AddTaskObserver(TaskObserver* task_observer) {
+void MessageLoop::AddClosureObserver(ClosureObserver* closure_observer) {
   DCHECK_EQ(this, current());
-  task_observers_.AddObserver(task_observer);
+  closure_observers_.AddObserver(closure_observer);
 }
 
-void MessageLoop::RemoveTaskObserver(TaskObserver* task_observer) {
+void MessageLoop::RemoveClosureObserver(ClosureObserver* closure_observer) {
   DCHECK_EQ(this, current());
-  task_observers_.RemoveObserver(task_observer);
+  closure_observers_.RemoveObserver(closure_observer);
 }
 
 void MessageLoop::AssertIdle() const {
   // We only check |incoming_queue_|, since we don't want to lock |work_queue_|.
   base::AutoLock lock(incoming_queue_lock_);
   DCHECK(incoming_queue_.empty());
 }
 
 //------------------------------------------------------------------------------
 
@@ skipping 40 matching lines @@
   pump_->Run(this);
 }
 
 bool MessageLoop::ProcessNextDelayedNonNestableTask() {
   if (state_->run_depth != 1)
     return false;
 
   if (deferred_non_nestable_work_queue_.empty())
     return false;
 
-  Task* task = deferred_non_nestable_work_queue_.front().task;
+  PendingClosure pending_closure = deferred_non_nestable_work_queue_.front();
   deferred_non_nestable_work_queue_.pop();
 
-  RunTask(task);
+  RunClosure(pending_closure);
   return true;
 }
 
-void MessageLoop::RunTask(Task* task) {
+void MessageLoop::RunClosure(const PendingClosure& pending_closure) {
   DCHECK(nestable_tasks_allowed_);
   // Execute the task and assume the worst: It is probably not reentrant.
   nestable_tasks_allowed_ = false;
 
   HistogramEvent(kTaskRunEvent);
-  FOR_EACH_OBSERVER(TaskObserver, task_observers_,
-                    WillProcessTask(task));
-  task->Run();
-  FOR_EACH_OBSERVER(TaskObserver, task_observers_, DidProcessTask(task));
-  delete task;
+  FOR_EACH_OBSERVER(ClosureObserver, closure_observers_,
+                    WillProcessClosure(pending_closure.time_posted));
+  pending_closure.closure.Run();
+  FOR_EACH_OBSERVER(ClosureObserver, closure_observers_,
+                    DidProcessClosure(pending_closure.time_posted));
+
+#if defined(TRACK_ALL_TASK_OBJECTS)
+  if (tracked_objects::ThreadData::IsActive() && pending_closure.post_births) {
+    tracked_objects::ThreadData::current()->TallyADeath(
+        *pending_closure.post_births,
+        TimeTicks::Now() - pending_closure.time_posted);
+  }
+#endif  // defined(TRACK_ALL_TASK_OBJECTS)
 
   nestable_tasks_allowed_ = true;
 }
 
-bool MessageLoop::DeferOrRunPendingTask(const PendingTask& pending_task) {
-  if (pending_task.nestable || state_->run_depth == 1) {
-    RunTask(pending_task.task);
+bool MessageLoop::DeferOrRunPendingClosure(
+    const PendingClosure& pending_closure) {
+  if (pending_closure.nestable || state_->run_depth == 1) {
+    RunClosure(pending_closure);
     // Show that we ran a task (Note: a new one might arrive as a
     // consequence!).
     return true;
   }
 
   // We couldn't run the task now because we're in a nested message loop
   // and the task isn't nestable.
-  deferred_non_nestable_work_queue_.push(pending_task);
+  deferred_non_nestable_work_queue_.push(pending_closure);
   return false;
 }
 
-void MessageLoop::AddToDelayedWorkQueue(const PendingTask& pending_task) {
+void MessageLoop::AddToDelayedWorkQueue(const PendingClosure& pending_closure) {
   // Move to the delayed work queue.  Initialize the sequence number
   // before inserting into the delayed_work_queue_.  The sequence number
   // is used to faciliate FIFO sorting when two tasks have the same
   // delayed_run_time value.
-  PendingTask new_pending_task(pending_task);
-  new_pending_task.sequence_num = next_sequence_num_++;
-  delayed_work_queue_.push(new_pending_task);
+  PendingClosure new_pending_closure(pending_closure);
+  new_pending_closure.sequence_num = next_sequence_num_++;
+  delayed_work_queue_.push(new_pending_closure);
 }
 
 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.
   {
     base::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 MessageLoop::DeletePendingClosures() {
   bool did_work = !work_queue_.empty();
+  // TODO(darin): Delete all tasks once it is safe to do so.
+  // Until it is totally safe, just do it when running Purify or
+  // Valgrind.
+  //
+#if defined(PURIFY) || defined(USE_HEAPCHECKER)
+  should_leak_tasks_ = false;
+#else
+      if (RunningOnValgrind())
+        should_leak_tasks_ = false;
+#endif  // defined(OS_POSIX)
   while (!work_queue_.empty()) {
-    PendingTask pending_task = work_queue_.front();
+    PendingClosure pending_closure = work_queue_.front();
     work_queue_.pop();
-    if (!pending_task.delayed_run_time.is_null()) {
+    if (!pending_closure.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);
-    } else {
-      // TODO(darin): Delete all tasks once it is safe to do so.
-      // Until it is totally safe, just do it when running Purify or
-      // Valgrind.
-#if defined(PURIFY) || defined(USE_HEAPCHECKER)
-      delete pending_task.task;
-#else
-      if (RunningOnValgrind())
-        delete pending_task.task;
-#endif  // defined(OS_POSIX)
+      AddToDelayedWorkQueue(pending_closure);
     }
   }
   did_work |= !deferred_non_nestable_work_queue_.empty();
   while (!deferred_non_nestable_work_queue_.empty()) {
-    // TODO(darin): Delete all tasks once it is safe to do so.
-    // Until it is totaly safe, only delete them under Purify and Valgrind.
-    Task* task = NULL;
-#if defined(PURIFY) || defined(USE_HEAPCHECKER)
-    task = deferred_non_nestable_work_queue_.front().task;
-#else
-    if (RunningOnValgrind())
-      task = deferred_non_nestable_work_queue_.front().task;
-#endif
     deferred_non_nestable_work_queue_.pop();
-    if (task)
-      delete task;
   }
   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.
+  should_leak_tasks_ = false;
   while (!delayed_work_queue_.empty()) {
-    Task* task = delayed_work_queue_.top().task;
     delayed_work_queue_.pop();
-    delete task;
   }
+  should_leak_tasks_ = true;
   return did_work;
 }
 
-// Possibly called on a background thread!
-void MessageLoop::PostTask_Helper(
-    const tracked_objects::Location& from_here, Task* task, int64 delay_ms,
-    bool nestable) {
-  task->SetBirthPlace(from_here);
-
-  PendingTask pending_task(task, nestable);
-
+TimeTicks MessageLoop::CalculateDelayedRuntime(int64 delay_ms) {
+  TimeTicks delayed_run_time;
   if (delay_ms > 0) {
-    pending_task.delayed_run_time =
+    delayed_run_time =
         TimeTicks::Now() + TimeDelta::FromMilliseconds(delay_ms);
 
 #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 =
@@ skipping 11 matching lines @@
 
 #if defined(OS_WIN)
   if (!high_resolution_timer_expiration_.is_null()) {
     if (TimeTicks::Now() > high_resolution_timer_expiration_) {
       base::Time::ActivateHighResolutionTimer(false);
       high_resolution_timer_expiration_ = TimeTicks();
     }
   }
 #endif
 
+  return delayed_run_time;
+}
+
+// Possibly called on a background thread!
+void MessageLoop::AddToIncomingQueue(PendingClosure* pending_closure) {
   // 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<base::MessagePump> pump;
   {
     base::AutoLock locked(incoming_queue_lock_);
 
     bool was_empty = incoming_queue_.empty();
-    incoming_queue_.push(pending_task);
+    incoming_queue_.push(*pending_closure);
+    pending_closure->closure.Reset();
     if (!was_empty)
       return;  // 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_.
 
@@ skipping 28 matching lines @@
     return false;
   }
 
   for (;;) {
     ReloadWorkQueue();
     if (work_queue_.empty())
       break;
 
     // Execute oldest task.
     do {
-      PendingTask pending_task = work_queue_.front();
+      PendingClosure pending_closure = work_queue_.front();
       work_queue_.pop();
-      if (!pending_task.delayed_run_time.is_null()) {
-        AddToDelayedWorkQueue(pending_task);
+      if (!pending_closure.delayed_run_time.is_null()) {
+        AddToDelayedWorkQueue(pending_closure);
         // If we changed the topmost task, then it is time to re-schedule.
-        if (delayed_work_queue_.top().task == pending_task.task)
-          pump_->ScheduleDelayedWork(pending_task.delayed_run_time);
+        if (delayed_work_queue_.top().closure.Equals(pending_closure.closure))
+          pump_->ScheduleDelayedWork(pending_closure.delayed_run_time);
       } else {
-        if (DeferOrRunPendingTask(pending_task))
+        if (DeferOrRunPendingClosure(pending_closure))
           return true;
       }
     } while (!work_queue_.empty());
   }
 
   // Nothing happened.
   return false;
 }
 
-bool MessageLoop::DoDelayedWork(base::TimeTicks* next_delayed_work_time) {
+bool MessageLoop::DoDelayedWork(TimeTicks* next_delayed_work_time) {
   if (!nestable_tasks_allowed_ || delayed_work_queue_.empty()) {
     recent_time_ = *next_delayed_work_time = TimeTicks();
     return false;
   }
 
   // When we "fall behind," there will be a lot of tasks in the delayed work
   // queue that are ready to run.  To increase efficiency when we fall behind,
   // we will only call Time::Now() intermittently, and then process all tasks
   // that are ready to run before calling it again.  As a result, the more we
   // fall behind (and have a lot of ready-to-run delayed tasks), the more
   // efficient we'll be at handling the tasks.
 
   TimeTicks next_run_time = delayed_work_queue_.top().delayed_run_time;
   if (next_run_time > recent_time_) {
     recent_time_ = TimeTicks::Now();  // Get a better view of Now();
     if (next_run_time > recent_time_) {
       *next_delayed_work_time = next_run_time;
       return false;
     }
   }
 
-  PendingTask pending_task = delayed_work_queue_.top();
+  PendingClosure pending_closure = delayed_work_queue_.top();
   delayed_work_queue_.pop();
 
   if (!delayed_work_queue_.empty())
     *next_delayed_work_time = delayed_work_queue_.top().delayed_run_time;
 
-  return DeferOrRunPendingTask(pending_task);
+  return DeferOrRunPendingClosure(pending_closure);
 }
 
 bool MessageLoop::DoIdleWork() {
   if (ProcessNextDelayedNonNestableTask())
     return true;
 
   if (state_->quit_received)
     pump_->Quit();
 
   return false;
@@ skipping 17 matching lines @@
 #if !defined(OS_MACOSX)
   dispatcher = NULL;
 #endif
 }
 
 MessageLoop::AutoRunState::~AutoRunState() {
   loop_->state_ = previous_state_;
 }
 
 //------------------------------------------------------------------------------
-// MessageLoop::PendingTask
+// MessageLoop::PendingClosure
 
-bool MessageLoop::PendingTask::operator<(const PendingTask& other) const {
+MessageLoop::PendingClosure::PendingClosure(
+    const base::Closure& closure,
+    const tracked_objects::Location& posted_from,
+    TimeTicks delayed_run_time,
+    bool nestable)
+    : closure(closure),
+      time_posted(TimeTicks::Now()),
+      delayed_run_time(delayed_run_time),
+      sequence_num(0),
+      nestable(nestable) {
+#if defined(TRACK_ALL_TASK_OBJECTS)
+    if (tracked_objects::ThreadData::IsActive()) {
+      tracked_objects::ThreadData* current_thread_data =
+          tracked_objects::ThreadData::current();
+      if (current_thread_data) {
+        post_births = current_thread_data->TallyABirth(posted_from);
+      } else {
+        // Shutdown started, and this thread wasn't registered.
+        post_births = NULL;
+      }
+    }
+#endif  // defined(TRACK_ALL_TASK_OBJECTS)
+  }
+
+bool MessageLoop::PendingClosure::operator<(const PendingClosure& other) const {
   // Since the top of a priority queue is defined as the "greatest" element, we
   // need to invert the comparison here.  We want the smaller time to be at the
   // top of the heap.
 
   if (delayed_run_time < other.delayed_run_time)
     return false;
 
   if (delayed_run_time > other.delayed_run_time)
     return true;
 
@@ skipping 55 matching lines @@
                                            Watcher *delegate) {
   return pump_libevent()->WatchFileDescriptor(
       fd,
       persistent,
       static_cast<base::MessagePumpLibevent::Mode>(mode),
       controller,
       delegate);
 }
 
 #endif