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