1 // Copyright 2010 The Chromium Authors
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 <memory>
8 #include <ostream>
9
10 #include "base/check.h"
11 #include "base/strings/string_number_conversions.h"
12 #include "base/synchronization/waitable_event.h"
13 #include "base/threading/platform_thread.h"
14 #include "base/threading/thread_restrictions.h"
15
16 namespace base {
17
SimpleThread(const std::string & name)18 SimpleThread::SimpleThread(const std::string& name)
19 : SimpleThread(name, Options()) {}
20
SimpleThread(const std::string & name,const Options & options)21 SimpleThread::SimpleThread(const std::string& name, const Options& options)
22 : name_(name),
23 options_(options),
24 event_(WaitableEvent::ResetPolicy::MANUAL,
25 WaitableEvent::InitialState::NOT_SIGNALED) {}
26
~SimpleThread()27 SimpleThread::~SimpleThread() {
28 DCHECK(HasBeenStarted()) << "SimpleThread was never started.";
29 DCHECK(!options_.joinable || HasBeenJoined())
30 << "Joinable SimpleThread destroyed without being Join()ed.";
31 }
32
Start()33 void SimpleThread::Start() {
34 StartAsync();
35 ScopedAllowBaseSyncPrimitives allow_wait;
36 event_.Wait(); // Wait for the thread to complete initialization.
37 }
38
Join()39 void SimpleThread::Join() {
40 DCHECK(options_.joinable) << "A non-joinable thread can't be joined.";
41 DCHECK(HasStartBeenAttempted()) << "Tried to Join a never-started thread.";
42 DCHECK(!HasBeenJoined()) << "Tried to Join a thread multiple times.";
43 BeforeJoin();
44 PlatformThread::Join(thread_);
45 thread_ = PlatformThreadHandle();
46 joined_ = true;
47 }
48
StartAsync()49 void SimpleThread::StartAsync() {
50 DCHECK(!HasStartBeenAttempted()) << "Tried to Start a thread multiple times.";
51 start_called_ = true;
52 BeforeStart();
53 bool success =
54 options_.joinable
55 ? PlatformThread::CreateWithType(options_.stack_size, this, &thread_,
56 options_.thread_type)
57 : PlatformThread::CreateNonJoinableWithType(options_.stack_size, this,
58 options_.thread_type);
59 CHECK(success);
60 }
61
tid()62 PlatformThreadId SimpleThread::tid() {
63 DCHECK(HasBeenStarted());
64 return tid_;
65 }
66
HasBeenStarted()67 bool SimpleThread::HasBeenStarted() {
68 return event_.IsSignaled();
69 }
70
ThreadMain()71 void SimpleThread::ThreadMain() {
72 tid_ = PlatformThread::CurrentId();
73 PlatformThread::SetName(name_);
74
75 // We've initialized our new thread, signal that we're done to Start().
76 event_.Signal();
77
78 BeforeRun();
79 Run();
80 }
81
DelegateSimpleThread(Delegate * delegate,const std::string & name)82 DelegateSimpleThread::DelegateSimpleThread(Delegate* delegate,
83 const std::string& name)
84 : DelegateSimpleThread(delegate, name, Options()) {}
85
DelegateSimpleThread(Delegate * delegate,const std::string & name,const Options & options)86 DelegateSimpleThread::DelegateSimpleThread(Delegate* delegate,
87 const std::string& name,
88 const Options& options)
89 : SimpleThread(name, options), delegate_(delegate) {
90 DCHECK(delegate_);
91 }
92
93 DelegateSimpleThread::~DelegateSimpleThread() = default;
94
Run()95 void DelegateSimpleThread::Run() {
96 DCHECK(delegate_) << "Tried to call Run without a delegate (called twice?)";
97
98 // Non-joinable DelegateSimpleThreads are allowed to be deleted during Run().
99 // Member state must not be accessed after invoking Run().
100 Delegate* delegate = delegate_;
101 delegate_ = nullptr;
102 delegate->Run();
103 }
104
DelegateSimpleThreadPool(const std::string & name_prefix,size_t num_threads)105 DelegateSimpleThreadPool::DelegateSimpleThreadPool(
106 const std::string& name_prefix,
107 size_t num_threads)
108 : name_prefix_(name_prefix),
109 num_threads_(num_threads),
110 dry_(WaitableEvent::ResetPolicy::MANUAL,
111 WaitableEvent::InitialState::NOT_SIGNALED) {}
112
~DelegateSimpleThreadPool()113 DelegateSimpleThreadPool::~DelegateSimpleThreadPool() {
114 DCHECK(threads_.empty());
115 DCHECK(delegates_.empty());
116 DCHECK(!dry_.IsSignaled());
117 }
118
Start()119 void DelegateSimpleThreadPool::Start() {
120 DCHECK(threads_.empty()) << "Start() called with outstanding threads.";
121 for (size_t i = 0; i < num_threads_; ++i) {
122 std::string name(name_prefix_);
123 name.push_back('/');
124 name.append(NumberToString(i));
125 auto thread = std::make_unique<DelegateSimpleThread>(this, name);
126 thread->Start();
127 threads_.push_back(std::move(thread));
128 }
129 }
130
JoinAll()131 void DelegateSimpleThreadPool::JoinAll() {
132 DCHECK(!threads_.empty()) << "JoinAll() called with no outstanding threads.";
133
134 // Tell all our threads to quit their worker loop.
135 AddWork(nullptr, num_threads_);
136
137 // Join and destroy all the worker threads.
138 for (size_t i = 0; i < num_threads_; ++i) {
139 threads_[i]->Join();
140 }
141 threads_.clear();
142 DCHECK(delegates_.empty());
143 }
144
AddWork(Delegate * delegate,size_t repeat_count)145 void DelegateSimpleThreadPool::AddWork(Delegate* delegate,
146 size_t repeat_count) {
147 AutoLock locked(lock_);
148 for (size_t i = 0; i < repeat_count; ++i)
149 delegates_.push(delegate);
150 // If we were empty, signal that we have work now.
151 if (!dry_.IsSignaled())
152 dry_.Signal();
153 }
154
Run()155 void DelegateSimpleThreadPool::Run() {
156 Delegate* work = nullptr;
157
158 while (true) {
159 dry_.Wait();
160 {
161 AutoLock locked(lock_);
162 if (!dry_.IsSignaled())
163 continue;
164
165 DCHECK(!delegates_.empty());
166 work = delegates_.front();
167 delegates_.pop();
168
169 // Signal to any other threads that we're currently out of work.
170 if (delegates_.empty())
171 dry_.Reset();
172 }
173
174 // A NULL delegate pointer signals us to quit.
175 if (!work)
176 break;
177
178 work->Run();
179 }
180 }
181
182 } // namespace base
183