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" priorty for this worker.
87   void SetPthreadPriority(int priority);
88 
GetThread()89   Thread* GetThread() const { return thread_; }
90 
91  protected:
92   ThreadPoolWorker(ThreadPool* thread_pool, const std::string& name, size_t stack_size);
93   static void* Callback(void* arg) REQUIRES(!Locks::mutator_lock_);
94   virtual void Run();
95 
96   ThreadPool* const thread_pool_;
97   const std::string name_;
98   MemMap stack_;
99   pthread_t pthread_;
100   Thread* thread_;
101 
102  private:
103   friend class ThreadPool;
104   DISALLOW_COPY_AND_ASSIGN(ThreadPoolWorker);
105 };
106 
107 // Note that thread pool workers will set Thread#setCanCallIntoJava to false.
108 class ThreadPool {
109  public:
110   // Returns the number of threads in the thread pool.
GetThreadCount()111   size_t GetThreadCount() const {
112     return threads_.size();
113   }
114 
115   const std::vector<ThreadPoolWorker*>& GetWorkers();
116 
117   // Broadcast to the workers and tell them to empty out the work queue.
118   void StartWorkers(Thread* self) REQUIRES(!task_queue_lock_);
119 
120   // Do not allow workers to grab any new tasks.
121   void StopWorkers(Thread* self) REQUIRES(!task_queue_lock_);
122 
123   // Add a new task, the first available started worker will process it. Does not delete the task
124   // after running it, it is the caller's responsibility.
125   void AddTask(Thread* self, Task* task) REQUIRES(!task_queue_lock_);
126 
127   // Remove all tasks in the queue.
128   void RemoveAllTasks(Thread* self) REQUIRES(!task_queue_lock_);
129 
130   // Create a named thread pool with the given number of threads.
131   //
132   // If create_peers is true, all worker threads will have a Java peer object. Note that if the
133   // pool is asked to do work on the current thread (see Wait), a peer may not be available. Wait
134   // will conservatively abort if create_peers and do_work are true.
135   ThreadPool(const char* name,
136              size_t num_threads,
137              bool create_peers = false,
138              size_t worker_stack_size = ThreadPoolWorker::kDefaultStackSize);
139   virtual ~ThreadPool();
140 
141   // Create the threads of this pool.
142   void CreateThreads();
143 
144   // Stops and deletes all threads in this pool.
145   void DeleteThreads();
146 
147   // Wait for all tasks currently on queue to get completed. If the pool has been stopped, only
148   // wait till all already running tasks are done.
149   // When the pool was created with peers for workers, do_work must not be true (see ThreadPool()).
150   void Wait(Thread* self, bool do_work, bool may_hold_locks) REQUIRES(!task_queue_lock_);
151 
152   size_t GetTaskCount(Thread* self) REQUIRES(!task_queue_lock_);
153 
154   // Returns the total amount of workers waited for tasks.
GetWaitTime()155   uint64_t GetWaitTime() const {
156     return total_wait_time_;
157   }
158 
159   // Provides a way to bound the maximum number of worker threads, threads must be less the the
160   // thread count of the thread pool.
161   void SetMaxActiveWorkers(size_t threads) REQUIRES(!task_queue_lock_);
162 
163   // Set the "nice" priorty for threads in the pool.
164   void SetPthreadPriority(int priority);
165 
166   // Wait for workers to be created.
167   void WaitForWorkersToBeCreated();
168 
169  protected:
170   // get a task to run, blocks if there are no tasks left
171   virtual Task* GetTask(Thread* self) REQUIRES(!task_queue_lock_);
172 
173   // Try to get a task, returning null if there is none available.
174   Task* TryGetTask(Thread* self) REQUIRES(!task_queue_lock_);
175   Task* TryGetTaskLocked() REQUIRES(task_queue_lock_);
176 
177   // Are we shutting down?
IsShuttingDown()178   bool IsShuttingDown() const REQUIRES(task_queue_lock_) {
179     return shutting_down_;
180   }
181 
HasOutstandingTasks()182   bool HasOutstandingTasks() const REQUIRES(task_queue_lock_) {
183     return started_ && !tasks_.empty();
184   }
185 
186   const std::string name_;
187   Mutex task_queue_lock_;
188   ConditionVariable task_queue_condition_ GUARDED_BY(task_queue_lock_);
189   ConditionVariable completion_condition_ GUARDED_BY(task_queue_lock_);
190   volatile bool started_ GUARDED_BY(task_queue_lock_);
191   volatile bool shutting_down_ GUARDED_BY(task_queue_lock_);
192   // How many worker threads are waiting on the condition.
193   volatile size_t waiting_count_ GUARDED_BY(task_queue_lock_);
194   std::deque<Task*> tasks_ GUARDED_BY(task_queue_lock_);
195   std::vector<ThreadPoolWorker*> threads_;
196   // Work balance detection.
197   uint64_t start_time_ GUARDED_BY(task_queue_lock_);
198   uint64_t total_wait_time_;
199   Barrier creation_barier_;
200   size_t max_active_workers_ GUARDED_BY(task_queue_lock_);
201   const bool create_peers_;
202   const size_t worker_stack_size_;
203 
204  private:
205   friend class ThreadPoolWorker;
206   friend class WorkStealingWorker;
207   DISALLOW_COPY_AND_ASSIGN(ThreadPool);
208 };
209 
210 }  // namespace art
211 
212 #endif  // ART_RUNTIME_THREAD_POOL_H_
213