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