• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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   // Don't run callbacks for ThreadPoolWorkers. These are created for JITThreadPool and
123   // HeapThreadPool and are purely internal threads of the runtime and we don't need to run
124   // callbacks for the thread attach / detach listeners.
125   // (b/251163712) Calling callbacks for heap thread pool workers causes deadlocks in some libjdwp
126   // tests. Deadlocks happen when a GC thread is attached while libjdwp holds the event handler
127   // lock for an event that triggers an entrypoint update from deopt manager.
128   CHECK(runtime->AttachCurrentThread(
129       worker->name_.c_str(),
130       true,
131       // Thread-groups are only tracked by the peer j.l.Thread objects. If we aren't creating peers
132       // we don't need to specify the thread group. We want to place these threads in the System
133       // thread group because that thread group is where important threads that debuggers and
134       // similar tools should not mess with are placed. As this is an internal-thread-pool we might
135       // rely on being able to (for example) wait for all threads to finish some task. If debuggers
136       // are suspending these threads that might not be possible.
137       worker->thread_pool_->create_peers_ ? runtime->GetSystemThreadGroup() : nullptr,
138       worker->thread_pool_->create_peers_,
139       /* should_run_callbacks= */ false));
140   worker->thread_ = Thread::Current();
141   // Mark thread pool workers as runtime-threads.
142   worker->thread_->SetIsRuntimeThread(true);
143   // Do work until its time to shut down.
144   worker->Run();
145   runtime->DetachCurrentThread(/* should_run_callbacks= */ false);
146   return nullptr;
147 }
148 
AddTask(Thread * self,Task * task)149 void ThreadPool::AddTask(Thread* self, Task* task) {
150   MutexLock mu(self, task_queue_lock_);
151   tasks_.push_back(task);
152   // If we have any waiters, signal one.
153   if (started_ && waiting_count_ != 0) {
154     task_queue_condition_.Signal(self);
155   }
156 }
157 
RemoveAllTasks(Thread * self)158 void ThreadPool::RemoveAllTasks(Thread* self) {
159   // The ThreadPool is responsible for calling Finalize (which usually delete
160   // the task memory) on all the tasks.
161   Task* task = nullptr;
162   while ((task = TryGetTask(self)) != nullptr) {
163     task->Finalize();
164   }
165   MutexLock mu(self, task_queue_lock_);
166   tasks_.clear();
167 }
168 
ThreadPool(const char * name,size_t num_threads,bool create_peers,size_t worker_stack_size)169 ThreadPool::ThreadPool(const char* name,
170                        size_t num_threads,
171                        bool create_peers,
172                        size_t worker_stack_size)
173   : name_(name),
174     task_queue_lock_("task queue lock", kGenericBottomLock),
175     task_queue_condition_("task queue condition", task_queue_lock_),
176     completion_condition_("task completion condition", task_queue_lock_),
177     started_(false),
178     shutting_down_(false),
179     waiting_count_(0),
180     start_time_(0),
181     total_wait_time_(0),
182     creation_barier_(0),
183     max_active_workers_(num_threads),
184     create_peers_(create_peers),
185     worker_stack_size_(worker_stack_size) {
186   CreateThreads();
187 }
188 
CreateThreads()189 void ThreadPool::CreateThreads() {
190   CHECK(threads_.empty());
191   Thread* self = Thread::Current();
192   {
193     MutexLock mu(self, task_queue_lock_);
194     shutting_down_ = false;
195     // Add one since the caller of constructor waits on the barrier too.
196     creation_barier_.Init(self, max_active_workers_);
197     while (GetThreadCount() < max_active_workers_) {
198       const std::string worker_name = StringPrintf("%s worker thread %zu", name_.c_str(),
199                                                    GetThreadCount());
200       threads_.push_back(
201           new ThreadPoolWorker(this, worker_name, worker_stack_size_));
202     }
203   }
204 }
205 
WaitForWorkersToBeCreated()206 void ThreadPool::WaitForWorkersToBeCreated() {
207   creation_barier_.Increment(Thread::Current(), 0);
208 }
209 
GetWorkers()210 const std::vector<ThreadPoolWorker*>& ThreadPool::GetWorkers() {
211   // Wait for all the workers to be created before returning them.
212   WaitForWorkersToBeCreated();
213   return threads_;
214 }
215 
DeleteThreads()216 void ThreadPool::DeleteThreads() {
217   {
218     Thread* self = Thread::Current();
219     MutexLock mu(self, task_queue_lock_);
220     // Tell any remaining workers to shut down.
221     shutting_down_ = true;
222     // Broadcast to everyone waiting.
223     task_queue_condition_.Broadcast(self);
224     completion_condition_.Broadcast(self);
225   }
226   // Wait for the threads to finish. We expect the user of the pool
227   // not to run multi-threaded calls to `CreateThreads` and `DeleteThreads`,
228   // so we don't guard the field here.
229   STLDeleteElements(&threads_);
230 }
231 
SetMaxActiveWorkers(size_t max_workers)232 void ThreadPool::SetMaxActiveWorkers(size_t max_workers) {
233   MutexLock mu(Thread::Current(), task_queue_lock_);
234   CHECK_LE(max_workers, GetThreadCount());
235   max_active_workers_ = max_workers;
236 }
237 
~ThreadPool()238 ThreadPool::~ThreadPool() {
239   DeleteThreads();
240   RemoveAllTasks(Thread::Current());
241 }
242 
StartWorkers(Thread * self)243 void ThreadPool::StartWorkers(Thread* self) {
244   MutexLock mu(self, task_queue_lock_);
245   started_ = true;
246   task_queue_condition_.Broadcast(self);
247   start_time_ = NanoTime();
248   total_wait_time_ = 0;
249 }
250 
StopWorkers(Thread * self)251 void ThreadPool::StopWorkers(Thread* self) {
252   MutexLock mu(self, task_queue_lock_);
253   started_ = false;
254 }
255 
HasStarted(Thread * self)256 bool ThreadPool::HasStarted(Thread* self) {
257   MutexLock mu(self, task_queue_lock_);
258   return started_;
259 }
260 
GetTask(Thread * self)261 Task* ThreadPool::GetTask(Thread* self) {
262   MutexLock mu(self, task_queue_lock_);
263   while (!IsShuttingDown()) {
264     const size_t thread_count = GetThreadCount();
265     // Ensure that we don't use more threads than the maximum active workers.
266     const size_t active_threads = thread_count - waiting_count_;
267     // <= since self is considered an active worker.
268     if (active_threads <= max_active_workers_) {
269       Task* task = TryGetTaskLocked();
270       if (task != nullptr) {
271         return task;
272       }
273     }
274 
275     ++waiting_count_;
276     if (waiting_count_ == GetThreadCount() && !HasOutstandingTasks()) {
277       // We may be done, lets broadcast to the completion condition.
278       completion_condition_.Broadcast(self);
279     }
280     const uint64_t wait_start = kMeasureWaitTime ? NanoTime() : 0;
281     task_queue_condition_.Wait(self);
282     if (kMeasureWaitTime) {
283       const uint64_t wait_end = NanoTime();
284       total_wait_time_ += wait_end - std::max(wait_start, start_time_);
285     }
286     --waiting_count_;
287   }
288 
289   // We are shutting down, return null to tell the worker thread to stop looping.
290   return nullptr;
291 }
292 
TryGetTask(Thread * self)293 Task* ThreadPool::TryGetTask(Thread* self) {
294   MutexLock mu(self, task_queue_lock_);
295   return TryGetTaskLocked();
296 }
297 
TryGetTaskLocked()298 Task* ThreadPool::TryGetTaskLocked() {
299   if (HasOutstandingTasks()) {
300     Task* task = tasks_.front();
301     tasks_.pop_front();
302     return task;
303   }
304   return nullptr;
305 }
306 
Wait(Thread * self,bool do_work,bool may_hold_locks)307 void ThreadPool::Wait(Thread* self, bool do_work, bool may_hold_locks) {
308   if (do_work) {
309     CHECK(!create_peers_);
310     Task* task = nullptr;
311     while ((task = TryGetTask(self)) != nullptr) {
312       task->Run(self);
313       task->Finalize();
314     }
315   }
316   // Wait until each thread is waiting and the task list is empty.
317   MutexLock mu(self, task_queue_lock_);
318   while (!shutting_down_ && (waiting_count_ != GetThreadCount() || HasOutstandingTasks())) {
319     if (!may_hold_locks) {
320       completion_condition_.Wait(self);
321     } else {
322       completion_condition_.WaitHoldingLocks(self);
323     }
324   }
325 }
326 
GetTaskCount(Thread * self)327 size_t ThreadPool::GetTaskCount(Thread* self) {
328   MutexLock mu(self, task_queue_lock_);
329   return tasks_.size();
330 }
331 
SetPthreadPriority(int priority)332 void ThreadPool::SetPthreadPriority(int priority) {
333   for (ThreadPoolWorker* worker : threads_) {
334     worker->SetPthreadPriority(priority);
335   }
336 }
337 
CheckPthreadPriority(int priority)338 void ThreadPool::CheckPthreadPriority(int priority) {
339 #if defined(ART_TARGET_ANDROID)
340   for (ThreadPoolWorker* worker : threads_) {
341     CHECK_EQ(worker->GetPthreadPriority(), priority);
342   }
343 #else
344   UNUSED(priority);
345 #endif
346 }
347 
348 }  // namespace art
349