base/threading/worker_pool_posix.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "base/threading/worker_pool_posix.h"
   6
   7 #include "base/bind.h"
   8 #include "base/callback.h"
   9 #include "base/lazy_instance.h"
  10 #include "base/logging.h"
  11 #include "base/memory/ref_counted.h"
  12 #include "base/strings/stringprintf.h"
  13 #include "base/threading/platform_thread.h"
  14 #include "base/threading/thread_local.h"
  15 #include "base/threading/worker_pool.h"
  16 #include "base/trace_event/trace_event.h"
  17 #include "base/tracked_objects.h"
  18
  19 using tracked_objects::TrackedTime;
  20
  21 namespace base {
  22
  23 namespace {
  24
  25 base::LazyInstance<ThreadLocalBoolean>::Leaky
  26     g_worker_pool_running_on_this_thread = LAZY_INSTANCE_INITIALIZER;
  27
  28 const int kIdleSecondsBeforeExit = 10 * 60;
  29
  30 class WorkerPoolImpl {
  31  public:
  32   WorkerPoolImpl();
  33   ~WorkerPoolImpl();
  34
  35   void PostTask(const tracked_objects::Location& from_here,
  36                 const base::Closure& task, bool task_is_slow);
  37
  38  private:
  39   scoped_refptr<base::PosixDynamicThreadPool> pool_;
  40 };
  41
  42 WorkerPoolImpl::WorkerPoolImpl()
  43     : pool_(new base::PosixDynamicThreadPool("WorkerPool",
  44                                              kIdleSecondsBeforeExit)) {
  45 }
  46
  47 WorkerPoolImpl::~WorkerPoolImpl() {
  48   pool_->Terminate();
  49 }
  50
  51 void WorkerPoolImpl::PostTask(const tracked_objects::Location& from_here,
  52                               const base::Closure& task, bool task_is_slow) {
  53   pool_->PostTask(from_here, task);
  54 }
  55
  56 base::LazyInstance<WorkerPoolImpl> g_lazy_worker_pool =
  57     LAZY_INSTANCE_INITIALIZER;
  58
  59 class WorkerThread : public PlatformThread::Delegate {
  60  public:
  61   WorkerThread(const std::string& name_prefix,
  62                base::PosixDynamicThreadPool* pool)
  63       : name_prefix_(name_prefix),
  64         pool_(pool) {}
  65
  66   void ThreadMain() override;
  67
  68  private:
  69   const std::string name_prefix_;
  70   scoped_refptr<base::PosixDynamicThreadPool> pool_;
  71
  72   DISALLOW_COPY_AND_ASSIGN(WorkerThread);
  73 };
  74
  75 void WorkerThread::ThreadMain() {
  76   g_worker_pool_running_on_this_thread.Get().Set(true);
  77   const std::string name = base::StringPrintf(
  78       "%s/%d", name_prefix_.c_str(), PlatformThread::CurrentId());
  79   // Note |name.c_str()| must remain valid for for the whole life of the thread.
  80   PlatformThread::SetName(name.c_str());
  81
  82   for (;;) {
  83     PendingTask pending_task = pool_->WaitForTask();
  84     if (pending_task.task.is_null())
  85       break;
  86     TRACE_EVENT2("toplevel", "WorkerThread::ThreadMain::Run",
  87         "src_file", pending_task.posted_from.file_name(),
  88         "src_func", pending_task.posted_from.function_name());
  89
  90     tracked_objects::ThreadData::PrepareForStartOfRun(pending_task.birth_tally);
  91     tracked_objects::TaskStopwatch stopwatch;
  92     stopwatch.Start();
  93     pending_task.task.Run();
  94     stopwatch.Stop();
  95
  96     tracked_objects::ThreadData::TallyRunOnWorkerThreadIfTracking(
  97         pending_task.birth_tally, pending_task.time_posted, stopwatch);
  98   }
  99
 100   // The WorkerThread is non-joinable, so it deletes itself.
 101   delete this;
 102 }
 103
 104 }  // namespace
 105
 106 // static
 107 bool WorkerPool::PostTask(const tracked_objects::Location& from_here,
 108                           const base::Closure& task, bool task_is_slow) {
 109   g_lazy_worker_pool.Pointer()->PostTask(from_here, task, task_is_slow);
 110   return true;
 111 }
 112
 113 // static
 114 bool WorkerPool::RunsTasksOnCurrentThread() {
 115   return g_worker_pool_running_on_this_thread.Get().Get();
 116 }
 117
 118 PosixDynamicThreadPool::PosixDynamicThreadPool(const std::string& name_prefix,
 119                                                int idle_seconds_before_exit)
 120     : name_prefix_(name_prefix),
 121       idle_seconds_before_exit_(idle_seconds_before_exit),
 122       pending_tasks_available_cv_(&lock_),
 123       num_idle_threads_(0),
 124       terminated_(false) {}
 125
 126 PosixDynamicThreadPool::~PosixDynamicThreadPool() {
 127   while (!pending_tasks_.empty())
 128     pending_tasks_.pop();
 129 }
 130
 131 void PosixDynamicThreadPool::Terminate() {
 132   {
 133     AutoLock locked(lock_);
 134     DCHECK(!terminated_) << "Thread pool is already terminated.";
 135     terminated_ = true;
 136   }
 137   pending_tasks_available_cv_.Broadcast();
 138 }
 139
 140 void PosixDynamicThreadPool::PostTask(
 141     const tracked_objects::Location& from_here,
 142     const base::Closure& task) {
 143   PendingTask pending_task(from_here, task);
 144   AddTask(&pending_task);
 145 }
 146
 147 void PosixDynamicThreadPool::AddTask(PendingTask* pending_task) {
 148   AutoLock locked(lock_);
 149   DCHECK(!terminated_) <<
 150       "This thread pool is already terminated.  Do not post new tasks.";
 151
 152   pending_tasks_.push(*pending_task);
 153   pending_task->task.Reset();
 154
 155   // We have enough worker threads.
 156   if (static_cast<size_t>(num_idle_threads_) >= pending_tasks_.size()) {
 157     pending_tasks_available_cv_.Signal();
 158   } else {
 159     // The new PlatformThread will take ownership of the WorkerThread object,
 160     // which will delete itself on exit.
 161     WorkerThread* worker =
 162         new WorkerThread(name_prefix_, this);
 163     PlatformThread::CreateNonJoinable(0, worker);
 164   }
 165 }
 166
 167 PendingTask PosixDynamicThreadPool::WaitForTask() {
 168   AutoLock locked(lock_);
 169
 170   if (terminated_)
 171     return PendingTask(FROM_HERE, base::Closure());
 172
 173   if (pending_tasks_.empty()) {  // No work available, wait for work.
 174     num_idle_threads_++;
 175     if (num_idle_threads_cv_.get())
 176       num_idle_threads_cv_->Signal();
 177     pending_tasks_available_cv_.TimedWait(
 178         TimeDelta::FromSeconds(idle_seconds_before_exit_));
 179     num_idle_threads_--;
 180     if (num_idle_threads_cv_.get())
 181       num_idle_threads_cv_->Signal();
 182     if (pending_tasks_.empty()) {
 183       // We waited for work, but there's still no work.  Return NULL to signal
 184       // the thread to terminate.
 185       return PendingTask(FROM_HERE, base::Closure());
 186     }
 187   }
 188
 189   PendingTask pending_task = pending_tasks_.front();
 190   pending_tasks_.pop();
 191   return pending_task;
 192 }
 193
 194 }  // namespace base