Support Closure in ALL the loops!

base/threading/worker_pool_posix.cc

Index: base/threading/worker_pool_posix.cc
diff --git a/base/threading/worker_pool_posix.cc b/base/threading/worker_pool_posix.cc
index bd6210f3165d136ef8be833f68a3c0cbad7b9e70..4d5075b6090ef4f4dfe780c06116d2b2cc429541 100644
--- a/base/threading/worker_pool_posix.cc
+++ b/base/threading/worker_pool_posix.cc
@@ -4,6 +4,7 @@
 
 #include "base/threading/worker_pool_posix.h"
 
+#include "base/bind.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/memory/ref_counted.h"
@@ -11,6 +12,7 @@
 #include "base/task.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/worker_pool.h"
+#include "base/tracked_objects.h"
 
 namespace base {
 
@@ -28,6 +30,8 @@ class WorkerPoolImpl {
 
   void PostTask(const tracked_objects::Location& from_here, Task* task,
                 bool task_is_slow);
+  void PostTask(const tracked_objects::Location& from_here,
+                const base::Closure& task, bool task_is_slow);
 
  private:
   scoped_refptr<base::PosixDynamicThreadPool> pool_;
@@ -44,25 +48,27 @@ WorkerPoolImpl::~WorkerPoolImpl() {
 
 void WorkerPoolImpl::PostTask(const tracked_objects::Location& from_here,
                               Task* task, bool task_is_slow) {
-  task->SetBirthPlace(from_here);
-  pool_->PostTask(task);
+  pool_->PostTask(from_here, task);
+}
+
+void WorkerPoolImpl::PostTask(const tracked_objects::Location& from_here,
+                              const base::Closure& task, bool task_is_slow) {
+  pool_->PostTask(from_here, task);
 }
 
 base::LazyInstance<WorkerPoolImpl> g_lazy_worker_pool(base::LINKER_INITIALIZED);
 
 class WorkerThread : public PlatformThread::Delegate {
  public:
-  WorkerThread(const std::string& name_prefix, int idle_seconds_before_exit,
+  WorkerThread(const std::string& name_prefix,
                base::PosixDynamicThreadPool* pool)
       : name_prefix_(name_prefix),
-        idle_seconds_before_exit_(idle_seconds_before_exit),
         pool_(pool) {}
 
   virtual void ThreadMain();
 
  private:
   const std::string name_prefix_;
-  const int idle_seconds_before_exit_;
   scoped_refptr<base::PosixDynamicThreadPool> pool_;
 
   DISALLOW_COPY_AND_ASSIGN(WorkerThread);
@@ -74,11 +80,15 @@ void WorkerThread::ThreadMain() {
   PlatformThread::SetName(name.c_str());
 
   for (;;) {
-    Task* task = pool_->WaitForTask();
-    if (!task)
+    PosixDynamicThreadPool::PendingTask pending_task = pool_->WaitForTask();
+    if (pending_task.task.is_null())
       break;
-    task->Run();
-    delete task;
+    pending_task.task.Run();
+#if defined(TRACK_ALL_TASK_OBJECTS)
+    tracked_objects::ThreadData::TallyADeathIfActive(
+        pending_task.post_births,
+        TimeTicks::Now() - pending_task.time_posted);
+#endif  // defined(TRACK_ALL_TASK_OBJECTS)
   }
 
   // The WorkerThread is non-joinable, so it deletes itself.
@@ -93,21 +103,39 @@ bool WorkerPool::PostTask(const tracked_objects::Location& from_here,
   return true;
 }
 
+bool WorkerPool::PostTask(const tracked_objects::Location& from_here,

[willchan no longer on Chromium, 2011/07/08 12:04:21]

Minor nit, perhaps not even worth fixing. So don't bother unless you care.

I'm a bit surprised we don't do the Task/Closure=>PendingTask conversion at the
outermost API, which is the static WorkerPool::PostTask() methods. I would have
thought we would want to eliminate the two paths at the earliest point possible.

[awong, 2011/07/08 18:36:51]

Yeah. That was my first attempt.  Because Closure is refcounted, you can't just
create a Closure of the task and pass it in on the spot since then you don't
know who will delete it.

Thus, we have to create a PendingTask carrier that can do a destructive pass off
of ownership for "task" safely.

Since the PendingTask handoff has strange semantics (eg., you construct it in
one spot, and then it gets destructively passed into the pool), it felt cleaner
to keep all the ugly stuff in PosixDynamicThreadPool.  That way WorkerPoolImpl
and WorkerPool are nice and thin.

Another question I was wondering about is whether or not we really needed
WorkerPoolImpl...but I didn't want to tackle that in the current change.

[willchan no longer on Chromium, 2011/07/10 16:47:33]

OK, I see. That makes sense. Comments explaining the destructive handoff would
be nice.
It's not clear to me what the alternative is...make each member variable be
global itself?

[awong, 2011/07/14 23:03:30]

Added a comment into one of the PostTasks.
Make  g_lazy_worker_pool be base::LazyInstance<PosixDynamicThreadPool>
g_lazy_worker_pool(base::LINKER_INITIALIZED);

Let PosixDynamicThreadPool's default constructor hard-code "WorkerPool" and
kIdleSecondsBeforeExit.  Then have WorkerPool::PostTask call
PosixDynamicThreadPool::PostTask directly (dropping the task_is_slow flag).

+                          const base::Closure& task, bool task_is_slow) {
+  g_lazy_worker_pool.Pointer()->PostTask(from_here, task, task_is_slow);
+  return true;
+}
+
+PosixDynamicThreadPool::PendingTask::PendingTask(
+    const tracked_objects::Location& posted_from,
+    const base::Closure& task)
+    : task(task) {
+#if defined(TRACK_ALL_TASK_OBJECTS)
+  post_births = tracked_objects::ThreadData::TallyABirthIfActive(posted_from);
+  time_posted = TimeTicks::Now();
+#endif  // defined(TRACK_ALL_TASK_OBJECTS)
+}
+
 PosixDynamicThreadPool::PosixDynamicThreadPool(
     const std::string& name_prefix,
     int idle_seconds_before_exit)
     : name_prefix_(name_prefix),
       idle_seconds_before_exit_(idle_seconds_before_exit),
-      tasks_available_cv_(&lock_),
+      pending_tasks_available_cv_(&lock_),
       num_idle_threads_(0),
       terminated_(false),
       num_idle_threads_cv_(NULL) {}
 
 PosixDynamicThreadPool::~PosixDynamicThreadPool() {
-  while (!tasks_.empty()) {
-    Task* task = tasks_.front();
-    tasks_.pop();
-    delete task;
+  while (!pending_tasks_.empty()) {
+    PendingTask pending_task = pending_tasks_.front();
+    pending_tasks_.pop();
+    tracked_objects::ThreadData::TallyADeathIfActive(

[willchan no longer on Chromium, 2011/07/08 12:04:21]

These calls are probably what's causing your shutdown crashes. They rely on
MessageLoop::current(), which uses LazyInstance. The LazyInstance probably
already got destroyed by the AtExitManager. I suspect ~AtExitManager's body
destroys the LazyInstance, but in between the window where the Lock is released
at the end of the body and before ~AtExitManager begins deleting the member
variables, this pool recreates MessageLoop::current()'s LazyInstance and
acquires the lock, so then when it gets destroyed, it aborts since it's illegal
to delete a lock while it's held.

[awong, 2011/07/08 18:36:51]

Ah.  Interesting...  I can see how that might do that, but the new code should
just be replicating existing behavior since "delete task;" previously should
have invoked the same code path on the Task's destructor.  I'll test by just
removing this and seeing what happens.

[willchan no longer on Chromium, 2011/07/10 16:47:33]

Looks like the Mac trybot run passed. Try again to see if it's flaky?

[awong, 2011/07/14 23:03:30]

I'm pretty consistently getting an error when the object tracking code is
enabled. However, I can't repro locally, and I can't see any code path by which
there should be a problem.  Maybe I should just remove the tracking from
workerpool for right now...does that seem okay to you?

[willchan no longer on Chromium, 2011/07/19 12:18:26]

Suboptimal, but I'm OK with it. I don't use it. I defer to Darin on whether this
is OK. I bet he defers to jar@.

+        pending_task.post_births,
+        TimeTicks::Now() - pending_task.time_posted);
   }
 }
 
@@ -117,53 +145,70 @@ void PosixDynamicThreadPool::Terminate() {
     DCHECK(!terminated_) << "Thread pool is already terminated.";
     terminated_ = true;
   }
-  tasks_available_cv_.Broadcast();
+  pending_tasks_available_cv_.Broadcast();
+}
+
+void PosixDynamicThreadPool::PostTask(
+    const tracked_objects::Location& from_here,
+    Task* task) {
+  PendingTask pending_task(from_here,
+                           base::Bind(&subtle::TaskClosureAdapter::Run,
+                                      new subtle::TaskClosureAdapter(task)));
+  AddTask(&pending_task);
+}
+
+void PosixDynamicThreadPool::PostTask(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task) {
+  PendingTask pending_task(from_here, task);
+  AddTask(&pending_task);
 }
 
-void PosixDynamicThreadPool::PostTask(Task* task) {
+void PosixDynamicThreadPool::AddTask(PendingTask* pending_task) {
   AutoLock locked(lock_);
   DCHECK(!terminated_) <<
       "This thread pool is already terminated.  Do not post new tasks.";
 
-  tasks_.push(task);
+  pending_tasks_.push(*pending_task);
+  pending_task->task.Reset();
 
   // We have enough worker threads.
-  if (static_cast<size_t>(num_idle_threads_) >= tasks_.size()) {
-    tasks_available_cv_.Signal();
+  if (static_cast<size_t>(num_idle_threads_) >= pending_tasks_.size()) {
+    pending_tasks_available_cv_.Signal();
   } else {
     // The new PlatformThread will take ownership of the WorkerThread object,
     // which will delete itself on exit.
     WorkerThread* worker =
-        new WorkerThread(name_prefix_, idle_seconds_before_exit_, this);
+        new WorkerThread(name_prefix_, this);
     PlatformThread::CreateNonJoinable(kWorkerThreadStackSize, worker);
   }
 }
 
-Task* PosixDynamicThreadPool::WaitForTask() {
+PosixDynamicThreadPool::PendingTask PosixDynamicThreadPool::WaitForTask() {
   AutoLock locked(lock_);
 
   if (terminated_)
-    return NULL;
+    return PendingTask(FROM_HERE, base::Closure());
 
-  if (tasks_.empty()) {  // No work available, wait for work.
+  if (pending_tasks_.empty()) {  // No work available, wait for work.
     num_idle_threads_++;
     if (num_idle_threads_cv_.get())
       num_idle_threads_cv_->Signal();
-    tasks_available_cv_.TimedWait(
-        TimeDelta::FromSeconds(kIdleSecondsBeforeExit));
+    pending_tasks_available_cv_.TimedWait(
+        TimeDelta::FromSeconds(idle_seconds_before_exit_));
     num_idle_threads_--;
     if (num_idle_threads_cv_.get())
       num_idle_threads_cv_->Signal();
-    if (tasks_.empty()) {
+    if (pending_tasks_.empty()) {
       // We waited for work, but there's still no work.  Return NULL to signal
       // the thread to terminate.
-      return NULL;
+      return PendingTask(FROM_HERE, base::Closure());
     }
   }
 
-  Task* task = tasks_.front();
-  tasks_.pop();
-  return task;
+  PendingTask pending_task = pending_tasks_.front();
+  pending_tasks_.pop();
+  return pending_task;
 }
 
 }  // namespace base