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 #ifndef ART_RUNTIME_THREAD_POOL_H_ 18 #define ART_RUNTIME_THREAD_POOL_H_ 19 20 #include <deque> 21 #include <vector> 22 23 #include "barrier.h" 24 #include "base/mutex.h" 25 #include "closure.h" 26 #include "mem_map.h" 27 28 namespace art { 29 30 class ThreadPool; 31 32 class Task : public Closure { 33 public: 34 // Called when references reaches 0. 35 virtual void Finalize() { } 36 }; 37 38 class ThreadPoolWorker { 39 public: 40 static const size_t kDefaultStackSize = 1 * MB; 41 42 size_t GetStackSize() const { 43 DCHECK(stack_.get() != nullptr); 44 return stack_->Size(); 45 } 46 47 virtual ~ThreadPoolWorker(); 48 49 protected: 50 ThreadPoolWorker(ThreadPool* thread_pool, const std::string& name, size_t stack_size); 51 static void* Callback(void* arg) LOCKS_EXCLUDED(Locks::mutator_lock_); 52 virtual void Run(); 53 54 ThreadPool* const thread_pool_; 55 const std::string name_; 56 std::unique_ptr<MemMap> stack_; 57 pthread_t pthread_; 58 59 private: 60 friend class ThreadPool; 61 DISALLOW_COPY_AND_ASSIGN(ThreadPoolWorker); 62 }; 63 64 class ThreadPool { 65 public: 66 // Returns the number of threads in the thread pool. 67 size_t GetThreadCount() const { 68 return threads_.size(); 69 } 70 71 // Broadcast to the workers and tell them to empty out the work queue. 72 void StartWorkers(Thread* self); 73 74 // Do not allow workers to grab any new tasks. 75 void StopWorkers(Thread* self); 76 77 // Add a new task, the first available started worker will process it. Does not delete the task 78 // after running it, it is the caller's responsibility. 79 void AddTask(Thread* self, Task* task); 80 81 explicit ThreadPool(const char* name, size_t num_threads); 82 virtual ~ThreadPool(); 83 84 // Wait for all tasks currently on queue to get completed. 85 void Wait(Thread* self, bool do_work, bool may_hold_locks); 86 87 size_t GetTaskCount(Thread* self); 88 89 // Returns the total amount of workers waited for tasks. 90 uint64_t GetWaitTime() const { 91 return total_wait_time_; 92 } 93 94 // Provides a way to bound the maximum number of worker threads, threads must be less the the 95 // thread count of the thread pool. 96 void SetMaxActiveWorkers(size_t threads); 97 98 protected: 99 // get a task to run, blocks if there are no tasks left 100 virtual Task* GetTask(Thread* self); 101 102 // Try to get a task, returning NULL if there is none available. 103 Task* TryGetTask(Thread* self); 104 Task* TryGetTaskLocked(Thread* self) EXCLUSIVE_LOCKS_REQUIRED(task_queue_lock_); 105 106 // Are we shutting down? 107 bool IsShuttingDown() const EXCLUSIVE_LOCKS_REQUIRED(task_queue_lock_) { 108 return shutting_down_; 109 } 110 111 const std::string name_; 112 Mutex task_queue_lock_; 113 ConditionVariable task_queue_condition_ GUARDED_BY(task_queue_lock_); 114 ConditionVariable completion_condition_ GUARDED_BY(task_queue_lock_); 115 volatile bool started_ GUARDED_BY(task_queue_lock_); 116 volatile bool shutting_down_ GUARDED_BY(task_queue_lock_); 117 // How many worker threads are waiting on the condition. 118 volatile size_t waiting_count_ GUARDED_BY(task_queue_lock_); 119 std::deque<Task*> tasks_ GUARDED_BY(task_queue_lock_); 120 // TODO: make this immutable/const? 121 std::vector<ThreadPoolWorker*> threads_; 122 // Work balance detection. 123 uint64_t start_time_ GUARDED_BY(task_queue_lock_); 124 uint64_t total_wait_time_; 125 Barrier creation_barier_; 126 size_t max_active_workers_ GUARDED_BY(task_queue_lock_); 127 128 private: 129 friend class ThreadPoolWorker; 130 friend class WorkStealingWorker; 131 DISALLOW_COPY_AND_ASSIGN(ThreadPool); 132 }; 133 134 class WorkStealingTask : public Task { 135 public: 136 WorkStealingTask() : ref_count_(0) {} 137 138 size_t GetRefCount() const { 139 return ref_count_; 140 } 141 142 virtual void StealFrom(Thread* self, WorkStealingTask* source) = 0; 143 144 private: 145 // How many people are referencing this task. 146 size_t ref_count_; 147 148 friend class WorkStealingWorker; 149 }; 150 151 class WorkStealingWorker : public ThreadPoolWorker { 152 public: 153 virtual ~WorkStealingWorker(); 154 155 bool IsRunningTask() const { 156 return task_ != NULL; 157 } 158 159 protected: 160 WorkStealingTask* task_; 161 162 WorkStealingWorker(ThreadPool* thread_pool, const std::string& name, size_t stack_size); 163 virtual void Run(); 164 165 private: 166 friend class WorkStealingThreadPool; 167 DISALLOW_COPY_AND_ASSIGN(WorkStealingWorker); 168 }; 169 170 class WorkStealingThreadPool : public ThreadPool { 171 public: 172 explicit WorkStealingThreadPool(const char* name, size_t num_threads); 173 virtual ~WorkStealingThreadPool(); 174 175 private: 176 Mutex work_steal_lock_; 177 // Which thread we are stealing from (round robin). 178 size_t steal_index_; 179 180 // Find a task to steal from 181 WorkStealingTask* FindTaskToStealFrom(Thread* self) EXCLUSIVE_LOCKS_REQUIRED(work_steal_lock_); 182 183 friend class WorkStealingWorker; 184 }; 185 186 } // namespace art 187 188 #endif // ART_RUNTIME_THREAD_POOL_H_ 189
