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 <functional>
22 #include <vector>
23 
24 #include "barrier.h"
25 #include "base/mem_map.h"
26 #include "base/mutex.h"
27 
28 namespace art {
29 
30 class ThreadPool;
31 
32 class Closure {
33  public:
~Closure()34   virtual ~Closure() { }
35   virtual void Run(Thread* self) = 0;
36 };
37 
38 class FunctionClosure : public Closure {
39  public:
FunctionClosure(std::function<void (Thread *)> && f)40   explicit FunctionClosure(std::function<void(Thread*)>&& f) : func_(std::move(f)) {}
Run(Thread * self)41   void Run(Thread* self) override {
42     func_(self);
43   }
44 
45  private:
46   std::function<void(Thread*)> func_;
47 };
48 
49 class Task : public Closure {
50  public:
51   // Called after Closure::Run has been called.
Finalize()52   virtual void Finalize() { }
53 };
54 
55 class SelfDeletingTask : public Task {
56  public:
~SelfDeletingTask()57   virtual ~SelfDeletingTask() { }
Finalize()58   virtual void Finalize() {
59     delete this;
60   }
61 };
62 
63 class FunctionTask : public SelfDeletingTask {
64  public:
FunctionTask(std::function<void (Thread *)> && func)65   explicit FunctionTask(std::function<void(Thread*)>&& func) : func_(std::move(func)) {}
66 
Run(Thread * self)67   void Run(Thread* self) override {
68     func_(self);
69   }
70 
71  private:
72   std::function<void(Thread*)> func_;
73 };
74 
75 class ThreadPoolWorker {
76  public:
77   static const size_t kDefaultStackSize = 1 * MB;
78 
GetStackSize()79   size_t GetStackSize() const {
80     DCHECK(stack_.IsValid());
81     return stack_.Size();
82   }
83 
84   virtual ~ThreadPoolWorker();
85 
86   // Set the "nice" priority for this worker.
87   void SetPthreadPriority(int priority);
88 
89   // Get the "nice" priority for this worker.
90   int GetPthreadPriority();
91 
GetThread()92   Thread* GetThread() const { return thread_; }
93 
94  protected:
95   ThreadPoolWorker(ThreadPool* thread_pool, const std::string& name, size_t stack_size);
96   static void* Callback(void* arg) REQUIRES(!Locks::mutator_lock_);
97   virtual void Run();
98 
99   ThreadPool* const thread_pool_;
100   const std::string name_;
101   MemMap stack_;
102   pthread_t pthread_;
103   Thread* thread_;
104 
105  private:
106   friend class ThreadPool;
107   DISALLOW_COPY_AND_ASSIGN(ThreadPoolWorker);
108 };
109 
110 // Note that thread pool workers will set Thread#setCanCallIntoJava to false.
111 class ThreadPool {
112  public:
113   // Returns the number of threads in the thread pool.
GetThreadCount()114   size_t GetThreadCount() const {
115     return threads_.size();
116   }
117 
118   const std::vector<ThreadPoolWorker*>& GetWorkers();
119 
120   // Broadcast to the workers and tell them to empty out the work queue.
121   void StartWorkers(Thread* self) REQUIRES(!task_queue_lock_);
122 
123   // Do not allow workers to grab any new tasks.
124   void StopWorkers(Thread* self) REQUIRES(!task_queue_lock_);
125 
126   // Add a new task, the first available started worker will process it. Does not delete the task
127   // after running it, it is the caller's responsibility.
128   void AddTask(Thread* self, Task* task) REQUIRES(!task_queue_lock_);
129 
130   // Remove all tasks in the queue.
131   void RemoveAllTasks(Thread* self) REQUIRES(!task_queue_lock_);
132 
133   // Create a named thread pool with the given number of threads.
134   //
135   // If create_peers is true, all worker threads will have a Java peer object. Note that if the
136   // pool is asked to do work on the current thread (see Wait), a peer may not be available. Wait
137   // will conservatively abort if create_peers and do_work are true.
138   ThreadPool(const char* name,
139              size_t num_threads,
140              bool create_peers = false,
141              size_t worker_stack_size = ThreadPoolWorker::kDefaultStackSize);
142   virtual ~ThreadPool();
143 
144   // Create the threads of this pool.
145   void CreateThreads();
146 
147   // Stops and deletes all threads in this pool.
148   void DeleteThreads();
149 
150   // Wait for all tasks currently on queue to get completed. If the pool has been stopped, only
151   // wait till all already running tasks are done.
152   // When the pool was created with peers for workers, do_work must not be true (see ThreadPool()).
153   void Wait(Thread* self, bool do_work, bool may_hold_locks) REQUIRES(!task_queue_lock_);
154 
155   size_t GetTaskCount(Thread* self) REQUIRES(!task_queue_lock_);
156 
157   // Returns the total amount of workers waited for tasks.
GetWaitTime()158   uint64_t GetWaitTime() const {
159     return total_wait_time_;
160   }
161 
162   // Provides a way to bound the maximum number of worker threads, threads must be less the the
163   // thread count of the thread pool.
164   void SetMaxActiveWorkers(size_t threads) REQUIRES(!task_queue_lock_);
165 
166   // Set the "nice" priority for threads in the pool.
167   void SetPthreadPriority(int priority);
168 
169   // CHECK that the "nice" priority of threads in the pool is the given
170   // `priority`.
171   void CheckPthreadPriority(int priority);
172 
173   // Wait for workers to be created.
174   void WaitForWorkersToBeCreated();
175 
176  protected:
177   // get a task to run, blocks if there are no tasks left
178   virtual Task* GetTask(Thread* self) REQUIRES(!task_queue_lock_);
179 
180   // Try to get a task, returning null if there is none available.
181   Task* TryGetTask(Thread* self) REQUIRES(!task_queue_lock_);
182   Task* TryGetTaskLocked() REQUIRES(task_queue_lock_);
183 
184   // Are we shutting down?
IsShuttingDown()185   bool IsShuttingDown() const REQUIRES(task_queue_lock_) {
186     return shutting_down_;
187   }
188 
HasOutstandingTasks()189   bool HasOutstandingTasks() const REQUIRES(task_queue_lock_) {
190     return started_ && !tasks_.empty();
191   }
192 
193   const std::string name_;
194   Mutex task_queue_lock_;
195   ConditionVariable task_queue_condition_ GUARDED_BY(task_queue_lock_);
196   ConditionVariable completion_condition_ GUARDED_BY(task_queue_lock_);
197   volatile bool started_ GUARDED_BY(task_queue_lock_);
198   volatile bool shutting_down_ GUARDED_BY(task_queue_lock_);
199   // How many worker threads are waiting on the condition.
200   volatile size_t waiting_count_ GUARDED_BY(task_queue_lock_);
201   std::deque<Task*> tasks_ GUARDED_BY(task_queue_lock_);
202   std::vector<ThreadPoolWorker*> threads_;
203   // Work balance detection.
204   uint64_t start_time_ GUARDED_BY(task_queue_lock_);
205   uint64_t total_wait_time_;
206   Barrier creation_barier_;
207   size_t max_active_workers_ GUARDED_BY(task_queue_lock_);
208   const bool create_peers_;
209   const size_t worker_stack_size_;
210 
211  private:
212   friend class ThreadPoolWorker;
213   friend class WorkStealingWorker;
214   DISALLOW_COPY_AND_ASSIGN(ThreadPool);
215 };
216 
217 }  // namespace art
218 
219 #endif  // ART_RUNTIME_THREAD_POOL_H_
220