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
43 #if defined(__BIONIC__)
44 static constexpr bool kUseCustomThreadPoolStack = false;
45 #else
46 static constexpr bool kUseCustomThreadPoolStack = true;
47 #endif
48
ThreadPoolWorker(ThreadPool * thread_pool,const std::string & name,size_t stack_size)49 ThreadPoolWorker::ThreadPoolWorker(ThreadPool* thread_pool, const std::string& name,
50 size_t stack_size)
51 : thread_pool_(thread_pool),
52 name_(name) {
53 std::string error_msg;
54 // On Bionic, we know pthreads will give us a big-enough stack with
55 // a guard page, so don't do anything special on Bionic libc.
56 if (kUseCustomThreadPoolStack) {
57 // Add an inaccessible page to catch stack overflow.
58 stack_size += kPageSize;
59 stack_ = MemMap::MapAnonymous(name.c_str(),
60 stack_size,
61 PROT_READ | PROT_WRITE,
62 /*low_4gb=*/ false,
63 &error_msg);
64 CHECK(stack_.IsValid()) << error_msg;
65 CHECK_ALIGNED(stack_.Begin(), kPageSize);
66 CheckedCall(mprotect,
67 "mprotect bottom page of thread pool worker stack",
68 stack_.Begin(),
69 kPageSize,
70 PROT_NONE);
71 }
72 const char* reason = "new thread pool worker thread";
73 pthread_attr_t attr;
74 CHECK_PTHREAD_CALL(pthread_attr_init, (&attr), reason);
75 if (kUseCustomThreadPoolStack) {
76 CHECK_PTHREAD_CALL(pthread_attr_setstack, (&attr, stack_.Begin(), stack_.Size()), reason);
77 } else {
78 CHECK_PTHREAD_CALL(pthread_attr_setstacksize, (&attr, stack_size), reason);
79 }
80 CHECK_PTHREAD_CALL(pthread_create, (&pthread_, &attr, &Callback, this), reason);
81 CHECK_PTHREAD_CALL(pthread_attr_destroy, (&attr), reason);
82 }
83
~ThreadPoolWorker()84 ThreadPoolWorker::~ThreadPoolWorker() {
85 CHECK_PTHREAD_CALL(pthread_join, (pthread_, nullptr), "thread pool worker shutdown");
86 }
87
SetPthreadPriority(int priority)88 void ThreadPoolWorker::SetPthreadPriority(int priority) {
89 CHECK_GE(priority, PRIO_MIN);
90 CHECK_LE(priority, PRIO_MAX);
91 #if defined(ART_TARGET_ANDROID)
92 int result = setpriority(PRIO_PROCESS, pthread_gettid_np(pthread_), priority);
93 if (result != 0) {
94 PLOG(ERROR) << "Failed to setpriority to :" << priority;
95 }
96 #else
97 UNUSED(priority);
98 #endif
99 }
100
GetPthreadPriority()101 int ThreadPoolWorker::GetPthreadPriority() {
102 #if defined(ART_TARGET_ANDROID)
103 return getpriority(PRIO_PROCESS, pthread_gettid_np(pthread_));
104 #else
105 return 0;
106 #endif
107 }
108
Run()109 void ThreadPoolWorker::Run() {
110 Thread* self = Thread::Current();
111 Task* task = nullptr;
112 thread_pool_->creation_barier_.Pass(self);
113 while ((task = thread_pool_->GetTask(self)) != nullptr) {
114 task->Run(self);
115 task->Finalize();
116 }
117 }
118
Callback(void * arg)119 void* ThreadPoolWorker::Callback(void* arg) {
120 ThreadPoolWorker* worker = reinterpret_cast<ThreadPoolWorker*>(arg);
121 Runtime* runtime = Runtime::Current();
122 CHECK(runtime->AttachCurrentThread(
123 worker->name_.c_str(),
124 true,
125 // Thread-groups are only tracked by the peer j.l.Thread objects. If we aren't creating peers
126 // we don't need to specify the thread group. We want to place these threads in the System
127 // thread group because that thread group is where important threads that debuggers and
128 // similar tools should not mess with are placed. As this is an internal-thread-pool we might
129 // rely on being able to (for example) wait for all threads to finish some task. If debuggers
130 // are suspending these threads that might not be possible.
131 worker->thread_pool_->create_peers_ ? runtime->GetSystemThreadGroup() : nullptr,
132 worker->thread_pool_->create_peers_));
133 worker->thread_ = Thread::Current();
134 // Mark thread pool workers as runtime-threads.
135 worker->thread_->SetIsRuntimeThread(true);
136 // Do work until its time to shut down.
137 worker->Run();
138 runtime->DetachCurrentThread();
139 return nullptr;
140 }
141
AddTask(Thread * self,Task * task)142 void ThreadPool::AddTask(Thread* self, Task* task) {
143 MutexLock mu(self, task_queue_lock_);
144 tasks_.push_back(task);
145 // If we have any waiters, signal one.
146 if (started_ && waiting_count_ != 0) {
147 task_queue_condition_.Signal(self);
148 }
149 }
150
RemoveAllTasks(Thread * self)151 void ThreadPool::RemoveAllTasks(Thread* self) {
152 // The ThreadPool is responsible for calling Finalize (which usually delete
153 // the task memory) on all the tasks.
154 Task* task = nullptr;
155 while ((task = TryGetTask(self)) != nullptr) {
156 task->Finalize();
157 }
158 MutexLock mu(self, task_queue_lock_);
159 tasks_.clear();
160 }
161
ThreadPool(const char * name,size_t num_threads,bool create_peers,size_t worker_stack_size)162 ThreadPool::ThreadPool(const char* name,
163 size_t num_threads,
164 bool create_peers,
165 size_t worker_stack_size)
166 : name_(name),
167 task_queue_lock_("task queue lock", kGenericBottomLock),
168 task_queue_condition_("task queue condition", task_queue_lock_),
169 completion_condition_("task completion condition", task_queue_lock_),
170 started_(false),
171 shutting_down_(false),
172 waiting_count_(0),
173 start_time_(0),
174 total_wait_time_(0),
175 creation_barier_(0),
176 max_active_workers_(num_threads),
177 create_peers_(create_peers),
178 worker_stack_size_(worker_stack_size) {
179 CreateThreads();
180 }
181
CreateThreads()182 void ThreadPool::CreateThreads() {
183 CHECK(threads_.empty());
184 Thread* self = Thread::Current();
185 {
186 MutexLock mu(self, task_queue_lock_);
187 shutting_down_ = false;
188 // Add one since the caller of constructor waits on the barrier too.
189 creation_barier_.Init(self, max_active_workers_);
190 while (GetThreadCount() < max_active_workers_) {
191 const std::string worker_name = StringPrintf("%s worker thread %zu", name_.c_str(),
192 GetThreadCount());
193 threads_.push_back(
194 new ThreadPoolWorker(this, worker_name, worker_stack_size_));
195 }
196 }
197 }
198
WaitForWorkersToBeCreated()199 void ThreadPool::WaitForWorkersToBeCreated() {
200 creation_barier_.Increment(Thread::Current(), 0);
201 }
202
GetWorkers()203 const std::vector<ThreadPoolWorker*>& ThreadPool::GetWorkers() {
204 // Wait for all the workers to be created before returning them.
205 WaitForWorkersToBeCreated();
206 return threads_;
207 }
208
DeleteThreads()209 void ThreadPool::DeleteThreads() {
210 {
211 Thread* self = Thread::Current();
212 MutexLock mu(self, task_queue_lock_);
213 // Tell any remaining workers to shut down.
214 shutting_down_ = true;
215 // Broadcast to everyone waiting.
216 task_queue_condition_.Broadcast(self);
217 completion_condition_.Broadcast(self);
218 }
219 // Wait for the threads to finish. We expect the user of the pool
220 // not to run multi-threaded calls to `CreateThreads` and `DeleteThreads`,
221 // so we don't guard the field here.
222 STLDeleteElements(&threads_);
223 }
224
SetMaxActiveWorkers(size_t max_workers)225 void ThreadPool::SetMaxActiveWorkers(size_t max_workers) {
226 MutexLock mu(Thread::Current(), task_queue_lock_);
227 CHECK_LE(max_workers, GetThreadCount());
228 max_active_workers_ = max_workers;
229 }
230
~ThreadPool()231 ThreadPool::~ThreadPool() {
232 DeleteThreads();
233 RemoveAllTasks(Thread::Current());
234 }
235
StartWorkers(Thread * self)236 void ThreadPool::StartWorkers(Thread* self) {
237 MutexLock mu(self, task_queue_lock_);
238 started_ = true;
239 task_queue_condition_.Broadcast(self);
240 start_time_ = NanoTime();
241 total_wait_time_ = 0;
242 }
243
StopWorkers(Thread * self)244 void ThreadPool::StopWorkers(Thread* self) {
245 MutexLock mu(self, task_queue_lock_);
246 started_ = false;
247 }
248
GetTask(Thread * self)249 Task* ThreadPool::GetTask(Thread* self) {
250 MutexLock mu(self, task_queue_lock_);
251 while (!IsShuttingDown()) {
252 const size_t thread_count = GetThreadCount();
253 // Ensure that we don't use more threads than the maximum active workers.
254 const size_t active_threads = thread_count - waiting_count_;
255 // <= since self is considered an active worker.
256 if (active_threads <= max_active_workers_) {
257 Task* task = TryGetTaskLocked();
258 if (task != nullptr) {
259 return task;
260 }
261 }
262
263 ++waiting_count_;
264 if (waiting_count_ == GetThreadCount() && !HasOutstandingTasks()) {
265 // We may be done, lets broadcast to the completion condition.
266 completion_condition_.Broadcast(self);
267 }
268 const uint64_t wait_start = kMeasureWaitTime ? NanoTime() : 0;
269 task_queue_condition_.Wait(self);
270 if (kMeasureWaitTime) {
271 const uint64_t wait_end = NanoTime();
272 total_wait_time_ += wait_end - std::max(wait_start, start_time_);
273 }
274 --waiting_count_;
275 }
276
277 // We are shutting down, return null to tell the worker thread to stop looping.
278 return nullptr;
279 }
280
TryGetTask(Thread * self)281 Task* ThreadPool::TryGetTask(Thread* self) {
282 MutexLock mu(self, task_queue_lock_);
283 return TryGetTaskLocked();
284 }
285
TryGetTaskLocked()286 Task* ThreadPool::TryGetTaskLocked() {
287 if (HasOutstandingTasks()) {
288 Task* task = tasks_.front();
289 tasks_.pop_front();
290 return task;
291 }
292 return nullptr;
293 }
294
Wait(Thread * self,bool do_work,bool may_hold_locks)295 void ThreadPool::Wait(Thread* self, bool do_work, bool may_hold_locks) {
296 if (do_work) {
297 CHECK(!create_peers_);
298 Task* task = nullptr;
299 while ((task = TryGetTask(self)) != nullptr) {
300 task->Run(self);
301 task->Finalize();
302 }
303 }
304 // Wait until each thread is waiting and the task list is empty.
305 MutexLock mu(self, task_queue_lock_);
306 while (!shutting_down_ && (waiting_count_ != GetThreadCount() || HasOutstandingTasks())) {
307 if (!may_hold_locks) {
308 completion_condition_.Wait(self);
309 } else {
310 completion_condition_.WaitHoldingLocks(self);
311 }
312 }
313 }
314
GetTaskCount(Thread * self)315 size_t ThreadPool::GetTaskCount(Thread* self) {
316 MutexLock mu(self, task_queue_lock_);
317 return tasks_.size();
318 }
319
SetPthreadPriority(int priority)320 void ThreadPool::SetPthreadPriority(int priority) {
321 for (ThreadPoolWorker* worker : threads_) {
322 worker->SetPthreadPriority(priority);
323 }
324 }
325
CheckPthreadPriority(int priority)326 void ThreadPool::CheckPthreadPriority(int priority) {
327 #if defined(ART_TARGET_ANDROID)
328 for (ThreadPoolWorker* worker : threads_) {
329 CHECK_EQ(worker->GetPthreadPriority(), priority);
330 }
331 #else
332 UNUSED(priority);
333 #endif
334 }
335
336 } // namespace art
337