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