1
2 /*
3 * Copyright (C) 2012 The Android Open Source Project
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 */
17
18 #include "thread_pool.h"
19
20 #include <sys/mman.h>
21 #include <sys/resource.h>
22 #include <sys/time.h>
23
24 #include <pthread.h>
25
26 #include <android-base/logging.h>
27 #include <android-base/stringprintf.h>
28
29 #include "base/bit_utils.h"
30 #include "base/casts.h"
31 #include "base/stl_util.h"
32 #include "base/time_utils.h"
33 #include "base/utils.h"
34 #include "runtime.h"
35 #include "thread-current-inl.h"
36
37 namespace art {
38
39 using android::base::StringPrintf;
40
41 static constexpr bool kMeasureWaitTime = false;
42
ThreadPoolWorker(ThreadPool * thread_pool,const std::string & name,size_t stack_size)43 ThreadPoolWorker::ThreadPoolWorker(ThreadPool* thread_pool, const std::string& name,
44 size_t stack_size)
45 : thread_pool_(thread_pool),
46 name_(name) {
47 // Add an inaccessible page to catch stack overflow.
48 stack_size += kPageSize;
49 std::string error_msg;
50 stack_ = MemMap::MapAnonymous(name.c_str(),
51 stack_size,
52 PROT_READ | PROT_WRITE,
53 /*low_4gb=*/ false,
54 &error_msg);
55 CHECK(stack_.IsValid()) << error_msg;
56 CHECK_ALIGNED(stack_.Begin(), kPageSize);
57 CheckedCall(mprotect,
58 "mprotect bottom page of thread pool worker stack",
59 stack_.Begin(),
60 kPageSize,
61 PROT_NONE);
62 const char* reason = "new thread pool worker thread";
63 pthread_attr_t attr;
64 CHECK_PTHREAD_CALL(pthread_attr_init, (&attr), reason);
65 CHECK_PTHREAD_CALL(pthread_attr_setstack, (&attr, stack_.Begin(), stack_.Size()), reason);
66 CHECK_PTHREAD_CALL(pthread_create, (&pthread_, &attr, &Callback, this), reason);
67 CHECK_PTHREAD_CALL(pthread_attr_destroy, (&attr), reason);
68 }
69
~ThreadPoolWorker()70 ThreadPoolWorker::~ThreadPoolWorker() {
71 CHECK_PTHREAD_CALL(pthread_join, (pthread_, nullptr), "thread pool worker shutdown");
72 }
73
SetPthreadPriority(int priority)74 void ThreadPoolWorker::SetPthreadPriority(int priority) {
75 CHECK_GE(priority, PRIO_MIN);
76 CHECK_LE(priority, PRIO_MAX);
77 #if defined(ART_TARGET_ANDROID)
78 int result = setpriority(PRIO_PROCESS, pthread_gettid_np(pthread_), priority);
79 if (result != 0) {
80 PLOG(ERROR) << "Failed to setpriority to :" << priority;
81 }
82 #else
83 UNUSED(priority);
84 #endif
85 }
86
Run()87 void ThreadPoolWorker::Run() {
88 Thread* self = Thread::Current();
89 Task* task = nullptr;
90 thread_pool_->creation_barier_.Pass(self);
91 while ((task = thread_pool_->GetTask(self)) != nullptr) {
92 task->Run(self);
93 task->Finalize();
94 }
95 }
96
Callback(void * arg)97 void* ThreadPoolWorker::Callback(void* arg) {
98 ThreadPoolWorker* worker = reinterpret_cast<ThreadPoolWorker*>(arg);
99 Runtime* runtime = Runtime::Current();
100 CHECK(runtime->AttachCurrentThread(worker->name_.c_str(),
101 true,
102 nullptr,
103 worker->thread_pool_->create_peers_));
104 worker->thread_ = Thread::Current();
105 // Mark thread pool workers as runtime-threads.
106 worker->thread_->SetIsRuntimeThread(true);
107 // Do work until its time to shut down.
108 worker->Run();
109 runtime->DetachCurrentThread();
110 return nullptr;
111 }
112
AddTask(Thread * self,Task * task)113 void ThreadPool::AddTask(Thread* self, Task* task) {
114 MutexLock mu(self, task_queue_lock_);
115 tasks_.push_back(task);
116 // If we have any waiters, signal one.
117 if (started_ && waiting_count_ != 0) {
118 task_queue_condition_.Signal(self);
119 }
120 }
121
RemoveAllTasks(Thread * self)122 void ThreadPool::RemoveAllTasks(Thread* self) {
123 MutexLock mu(self, task_queue_lock_);
124 tasks_.clear();
125 }
126
ThreadPool(const char * name,size_t num_threads,bool create_peers,size_t worker_stack_size)127 ThreadPool::ThreadPool(const char* name,
128 size_t num_threads,
129 bool create_peers,
130 size_t worker_stack_size)
131 : name_(name),
132 task_queue_lock_("task queue lock"),
133 task_queue_condition_("task queue condition", task_queue_lock_),
134 completion_condition_("task completion condition", task_queue_lock_),
135 started_(false),
136 shutting_down_(false),
137 waiting_count_(0),
138 start_time_(0),
139 total_wait_time_(0),
140 creation_barier_(0),
141 max_active_workers_(num_threads),
142 create_peers_(create_peers),
143 worker_stack_size_(worker_stack_size) {
144 CreateThreads();
145 }
146
CreateThreads()147 void ThreadPool::CreateThreads() {
148 CHECK(threads_.empty());
149 Thread* self = Thread::Current();
150 {
151 MutexLock mu(self, task_queue_lock_);
152 shutting_down_ = false;
153 // Add one since the caller of constructor waits on the barrier too.
154 creation_barier_.Init(self, max_active_workers_);
155 while (GetThreadCount() < max_active_workers_) {
156 const std::string worker_name = StringPrintf("%s worker thread %zu", name_.c_str(),
157 GetThreadCount());
158 threads_.push_back(
159 new ThreadPoolWorker(this, worker_name, worker_stack_size_));
160 }
161 }
162 }
163
WaitForWorkersToBeCreated()164 void ThreadPool::WaitForWorkersToBeCreated() {
165 creation_barier_.Increment(Thread::Current(), 0);
166 }
167
GetWorkers()168 const std::vector<ThreadPoolWorker*>& ThreadPool::GetWorkers() {
169 // Wait for all the workers to be created before returning them.
170 WaitForWorkersToBeCreated();
171 return threads_;
172 }
173
DeleteThreads()174 void ThreadPool::DeleteThreads() {
175 {
176 Thread* self = Thread::Current();
177 MutexLock mu(self, task_queue_lock_);
178 // Tell any remaining workers to shut down.
179 shutting_down_ = true;
180 // Broadcast to everyone waiting.
181 task_queue_condition_.Broadcast(self);
182 completion_condition_.Broadcast(self);
183 }
184 // Wait for the threads to finish. We expect the user of the pool
185 // not to run multi-threaded calls to `CreateThreads` and `DeleteThreads`,
186 // so we don't guard the field here.
187 STLDeleteElements(&threads_);
188 }
189
SetMaxActiveWorkers(size_t max_workers)190 void ThreadPool::SetMaxActiveWorkers(size_t max_workers) {
191 MutexLock mu(Thread::Current(), task_queue_lock_);
192 CHECK_LE(max_workers, GetThreadCount());
193 max_active_workers_ = max_workers;
194 }
195
~ThreadPool()196 ThreadPool::~ThreadPool() {
197 DeleteThreads();
198 }
199
StartWorkers(Thread * self)200 void ThreadPool::StartWorkers(Thread* self) {
201 MutexLock mu(self, task_queue_lock_);
202 started_ = true;
203 task_queue_condition_.Broadcast(self);
204 start_time_ = NanoTime();
205 total_wait_time_ = 0;
206 }
207
StopWorkers(Thread * self)208 void ThreadPool::StopWorkers(Thread* self) {
209 MutexLock mu(self, task_queue_lock_);
210 started_ = false;
211 }
212
GetTask(Thread * self)213 Task* ThreadPool::GetTask(Thread* self) {
214 MutexLock mu(self, task_queue_lock_);
215 while (!IsShuttingDown()) {
216 const size_t thread_count = GetThreadCount();
217 // Ensure that we don't use more threads than the maximum active workers.
218 const size_t active_threads = thread_count - waiting_count_;
219 // <= since self is considered an active worker.
220 if (active_threads <= max_active_workers_) {
221 Task* task = TryGetTaskLocked();
222 if (task != nullptr) {
223 return task;
224 }
225 }
226
227 ++waiting_count_;
228 if (waiting_count_ == GetThreadCount() && !HasOutstandingTasks()) {
229 // We may be done, lets broadcast to the completion condition.
230 completion_condition_.Broadcast(self);
231 }
232 const uint64_t wait_start = kMeasureWaitTime ? NanoTime() : 0;
233 task_queue_condition_.Wait(self);
234 if (kMeasureWaitTime) {
235 const uint64_t wait_end = NanoTime();
236 total_wait_time_ += wait_end - std::max(wait_start, start_time_);
237 }
238 --waiting_count_;
239 }
240
241 // We are shutting down, return null to tell the worker thread to stop looping.
242 return nullptr;
243 }
244
TryGetTask(Thread * self)245 Task* ThreadPool::TryGetTask(Thread* self) {
246 MutexLock mu(self, task_queue_lock_);
247 return TryGetTaskLocked();
248 }
249
TryGetTaskLocked()250 Task* ThreadPool::TryGetTaskLocked() {
251 if (HasOutstandingTasks()) {
252 Task* task = tasks_.front();
253 tasks_.pop_front();
254 return task;
255 }
256 return nullptr;
257 }
258
Wait(Thread * self,bool do_work,bool may_hold_locks)259 void ThreadPool::Wait(Thread* self, bool do_work, bool may_hold_locks) {
260 if (do_work) {
261 CHECK(!create_peers_);
262 Task* task = nullptr;
263 while ((task = TryGetTask(self)) != nullptr) {
264 task->Run(self);
265 task->Finalize();
266 }
267 }
268 // Wait until each thread is waiting and the task list is empty.
269 MutexLock mu(self, task_queue_lock_);
270 while (!shutting_down_ && (waiting_count_ != GetThreadCount() || HasOutstandingTasks())) {
271 if (!may_hold_locks) {
272 completion_condition_.Wait(self);
273 } else {
274 completion_condition_.WaitHoldingLocks(self);
275 }
276 }
277 }
278
GetTaskCount(Thread * self)279 size_t ThreadPool::GetTaskCount(Thread* self) {
280 MutexLock mu(self, task_queue_lock_);
281 return tasks_.size();
282 }
283
SetPthreadPriority(int priority)284 void ThreadPool::SetPthreadPriority(int priority) {
285 for (ThreadPoolWorker* worker : threads_) {
286 worker->SetPthreadPriority(priority);
287 }
288 }
289
290 } // namespace art
291