blob: 26ca19054d8b17007f544a16e12be8cc7624033d [file] [log] [blame]
Elliott Hughes1aa246d2012-12-13 09:29:36 -08001/*
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
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070017#include "thread_pool.h"
18
Andreas Gamped4901292017-05-30 18:41:34 -070019#include <sys/mman.h>
Andreas Gampe9e927f52016-02-29 20:49:38 -080020#include <sys/resource.h>
Andreas Gampe8cf9cb32017-07-19 09:28:38 -070021#include <sys/time.h>
22
23#include <pthread.h>
Andreas Gampe9e927f52016-02-29 20:49:38 -080024
Andreas Gampe57943812017-12-06 21:39:13 -080025#include <android-base/logging.h>
26#include <android-base/stringprintf.h>
Andreas Gampe46ee31b2016-12-14 10:11:49 -080027
Vladimir Marko0b6e2832015-09-24 10:41:33 +010028#include "base/bit_utils.h"
Elliott Hughes1aa246d2012-12-13 09:29:36 -080029#include "base/casts.h"
30#include "base/stl_util.h"
Vladimir Marko80afd022015-05-19 18:08:00 +010031#include "base/time_utils.h"
David Sehrc431b9d2018-03-02 12:01:51 -080032#include "base/utils.h"
Elliott Hughes1aa246d2012-12-13 09:29:36 -080033#include "runtime.h"
Andreas Gampeb486a982017-06-01 13:45:54 -070034#include "thread-current-inl.h"
Elliott Hughes1aa246d2012-12-13 09:29:36 -080035
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070036namespace art {
37
Andreas Gampe46ee31b2016-12-14 10:11:49 -080038using android::base::StringPrintf;
39
Mathieu Chartier720ef762013-08-17 14:46:54 -070040static constexpr bool kMeasureWaitTime = false;
Mathieu Chartier94c32c52013-08-09 11:14:04 -070041
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070042ThreadPoolWorker::ThreadPoolWorker(ThreadPool* thread_pool, const std::string& name,
43 size_t stack_size)
44 : thread_pool_(thread_pool),
Mathieu Chartierbcd5e9d2013-11-13 14:33:28 -080045 name_(name) {
Vladimir Marko0b6e2832015-09-24 10:41:33 +010046 // Add an inaccessible page to catch stack overflow.
47 stack_size += kPageSize;
Mathieu Chartierbcd5e9d2013-11-13 14:33:28 -080048 std::string error_msg;
49 stack_.reset(MemMap::MapAnonymous(name.c_str(), nullptr, stack_size, PROT_READ | PROT_WRITE,
Vladimir Marko5c42c292015-02-25 12:02:49 +000050 false, false, &error_msg));
Mathieu Chartierbcd5e9d2013-11-13 14:33:28 -080051 CHECK(stack_.get() != nullptr) << error_msg;
Vladimir Marko0b6e2832015-09-24 10:41:33 +010052 CHECK_ALIGNED(stack_->Begin(), kPageSize);
Mathieu Chartier8d8de0c2017-10-04 09:35:30 -070053 CheckedCall(mprotect,
54 "mprotect bottom page of thread pool worker stack",
55 stack_->Begin(),
56 kPageSize,
57 PROT_NONE);
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070058 const char* reason = "new thread pool worker thread";
Brian Carlstrombcc29262012-11-02 11:36:03 -070059 pthread_attr_t attr;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070060 CHECK_PTHREAD_CALL(pthread_attr_init, (&attr), reason);
Mathieu Chartierbcd5e9d2013-11-13 14:33:28 -080061 CHECK_PTHREAD_CALL(pthread_attr_setstack, (&attr, stack_->Begin(), stack_->Size()), reason);
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070062 CHECK_PTHREAD_CALL(pthread_create, (&pthread_, &attr, &Callback, this), reason);
63 CHECK_PTHREAD_CALL(pthread_attr_destroy, (&attr), reason);
64}
65
66ThreadPoolWorker::~ThreadPoolWorker() {
Ian Rogers6a3c1fc2014-10-31 00:33:20 -070067 CHECK_PTHREAD_CALL(pthread_join, (pthread_, nullptr), "thread pool worker shutdown");
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070068}
69
Andreas Gampe9e927f52016-02-29 20:49:38 -080070void ThreadPoolWorker::SetPthreadPriority(int priority) {
71 CHECK_GE(priority, PRIO_MIN);
72 CHECK_LE(priority, PRIO_MAX);
Bilyan Borisovbb661c02016-04-04 16:27:32 +010073#if defined(ART_TARGET_ANDROID)
Andreas Gampe9e927f52016-02-29 20:49:38 -080074 int result = setpriority(PRIO_PROCESS, pthread_gettid_np(pthread_), priority);
75 if (result != 0) {
76 PLOG(ERROR) << "Failed to setpriority to :" << priority;
77 }
78#else
79 UNUSED(priority);
80#endif
81}
82
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070083void ThreadPoolWorker::Run() {
84 Thread* self = Thread::Current();
Ian Rogers6a3c1fc2014-10-31 00:33:20 -070085 Task* task = nullptr;
Mathieu Chartier35883cc2012-11-13 14:08:12 -080086 thread_pool_->creation_barier_.Wait(self);
Ian Rogers6a3c1fc2014-10-31 00:33:20 -070087 while ((task = thread_pool_->GetTask(self)) != nullptr) {
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070088 task->Run(self);
Mathieu Chartier02b6a782012-10-26 13:51:26 -070089 task->Finalize();
Mathieu Chartier0e4627e2012-10-23 16:13:36 -070090 }
91}
92
93void* ThreadPoolWorker::Callback(void* arg) {
94 ThreadPoolWorker* worker = reinterpret_cast<ThreadPoolWorker*>(arg);
95 Runtime* runtime = Runtime::Current();
Andreas Gampeb15de0c2017-01-24 13:12:19 -080096 CHECK(runtime->AttachCurrentThread(worker->name_.c_str(),
97 true,
98 nullptr,
99 worker->thread_pool_->create_peers_));
Nicolas Geoffray340dafa2016-11-18 16:03:10 +0000100 worker->thread_ = Thread::Current();
Calin Juravleccd56952016-12-15 17:57:38 +0000101 // Thread pool workers cannot call into java.
102 worker->thread_->SetCanCallIntoJava(false);
Alex Light53570672018-07-12 11:09:59 -0700103 // Thread pool workers should not be getting paused by user-code.
104 worker->thread_->SetCanBeSuspendedByUserCode(false);
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700105 // Do work until its time to shut down.
106 worker->Run();
Alex Light53570672018-07-12 11:09:59 -0700107 // Thread pool worker is finished. We want to allow suspension during shutdown.
108 worker->thread_->SetCanBeSuspendedByUserCode(true);
109 // Thread shuts down.
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700110 runtime->DetachCurrentThread();
Ian Rogers6a3c1fc2014-10-31 00:33:20 -0700111 return nullptr;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700112}
113
Brian Carlstrom2ce745c2013-07-17 17:44:30 -0700114void ThreadPool::AddTask(Thread* self, Task* task) {
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700115 MutexLock mu(self, task_queue_lock_);
116 tasks_.push_back(task);
117 // If we have any waiters, signal one.
Mathieu Chartier94c32c52013-08-09 11:14:04 -0700118 if (started_ && waiting_count_ != 0) {
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700119 task_queue_condition_.Signal(self);
120 }
121}
122
Nicolas Geoffray629e9352015-11-04 17:22:16 +0000123void ThreadPool::RemoveAllTasks(Thread* self) {
124 MutexLock mu(self, task_queue_lock_);
125 tasks_.clear();
126}
127
Andreas Gampeb15de0c2017-01-24 13:12:19 -0800128ThreadPool::ThreadPool(const char* name, size_t num_threads, bool create_peers)
Mathieu Chartierbcd5e9d2013-11-13 14:33:28 -0800129 : name_(name),
130 task_queue_lock_("task queue lock"),
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700131 task_queue_condition_("task queue condition", task_queue_lock_),
132 completion_condition_("task completion condition", task_queue_lock_),
133 started_(false),
134 shutting_down_(false),
Mathieu Chartier35883cc2012-11-13 14:08:12 -0800135 waiting_count_(0),
Ian Rogersd914eb22013-04-18 16:11:15 -0700136 start_time_(0),
137 total_wait_time_(0),
Mathieu Chartier35883cc2012-11-13 14:08:12 -0800138 // Add one since the caller of constructor waits on the barrier too.
Mathieu Chartier2775ee42013-08-20 17:43:47 -0700139 creation_barier_(num_threads + 1),
Andreas Gampeb15de0c2017-01-24 13:12:19 -0800140 max_active_workers_(num_threads),
141 create_peers_(create_peers) {
Mathieu Chartier35883cc2012-11-13 14:08:12 -0800142 Thread* self = Thread::Current();
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700143 while (GetThreadCount() < num_threads) {
Andreas Gampe277ccbd2014-11-03 21:36:10 -0800144 const std::string worker_name = StringPrintf("%s worker thread %zu", name_.c_str(),
145 GetThreadCount());
Vladimir Marko0b6e2832015-09-24 10:41:33 +0100146 threads_.push_back(
147 new ThreadPoolWorker(this, worker_name, ThreadPoolWorker::kDefaultStackSize));
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700148 }
Mathieu Chartier35883cc2012-11-13 14:08:12 -0800149 // Wait for all of the threads to attach.
150 creation_barier_.Wait(self);
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700151}
152
Mathieu Chartier2775ee42013-08-20 17:43:47 -0700153void ThreadPool::SetMaxActiveWorkers(size_t threads) {
154 MutexLock mu(Thread::Current(), task_queue_lock_);
155 CHECK_LE(threads, GetThreadCount());
156 max_active_workers_ = threads;
157}
158
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700159ThreadPool::~ThreadPool() {
Mathieu Chartiere46cd752012-10-31 16:56:18 -0700160 {
161 Thread* self = Thread::Current();
162 MutexLock mu(self, task_queue_lock_);
163 // Tell any remaining workers to shut down.
164 shutting_down_ = true;
Mathieu Chartiere46cd752012-10-31 16:56:18 -0700165 // Broadcast to everyone waiting.
166 task_queue_condition_.Broadcast(self);
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700167 completion_condition_.Broadcast(self);
Mathieu Chartiere46cd752012-10-31 16:56:18 -0700168 }
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700169 // Wait for the threads to finish.
170 STLDeleteElements(&threads_);
171}
172
173void ThreadPool::StartWorkers(Thread* self) {
174 MutexLock mu(self, task_queue_lock_);
175 started_ = true;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700176 task_queue_condition_.Broadcast(self);
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700177 start_time_ = NanoTime();
178 total_wait_time_ = 0;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700179}
180
181void ThreadPool::StopWorkers(Thread* self) {
182 MutexLock mu(self, task_queue_lock_);
183 started_ = false;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700184}
185
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700186Task* ThreadPool::GetTask(Thread* self) {
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700187 MutexLock mu(self, task_queue_lock_);
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700188 while (!IsShuttingDown()) {
Mathieu Chartier2775ee42013-08-20 17:43:47 -0700189 const size_t thread_count = GetThreadCount();
190 // Ensure that we don't use more threads than the maximum active workers.
191 const size_t active_threads = thread_count - waiting_count_;
192 // <= since self is considered an active worker.
193 if (active_threads <= max_active_workers_) {
Ian Rogers6a3c1fc2014-10-31 00:33:20 -0700194 Task* task = TryGetTaskLocked();
195 if (task != nullptr) {
Mathieu Chartier2775ee42013-08-20 17:43:47 -0700196 return task;
197 }
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700198 }
199
Mathieu Chartier2775ee42013-08-20 17:43:47 -0700200 ++waiting_count_;
Andreas Gampe6f3a70f2016-11-16 13:58:05 -0800201 if (waiting_count_ == GetThreadCount() && !HasOutstandingTasks()) {
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700202 // We may be done, lets broadcast to the completion condition.
203 completion_condition_.Broadcast(self);
204 }
Mathieu Chartier94c32c52013-08-09 11:14:04 -0700205 const uint64_t wait_start = kMeasureWaitTime ? NanoTime() : 0;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700206 task_queue_condition_.Wait(self);
Mathieu Chartier94c32c52013-08-09 11:14:04 -0700207 if (kMeasureWaitTime) {
208 const uint64_t wait_end = NanoTime();
209 total_wait_time_ += wait_end - std::max(wait_start, start_time_);
210 }
211 --waiting_count_;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700212 }
213
Mathieu Chartier2cebb242015-04-21 16:50:40 -0700214 // We are shutting down, return null to tell the worker thread to stop looping.
Ian Rogers6a3c1fc2014-10-31 00:33:20 -0700215 return nullptr;
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700216}
217
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700218Task* ThreadPool::TryGetTask(Thread* self) {
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700219 MutexLock mu(self, task_queue_lock_);
Ian Rogers6a3c1fc2014-10-31 00:33:20 -0700220 return TryGetTaskLocked();
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700221}
222
Ian Rogers6a3c1fc2014-10-31 00:33:20 -0700223Task* ThreadPool::TryGetTaskLocked() {
Andreas Gampe6f3a70f2016-11-16 13:58:05 -0800224 if (HasOutstandingTasks()) {
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700225 Task* task = tasks_.front();
226 tasks_.pop_front();
227 return task;
228 }
Ian Rogers6a3c1fc2014-10-31 00:33:20 -0700229 return nullptr;
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700230}
231
Ian Rogers1d54e732013-05-02 21:10:01 -0700232void ThreadPool::Wait(Thread* self, bool do_work, bool may_hold_locks) {
233 if (do_work) {
Andreas Gampeb15de0c2017-01-24 13:12:19 -0800234 CHECK(!create_peers_);
Ian Rogers6a3c1fc2014-10-31 00:33:20 -0700235 Task* task = nullptr;
236 while ((task = TryGetTask(self)) != nullptr) {
Ian Rogers1d54e732013-05-02 21:10:01 -0700237 task->Run(self);
238 task->Finalize();
239 }
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700240 }
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700241 // Wait until each thread is waiting and the task list is empty.
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700242 MutexLock mu(self, task_queue_lock_);
Andreas Gampe6f3a70f2016-11-16 13:58:05 -0800243 while (!shutting_down_ && (waiting_count_ != GetThreadCount() || HasOutstandingTasks())) {
Ian Rogers1d54e732013-05-02 21:10:01 -0700244 if (!may_hold_locks) {
245 completion_condition_.Wait(self);
246 } else {
247 completion_condition_.WaitHoldingLocks(self);
248 }
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700249 }
250}
251
Brian Carlstrom2ce745c2013-07-17 17:44:30 -0700252size_t ThreadPool::GetTaskCount(Thread* self) {
Mathieu Chartier02b6a782012-10-26 13:51:26 -0700253 MutexLock mu(self, task_queue_lock_);
254 return tasks_.size();
255}
256
Andreas Gampe9e927f52016-02-29 20:49:38 -0800257void ThreadPool::SetPthreadPriority(int priority) {
258 for (ThreadPoolWorker* worker : threads_) {
259 worker->SetPthreadPriority(priority);
260 }
261}
262
Mathieu Chartier0e4627e2012-10-23 16:13:36 -0700263} // namespace art