base/threading/simple_thread.cc

   1 // Copyright (c) 2010 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/simple_thread.h"
   6
   7 #include "base/logging.h"
   8 #include "base/strings/string_number_conversions.h"
   9 #include "base/threading/platform_thread.h"
  10 #include "base/threading/thread_restrictions.h"
  11
  12 namespace base {
  13
  14 SimpleThread::SimpleThread(const std::string& name_prefix)
  15     : name_prefix_(name_prefix), name_(name_prefix),
  16       thread_(), event_(true, false), tid_(0), joined_(false) {
  17 }
  18
  19 SimpleThread::SimpleThread(const std::string& name_prefix,
  20                            const Options& options)
  21     : name_prefix_(name_prefix), name_(name_prefix), options_(options),
  22       thread_(), event_(true, false), tid_(0), joined_(false) {
  23 }
  24
  25 SimpleThread::~SimpleThread() {
  26   DCHECK(HasBeenStarted()) << "SimpleThread was never started.";
  27   DCHECK(HasBeenJoined()) << "SimpleThread destroyed without being Join()ed.";
  28 }
  29
  30 void SimpleThread::Start() {
  31   DCHECK(!HasBeenStarted()) << "Tried to Start a thread multiple times.";
  32   bool success = PlatformThread::Create(options_.stack_size(), this, &thread_);
  33   DCHECK(success);
  34   base::ThreadRestrictions::ScopedAllowWait allow_wait;
  35   event_.Wait();  // Wait for the thread to complete initialization.
  36 }
  37
  38 void SimpleThread::Join() {
  39   DCHECK(HasBeenStarted()) << "Tried to Join a never-started thread.";
  40   DCHECK(!HasBeenJoined()) << "Tried to Join a thread multiple times.";
  41   PlatformThread::Join(thread_);
  42   joined_ = true;
  43 }
  44
  45 bool SimpleThread::HasBeenStarted() {
  46   base::ThreadRestrictions::ScopedAllowWait allow_wait;
  47   return event_.IsSignaled();
  48 }
  49
  50 void SimpleThread::ThreadMain() {
  51   tid_ = PlatformThread::CurrentId();
  52   // Construct our full name of the form "name_prefix_/TID".
  53   name_.push_back('/');
  54   name_.append(IntToString(tid_));
  55   PlatformThread::SetName(name_);
  56
  57   // We've initialized our new thread, signal that we're done to Start().
  58   event_.Signal();
  59
  60   Run();
  61 }
  62
  63 DelegateSimpleThread::DelegateSimpleThread(Delegate* delegate,
  64                                            const std::string& name_prefix)
  65     : SimpleThread(name_prefix),
  66       delegate_(delegate) {
  67 }
  68
  69 DelegateSimpleThread::DelegateSimpleThread(Delegate* delegate,
  70                                            const std::string& name_prefix,
  71                                            const Options& options)
  72     : SimpleThread(name_prefix, options),
  73       delegate_(delegate) {
  74 }
  75
  76 DelegateSimpleThread::~DelegateSimpleThread() {
  77 }
  78
  79 void DelegateSimpleThread::Run() {
  80   DCHECK(delegate_) << "Tried to call Run without a delegate (called twice?)";
  81   delegate_->Run();
  82   delegate_ = NULL;
  83 }
  84
  85 DelegateSimpleThreadPool::DelegateSimpleThreadPool(
  86     const std::string& name_prefix,
  87     int num_threads)
  88     : name_prefix_(name_prefix),
  89       num_threads_(num_threads),
  90       dry_(true, false) {
  91 }
  92
  93 DelegateSimpleThreadPool::~DelegateSimpleThreadPool() {
  94   DCHECK(threads_.empty());
  95   DCHECK(delegates_.empty());
  96   DCHECK(!dry_.IsSignaled());
  97 }
  98
  99 void DelegateSimpleThreadPool::Start() {
 100   DCHECK(threads_.empty()) << "Start() called with outstanding threads.";
 101   for (int i = 0; i < num_threads_; ++i) {
 102     DelegateSimpleThread* thread = new DelegateSimpleThread(this, name_prefix_);
 103     thread->Start();
 104     threads_.push_back(thread);
 105   }
 106 }
 107
 108 void DelegateSimpleThreadPool::JoinAll() {
 109   DCHECK(!threads_.empty()) << "JoinAll() called with no outstanding threads.";
 110
 111   // Tell all our threads to quit their worker loop.
 112   AddWork(NULL, num_threads_);
 113
 114   // Join and destroy all the worker threads.
 115   for (int i = 0; i < num_threads_; ++i) {
 116     threads_[i]->Join();
 117     delete threads_[i];
 118   }
 119   threads_.clear();
 120   DCHECK(delegates_.empty());
 121 }
 122
 123 void DelegateSimpleThreadPool::AddWork(Delegate* delegate, int repeat_count) {
 124   AutoLock locked(lock_);
 125   for (int i = 0; i < repeat_count; ++i)
 126     delegates_.push(delegate);
 127   // If we were empty, signal that we have work now.
 128   if (!dry_.IsSignaled())
 129     dry_.Signal();
 130 }
 131
 132 void DelegateSimpleThreadPool::Run() {
 133   Delegate* work = NULL;
 134
 135   while (true) {
 136     dry_.Wait();
 137     {
 138       AutoLock locked(lock_);
 139       if (!dry_.IsSignaled())
 140         continue;
 141
 142       DCHECK(!delegates_.empty());
 143       work = delegates_.front();
 144       delegates_.pop();
 145
 146       // Signal to any other threads that we're currently out of work.
 147       if (delegates_.empty())
 148         dry_.Reset();
 149     }
 150
 151     // A NULL delegate pointer signals us to quit.
 152     if (!work)
 153       break;
 154
 155     work->Run();
 156   }
 157 }
 158
 159 }  // namespace base