OLD | NEW |
1 // Copyright 2016 The Chromium Authors. All rights reserved. | 1 // Copyright 2016 The Chromium Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #include "base/task_scheduler/scheduler_thread_pool_impl.h" | 5 #include "base/task_scheduler/scheduler_thread_pool_impl.h" |
6 | 6 |
| 7 #include <algorithm> |
7 #include <utility> | 8 #include <utility> |
8 | 9 |
9 #include "base/bind.h" | 10 #include "base/bind.h" |
10 #include "base/bind_helpers.h" | 11 #include "base/bind_helpers.h" |
11 #include "base/lazy_instance.h" | 12 #include "base/lazy_instance.h" |
12 #include "base/logging.h" | |
13 #include "base/memory/ptr_util.h" | 13 #include "base/memory/ptr_util.h" |
14 #include "base/sequenced_task_runner.h" | 14 #include "base/sequenced_task_runner.h" |
| 15 #include "base/single_thread_task_runner.h" |
15 #include "base/task_scheduler/delayed_task_manager.h" | 16 #include "base/task_scheduler/delayed_task_manager.h" |
16 #include "base/task_scheduler/task_tracker.h" | 17 #include "base/task_scheduler/task_tracker.h" |
17 #include "base/threading/thread_local.h" | 18 #include "base/threading/thread_local.h" |
18 | 19 |
19 namespace base { | 20 namespace base { |
20 namespace internal { | 21 namespace internal { |
21 | 22 |
22 namespace { | 23 namespace { |
23 | 24 |
24 // SchedulerThreadPool that owns the current thread, if any. | 25 // SchedulerThreadPool that owns the current thread, if any. |
25 LazyInstance<ThreadLocalPointer<const SchedulerThreadPool>>::Leaky | 26 LazyInstance<ThreadLocalPointer<const SchedulerThreadPool>>::Leaky |
26 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER; | 27 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER; |
27 | 28 |
| 29 // SchedulerWorkerThread that owns the current thread, if any. |
| 30 LazyInstance<ThreadLocalPointer<const SchedulerWorkerThread>>::Leaky |
| 31 tls_current_worker_thread = LAZY_INSTANCE_INITIALIZER; |
| 32 |
28 // A task runner that runs tasks with the PARALLEL ExecutionMode. | 33 // A task runner that runs tasks with the PARALLEL ExecutionMode. |
29 class SchedulerParallelTaskRunner : public TaskRunner { | 34 class SchedulerParallelTaskRunner : public TaskRunner { |
30 public: | 35 public: |
31 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so | 36 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so |
32 // long as |thread_pool| is alive. | 37 // long as |thread_pool| is alive. |
33 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. | 38 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. |
34 SchedulerParallelTaskRunner(const TaskTraits& traits, | 39 SchedulerParallelTaskRunner(const TaskTraits& traits, |
35 SchedulerThreadPool* thread_pool) | 40 SchedulerThreadPool* thread_pool) |
36 : traits_(traits), thread_pool_(thread_pool) {} | 41 : traits_(traits), thread_pool_(thread_pool) {} |
37 | 42 |
38 // TaskRunner: | 43 // TaskRunner: |
39 bool PostDelayedTask(const tracked_objects::Location& from_here, | 44 bool PostDelayedTask(const tracked_objects::Location& from_here, |
40 const Closure& closure, | 45 const Closure& closure, |
41 TimeDelta delay) override { | 46 TimeDelta delay) override { |
42 // Post the task as part of a one-off single-task Sequence. | 47 // Post the task as part of a one-off single-task Sequence. |
43 return thread_pool_->PostTaskWithSequence( | 48 return thread_pool_->PostTaskWithSequence( |
44 WrapUnique(new Task(from_here, closure, traits_, delay)), | 49 WrapUnique(new Task(from_here, closure, traits_, delay)), |
45 make_scoped_refptr(new Sequence)); | 50 make_scoped_refptr(new Sequence), nullptr); |
46 } | 51 } |
47 | 52 |
48 bool RunsTasksOnCurrentThread() const override { | 53 bool RunsTasksOnCurrentThread() const override { |
49 return tls_current_thread_pool.Get().Get() == thread_pool_; | 54 return tls_current_thread_pool.Get().Get() == thread_pool_; |
50 } | 55 } |
51 | 56 |
52 private: | 57 private: |
53 ~SchedulerParallelTaskRunner() override = default; | 58 ~SchedulerParallelTaskRunner() override = default; |
54 | 59 |
55 const TaskTraits traits_; | 60 const TaskTraits traits_; |
56 SchedulerThreadPool* const thread_pool_; | 61 SchedulerThreadPool* const thread_pool_; |
57 | 62 |
58 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); | 63 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); |
59 }; | 64 }; |
60 | 65 |
61 // A task runner that runs tasks with the SEQUENCED ExecutionMode. | 66 // A task runner that runs tasks with the SEQUENCED ExecutionMode. |
62 class SchedulerSequencedTaskRunner : public SequencedTaskRunner { | 67 class SchedulerSequencedTaskRunner : public SequencedTaskRunner { |
63 public: | 68 public: |
64 // Constructs a SchedulerSequencedTaskRunner which can be used to post tasks | 69 // Constructs a SchedulerSequencedTaskRunner which can be used to post tasks |
65 // so long as |thread_pool| is alive. | 70 // so long as |thread_pool| is alive. |
66 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. | 71 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. |
67 SchedulerSequencedTaskRunner(const TaskTraits& traits, | 72 SchedulerSequencedTaskRunner(const TaskTraits& traits, |
68 SchedulerThreadPool* thread_pool) | 73 SchedulerThreadPool* thread_pool) |
69 : traits_(traits), thread_pool_(thread_pool) {} | 74 : traits_(traits), thread_pool_(thread_pool) {} |
70 | 75 |
71 // SequencedTaskRunner: | 76 // SequencedTaskRunner: |
72 bool PostDelayedTask(const tracked_objects::Location& from_here, | 77 bool PostDelayedTask(const tracked_objects::Location& from_here, |
73 const Closure& closure, | 78 const Closure& closure, |
74 TimeDelta delay) override { | 79 TimeDelta delay) override { |
75 // Post the task as part of |sequence|. | 80 // Post the task as part of |sequence_|. |
76 return thread_pool_->PostTaskWithSequence( | 81 return thread_pool_->PostTaskWithSequence( |
77 WrapUnique(new Task(from_here, closure, traits_, delay)), sequence_); | 82 WrapUnique(new Task(from_here, closure, traits_, delay)), sequence_, |
| 83 nullptr); |
78 } | 84 } |
79 | 85 |
80 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, | 86 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, |
81 const Closure& closure, | 87 const Closure& closure, |
82 base::TimeDelta delay) override { | 88 base::TimeDelta delay) override { |
83 // Tasks are never nested within the task scheduler. | 89 // Tasks are never nested within the task scheduler. |
84 return PostDelayedTask(from_here, closure, delay); | 90 return PostDelayedTask(from_here, closure, delay); |
85 } | 91 } |
86 | 92 |
87 bool RunsTasksOnCurrentThread() const override { | 93 bool RunsTasksOnCurrentThread() const override { |
88 return tls_current_thread_pool.Get().Get() == thread_pool_; | 94 return tls_current_thread_pool.Get().Get() == thread_pool_; |
89 } | 95 } |
90 | 96 |
91 private: | 97 private: |
92 ~SchedulerSequencedTaskRunner() override = default; | 98 ~SchedulerSequencedTaskRunner() override = default; |
93 | 99 |
94 // Sequence for all Tasks posted through this TaskRunner. | 100 // Sequence for all Tasks posted through this TaskRunner. |
95 const scoped_refptr<Sequence> sequence_ = new Sequence; | 101 const scoped_refptr<Sequence> sequence_ = new Sequence; |
96 | 102 |
97 const TaskTraits traits_; | 103 const TaskTraits traits_; |
98 SchedulerThreadPool* const thread_pool_; | 104 SchedulerThreadPool* const thread_pool_; |
99 | 105 |
100 DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner); | 106 DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner); |
101 }; | 107 }; |
102 | 108 |
| 109 // A task runner that runs tasks with the SINGLE_THREADED ExecutionMode. |
| 110 class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner { |
| 111 public: |
| 112 // Constructs a SchedulerSingleThreadTaskRunner which can be used to post |
| 113 // tasks so long as |thread_pool| and |worker_thread| are alive. |
| 114 // TODO(robliao): Find a concrete way to manage the memory of |thread_pool| |
| 115 // and |worker_thread|. |
| 116 SchedulerSingleThreadTaskRunner(const TaskTraits& traits, |
| 117 SchedulerThreadPool* thread_pool, |
| 118 SchedulerWorkerThread* worker_thread) |
| 119 : traits_(traits), |
| 120 thread_pool_(thread_pool), |
| 121 worker_thread_(worker_thread) {} |
| 122 |
| 123 // SingleThreadTaskRunner: |
| 124 bool PostDelayedTask(const tracked_objects::Location& from_here, |
| 125 const Closure& closure, |
| 126 TimeDelta delay) override { |
| 127 // Post the task to be executed by |worker_thread_| as part of |sequence_|. |
| 128 return thread_pool_->PostTaskWithSequence( |
| 129 WrapUnique(new Task(from_here, closure, traits_, delay)), sequence_, |
| 130 worker_thread_); |
| 131 } |
| 132 |
| 133 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, |
| 134 const Closure& closure, |
| 135 base::TimeDelta delay) override { |
| 136 // Tasks are never nested within the task scheduler. |
| 137 return PostDelayedTask(from_here, closure, delay); |
| 138 } |
| 139 |
| 140 bool RunsTasksOnCurrentThread() const override { |
| 141 return tls_current_worker_thread.Get().Get() == worker_thread_; |
| 142 } |
| 143 |
| 144 private: |
| 145 ~SchedulerSingleThreadTaskRunner() override = default; |
| 146 |
| 147 // Sequence for all Tasks posted through this TaskRunner. |
| 148 const scoped_refptr<Sequence> sequence_ = new Sequence; |
| 149 |
| 150 const TaskTraits traits_; |
| 151 SchedulerThreadPool* const thread_pool_; |
| 152 SchedulerWorkerThread* const worker_thread_; |
| 153 |
| 154 DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner); |
| 155 }; |
| 156 |
| 157 // Only used in DCHECKs. |
| 158 bool ContainsWorkerThread( |
| 159 const std::vector<std::unique_ptr<SchedulerWorkerThread>>& worker_threads, |
| 160 const SchedulerWorkerThread* worker_thread) { |
| 161 auto it = std::find_if( |
| 162 worker_threads.begin(), worker_threads.end(), |
| 163 [worker_thread](const std::unique_ptr<SchedulerWorkerThread>& i) { |
| 164 return i.get() == worker_thread; |
| 165 }); |
| 166 return it != worker_threads.end(); |
| 167 } |
| 168 |
103 } // namespace | 169 } // namespace |
104 | 170 |
105 class SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl | 171 class SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl |
106 : public SchedulerWorkerThread::Delegate { | 172 : public SchedulerWorkerThread::Delegate { |
107 public: | 173 public: |
| 174 // |outer| owns the worker thread for which this delegate is constructed. |
| 175 // |re_enqueue_sequence_callback| is invoked when ReEnqueueSequence() is |
| 176 // called with a non-single-threaded Sequence. |shared_priority_queue| is a |
| 177 // PriorityQueue whose transactions may overlap with the worker thread's |
| 178 // single-threaded PriorityQueue's transactions. |
108 SchedulerWorkerThreadDelegateImpl( | 179 SchedulerWorkerThreadDelegateImpl( |
109 SchedulerThreadPoolImpl* outer, | 180 SchedulerThreadPoolImpl* outer, |
110 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback); | 181 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback, |
| 182 const PriorityQueue* shared_priority_queue); |
111 ~SchedulerWorkerThreadDelegateImpl() override; | 183 ~SchedulerWorkerThreadDelegateImpl() override; |
112 | 184 |
| 185 PriorityQueue* single_threaded_priority_queue() { |
| 186 return &single_threaded_priority_queue_; |
| 187 } |
| 188 |
113 // SchedulerWorkerThread::Delegate: | 189 // SchedulerWorkerThread::Delegate: |
114 void OnMainEntry() override; | 190 void OnMainEntry(SchedulerWorkerThread* worker_thread) override; |
115 scoped_refptr<Sequence> GetWork( | 191 scoped_refptr<Sequence> GetWork( |
116 SchedulerWorkerThread* worker_thread) override; | 192 SchedulerWorkerThread* worker_thread) override; |
117 void ReEnqueueSequence(scoped_refptr<Sequence> sequence) override; | 193 void ReEnqueueSequence(scoped_refptr<Sequence> sequence) override; |
118 | 194 |
119 private: | 195 private: |
120 SchedulerThreadPoolImpl* outer_; | 196 SchedulerThreadPoolImpl* outer_; |
121 const ReEnqueueSequenceCallback re_enqueue_sequence_callback_; | 197 const ReEnqueueSequenceCallback re_enqueue_sequence_callback_; |
122 | 198 |
| 199 // Single-threaded PriorityQueue for the worker thread. |
| 200 PriorityQueue single_threaded_priority_queue_; |
| 201 |
| 202 // True if the last Sequence returned by GetWork() was extracted from |
| 203 // |single_threaded_priority_queue_|. |
| 204 bool last_sequence_is_single_threaded_ = false; |
| 205 |
123 DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl); | 206 DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl); |
124 }; | 207 }; |
125 | 208 |
126 SchedulerThreadPoolImpl::~SchedulerThreadPoolImpl() { | 209 SchedulerThreadPoolImpl::~SchedulerThreadPoolImpl() { |
127 // SchedulerThreadPool should never be deleted in production unless its | 210 // SchedulerThreadPool should never be deleted in production unless its |
128 // initialization failed. | 211 // initialization failed. |
129 DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty()); | 212 DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty()); |
130 } | 213 } |
131 | 214 |
| 215 // static |
132 std::unique_ptr<SchedulerThreadPoolImpl> SchedulerThreadPoolImpl::Create( | 216 std::unique_ptr<SchedulerThreadPoolImpl> SchedulerThreadPoolImpl::Create( |
133 ThreadPriority thread_priority, | 217 ThreadPriority thread_priority, |
134 size_t max_threads, | 218 size_t max_threads, |
135 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback, | 219 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback, |
136 TaskTracker* task_tracker, | 220 TaskTracker* task_tracker, |
137 DelayedTaskManager* delayed_task_manager) { | 221 DelayedTaskManager* delayed_task_manager) { |
138 std::unique_ptr<SchedulerThreadPoolImpl> thread_pool( | 222 std::unique_ptr<SchedulerThreadPoolImpl> thread_pool( |
139 new SchedulerThreadPoolImpl(task_tracker, delayed_task_manager)); | 223 new SchedulerThreadPoolImpl(task_tracker, delayed_task_manager)); |
140 if (thread_pool->Initialize(thread_priority, max_threads, | 224 if (thread_pool->Initialize(thread_priority, max_threads, |
141 re_enqueue_sequence_callback)) { | 225 re_enqueue_sequence_callback)) { |
(...skipping 19 matching lines...) Expand all Loading... |
161 scoped_refptr<TaskRunner> SchedulerThreadPoolImpl::CreateTaskRunnerWithTraits( | 245 scoped_refptr<TaskRunner> SchedulerThreadPoolImpl::CreateTaskRunnerWithTraits( |
162 const TaskTraits& traits, | 246 const TaskTraits& traits, |
163 ExecutionMode execution_mode) { | 247 ExecutionMode execution_mode) { |
164 switch (execution_mode) { | 248 switch (execution_mode) { |
165 case ExecutionMode::PARALLEL: | 249 case ExecutionMode::PARALLEL: |
166 return make_scoped_refptr(new SchedulerParallelTaskRunner(traits, this)); | 250 return make_scoped_refptr(new SchedulerParallelTaskRunner(traits, this)); |
167 | 251 |
168 case ExecutionMode::SEQUENCED: | 252 case ExecutionMode::SEQUENCED: |
169 return make_scoped_refptr(new SchedulerSequencedTaskRunner(traits, this)); | 253 return make_scoped_refptr(new SchedulerSequencedTaskRunner(traits, this)); |
170 | 254 |
171 case ExecutionMode::SINGLE_THREADED: | 255 case ExecutionMode::SINGLE_THREADED: { |
172 // TODO(fdoray): Support SINGLE_THREADED TaskRunners. | 256 // TODO(fdoray): Find a way to take load into account when assigning a |
173 NOTREACHED(); | 257 // SchedulerWorkerThread to a SingleThreadTaskRunner. Also, this code |
174 return nullptr; | 258 // assumes that all SchedulerWorkerThreads are alive. Eventually, we might |
| 259 // decide to tear down threads that haven't run tasks for a long time. |
| 260 size_t worker_thread_index; |
| 261 { |
| 262 AutoSchedulerLock auto_lock(next_worker_thread_index_lock_); |
| 263 worker_thread_index = next_worker_thread_index_; |
| 264 next_worker_thread_index_ = |
| 265 (next_worker_thread_index_ + 1) % worker_threads_.size(); |
| 266 } |
| 267 return make_scoped_refptr(new SchedulerSingleThreadTaskRunner( |
| 268 traits, this, worker_threads_[worker_thread_index].get())); |
| 269 } |
175 } | 270 } |
176 | 271 |
177 NOTREACHED(); | 272 NOTREACHED(); |
178 return nullptr; | 273 return nullptr; |
179 } | 274 } |
180 | 275 |
181 void SchedulerThreadPoolImpl::ReEnqueueSequence( | 276 void SchedulerThreadPoolImpl::ReEnqueueSequence( |
182 scoped_refptr<Sequence> sequence, | 277 scoped_refptr<Sequence> sequence, |
183 const SequenceSortKey& sequence_sort_key) { | 278 const SequenceSortKey& sequence_sort_key) { |
184 shared_priority_queue_.BeginTransaction()->Push( | 279 shared_priority_queue_.BeginTransaction()->Push( |
185 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), | 280 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), |
186 sequence_sort_key))); | 281 sequence_sort_key))); |
187 | 282 |
188 // The thread calling this method just ran a Task from |sequence| and will | 283 // The thread calling this method just ran a Task from |sequence| and will |
189 // soon try to get another Sequence from which to run a Task. If the thread | 284 // soon try to get another Sequence from which to run a Task. If the thread |
190 // belongs to this pool, it will get that Sequence from | 285 // belongs to this pool, it will get that Sequence from |
191 // |shared_priority_queue_|. When that's the case, there is no need to wake up | 286 // |shared_priority_queue_|. When that's the case, there is no need to wake up |
192 // another thread after |sequence| is inserted in |shared_priority_queue_|. If | 287 // another thread after |sequence| is inserted in |shared_priority_queue_|. If |
193 // we did wake up another thread, we would waste resources by having more | 288 // we did wake up another thread, we would waste resources by having more |
194 // threads trying to get a Sequence from |shared_priority_queue_| than the | 289 // threads trying to get a Sequence from |shared_priority_queue_| than the |
195 // number of Sequences in it. | 290 // number of Sequences in it. |
196 if (tls_current_thread_pool.Get().Get() != this) | 291 if (tls_current_thread_pool.Get().Get() != this) |
197 WakeUpOneThread(); | 292 WakeUpOneThread(); |
198 } | 293 } |
199 | 294 |
200 bool SchedulerThreadPoolImpl::PostTaskWithSequence( | 295 bool SchedulerThreadPoolImpl::PostTaskWithSequence( |
201 std::unique_ptr<Task> task, | 296 std::unique_ptr<Task> task, |
202 scoped_refptr<Sequence> sequence) { | 297 scoped_refptr<Sequence> sequence, |
| 298 SchedulerWorkerThread* worker_thread) { |
203 DCHECK(task); | 299 DCHECK(task); |
204 DCHECK(sequence); | 300 DCHECK(sequence); |
| 301 DCHECK(!worker_thread || |
| 302 ContainsWorkerThread(worker_threads_, worker_thread)); |
205 | 303 |
206 if (!task_tracker_->WillPostTask(task.get())) | 304 if (!task_tracker_->WillPostTask(task.get())) |
207 return false; | 305 return false; |
208 | 306 |
209 if (task->delayed_run_time.is_null()) { | 307 if (task->delayed_run_time.is_null()) { |
210 PostTaskWithSequenceNow(std::move(task), std::move(sequence)); | 308 PostTaskWithSequenceNow(std::move(task), std::move(sequence), |
| 309 worker_thread); |
211 } else { | 310 } else { |
212 delayed_task_manager_->AddDelayedTask(std::move(task), std::move(sequence), | 311 delayed_task_manager_->AddDelayedTask(std::move(task), std::move(sequence), |
213 this); | 312 worker_thread, this); |
214 } | 313 } |
215 | 314 |
216 return true; | 315 return true; |
217 } | 316 } |
218 | 317 |
219 void SchedulerThreadPoolImpl::PostTaskWithSequenceNow( | 318 void SchedulerThreadPoolImpl::PostTaskWithSequenceNow( |
220 std::unique_ptr<Task> task, | 319 std::unique_ptr<Task> task, |
221 scoped_refptr<Sequence> sequence) { | 320 scoped_refptr<Sequence> sequence, |
| 321 SchedulerWorkerThread* worker_thread) { |
222 DCHECK(task); | 322 DCHECK(task); |
223 DCHECK(sequence); | 323 DCHECK(sequence); |
| 324 DCHECK(!worker_thread || |
| 325 ContainsWorkerThread(worker_threads_, worker_thread)); |
224 | 326 |
225 // Confirm that |task| is ready to run (its delayed run time is either null or | 327 // Confirm that |task| is ready to run (its delayed run time is either null or |
226 // in the past). | 328 // in the past). |
227 DCHECK_LE(task->delayed_run_time, delayed_task_manager_->Now()); | 329 DCHECK_LE(task->delayed_run_time, delayed_task_manager_->Now()); |
228 | 330 |
| 331 // Because |worker_thread| belongs to this thread pool, we know that the type |
| 332 // of its delegate is SchedulerWorkerThreadDelegateImpl. |
| 333 PriorityQueue* const priority_queue = |
| 334 worker_thread |
| 335 ? static_cast<SchedulerWorkerThreadDelegateImpl*>( |
| 336 worker_thread->delegate()) |
| 337 ->single_threaded_priority_queue() |
| 338 : &shared_priority_queue_; |
| 339 DCHECK(priority_queue); |
| 340 |
229 const bool sequence_was_empty = sequence->PushTask(std::move(task)); | 341 const bool sequence_was_empty = sequence->PushTask(std::move(task)); |
230 if (sequence_was_empty) { | 342 if (sequence_was_empty) { |
231 // Insert |sequence| in |shared_priority_queue_| if it was empty before | 343 // Insert |sequence| in |priority_queue| if it was empty before |task| was |
232 // |task| was inserted into it. Otherwise, one of these must be true: | 344 // inserted into it. Otherwise, one of these must be true: |
233 // - |sequence| is already in a PriorityQueue (not necessarily | 345 // - |sequence| is already in a PriorityQueue (not necessarily |
234 // |shared_priority_queue_|), or, | 346 // |shared_priority_queue_|), or, |
235 // - A worker thread is running a Task from |sequence|. It will insert | 347 // - A worker thread is running a Task from |sequence|. It will insert |
236 // |sequence| in a PriorityQueue once it's done running the Task. | 348 // |sequence| in a PriorityQueue once it's done running the Task. |
237 const auto sequence_sort_key = sequence->GetSortKey(); | 349 const auto sequence_sort_key = sequence->GetSortKey(); |
238 shared_priority_queue_.BeginTransaction()->Push( | 350 priority_queue->BeginTransaction()->Push( |
239 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), | 351 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), |
240 sequence_sort_key))); | 352 sequence_sort_key))); |
241 | 353 |
242 // Wake up a worker thread to process |sequence|. | 354 // Wake up a worker thread to process |sequence|. |
243 WakeUpOneThread(); | 355 if (worker_thread) |
| 356 worker_thread->WakeUp(); |
| 357 else |
| 358 WakeUpOneThread(); |
244 } | 359 } |
245 } | 360 } |
246 | 361 |
247 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: | 362 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: |
248 SchedulerWorkerThreadDelegateImpl( | 363 SchedulerWorkerThreadDelegateImpl( |
249 SchedulerThreadPoolImpl* outer, | 364 SchedulerThreadPoolImpl* outer, |
250 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) | 365 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback, |
| 366 const PriorityQueue* shared_priority_queue) |
251 : outer_(outer), | 367 : outer_(outer), |
252 re_enqueue_sequence_callback_(re_enqueue_sequence_callback) {} | 368 re_enqueue_sequence_callback_(re_enqueue_sequence_callback), |
| 369 single_threaded_priority_queue_(shared_priority_queue) {} |
253 | 370 |
254 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: | 371 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: |
255 ~SchedulerWorkerThreadDelegateImpl() = default; | 372 ~SchedulerWorkerThreadDelegateImpl() = default; |
256 | 373 |
257 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::OnMainEntry() { | 374 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::OnMainEntry( |
| 375 SchedulerWorkerThread* worker_thread) { |
| 376 #if DCHECK_IS_ON() |
| 377 // Wait for |outer_->threads_created_| to avoid traversing |
| 378 // |outer_->worker_threads_| while it is being filled by Initialize(). |
| 379 outer_->threads_created_.Wait(); |
| 380 DCHECK(ContainsWorkerThread(outer_->worker_threads_, worker_thread)); |
| 381 #endif |
| 382 |
| 383 DCHECK(!tls_current_worker_thread.Get().Get()); |
258 DCHECK(!tls_current_thread_pool.Get().Get()); | 384 DCHECK(!tls_current_thread_pool.Get().Get()); |
| 385 tls_current_worker_thread.Get().Set(worker_thread); |
259 tls_current_thread_pool.Get().Set(outer_); | 386 tls_current_thread_pool.Get().Set(outer_); |
260 } | 387 } |
261 | 388 |
262 scoped_refptr<Sequence> | 389 scoped_refptr<Sequence> |
263 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::GetWork( | 390 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::GetWork( |
264 SchedulerWorkerThread* worker_thread) { | 391 SchedulerWorkerThread* worker_thread) { |
265 std::unique_ptr<PriorityQueue::Transaction> transaction( | 392 DCHECK(ContainsWorkerThread(outer_->worker_threads_, worker_thread)); |
266 outer_->shared_priority_queue_.BeginTransaction()); | |
267 const auto& sequence_and_sort_key = transaction->Peek(); | |
268 | 393 |
269 if (sequence_and_sort_key.is_null()) { | 394 scoped_refptr<Sequence> sequence; |
270 // |transaction| is kept alive while |worker_thread| is added to | 395 { |
271 // |idle_worker_threads_stack_| to avoid this race: | 396 std::unique_ptr<PriorityQueue::Transaction> shared_transaction( |
272 // 1. This thread creates a Transaction, finds |shared_priority_queue_| | 397 outer_->shared_priority_queue_.BeginTransaction()); |
273 // empty and ends the Transaction. | 398 const auto& shared_sequence_and_sort_key = shared_transaction->Peek(); |
274 // 2. Other thread creates a Transaction, inserts a Sequence into | 399 |
275 // |shared_priority_queue_| and ends the Transaction. This can't happen | 400 std::unique_ptr<PriorityQueue::Transaction> single_threaded_transaction( |
276 // if the Transaction of step 1 is still active because because there can | 401 single_threaded_priority_queue_.BeginTransaction()); |
277 // only be one active Transaction per PriorityQueue at a time. | 402 const auto& single_threaded_sequence_and_sort_key = |
278 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because | 403 single_threaded_transaction->Peek(); |
279 // |idle_worker_threads_stack_| is empty. | 404 |
280 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to | 405 if (shared_sequence_and_sort_key.is_null() && |
281 // sleep. No thread runs the Sequence inserted in step 2. | 406 single_threaded_sequence_and_sort_key.is_null()) { |
282 outer_->AddToIdleWorkerThreadsStack(worker_thread); | 407 single_threaded_transaction.reset(); |
283 return nullptr; | 408 |
| 409 // |shared_transaction| is kept alive while |worker_thread| is added to |
| 410 // |idle_worker_threads_stack_| to avoid this race: |
| 411 // 1. This thread creates a Transaction, finds |shared_priority_queue_| |
| 412 // empty and ends the Transaction. |
| 413 // 2. Other thread creates a Transaction, inserts a Sequence into |
| 414 // |shared_priority_queue_| and ends the Transaction. This can't happen |
| 415 // if the Transaction of step 1 is still active because because there |
| 416 // can only be one active Transaction per PriorityQueue at a time. |
| 417 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because |
| 418 // |idle_worker_threads_stack_| is empty. |
| 419 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to |
| 420 // sleep. No thread runs the Sequence inserted in step 2. |
| 421 outer_->AddToIdleWorkerThreadsStack(worker_thread); |
| 422 return nullptr; |
| 423 } |
| 424 |
| 425 // True if both PriorityQueues have Sequences and the Sequence at the top of |
| 426 // the shared PriorityQueue is more important. |
| 427 const bool shared_sequence_is_more_important = |
| 428 !shared_sequence_and_sort_key.is_null() && |
| 429 !single_threaded_sequence_and_sort_key.is_null() && |
| 430 shared_sequence_and_sort_key.sort_key > |
| 431 single_threaded_sequence_and_sort_key.sort_key; |
| 432 |
| 433 if (single_threaded_sequence_and_sort_key.is_null() || |
| 434 shared_sequence_is_more_important) { |
| 435 sequence = shared_sequence_and_sort_key.sequence; |
| 436 shared_transaction->Pop(); |
| 437 last_sequence_is_single_threaded_ = false; |
| 438 } else { |
| 439 DCHECK(!single_threaded_sequence_and_sort_key.is_null()); |
| 440 sequence = single_threaded_sequence_and_sort_key.sequence; |
| 441 single_threaded_transaction->Pop(); |
| 442 last_sequence_is_single_threaded_ = true; |
| 443 } |
284 } | 444 } |
| 445 DCHECK(sequence); |
285 | 446 |
286 scoped_refptr<Sequence> sequence = sequence_and_sort_key.sequence; | 447 outer_->RemoveFromIdleWorkerThreadsStack(worker_thread); |
287 transaction->Pop(); | |
288 return sequence; | 448 return sequence; |
289 } | 449 } |
290 | 450 |
291 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: | 451 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: |
292 ReEnqueueSequence(scoped_refptr<Sequence> sequence) { | 452 ReEnqueueSequence(scoped_refptr<Sequence> sequence) { |
293 re_enqueue_sequence_callback_.Run(std::move(sequence)); | 453 if (last_sequence_is_single_threaded_) { |
| 454 // A single-threaded Sequence is always re-enqueued in the single-threaded |
| 455 // PriorityQueue from which it was extracted. |
| 456 const SequenceSortKey sequence_sort_key = sequence->GetSortKey(); |
| 457 single_threaded_priority_queue_.BeginTransaction()->Push( |
| 458 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), |
| 459 sequence_sort_key))); |
| 460 } else { |
| 461 // |re_enqueue_sequence_callback_| will determine in which PriorityQueue |
| 462 // |sequence| must be enqueued. |
| 463 re_enqueue_sequence_callback_.Run(std::move(sequence)); |
| 464 } |
294 } | 465 } |
295 | 466 |
296 SchedulerThreadPoolImpl::SchedulerThreadPoolImpl( | 467 SchedulerThreadPoolImpl::SchedulerThreadPoolImpl( |
297 TaskTracker* task_tracker, | 468 TaskTracker* task_tracker, |
298 DelayedTaskManager* delayed_task_manager) | 469 DelayedTaskManager* delayed_task_manager) |
299 : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()), | 470 : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()), |
300 idle_worker_threads_stack_cv_for_testing_( | 471 idle_worker_threads_stack_cv_for_testing_( |
301 idle_worker_threads_stack_lock_.CreateConditionVariable()), | 472 idle_worker_threads_stack_lock_.CreateConditionVariable()), |
302 join_for_testing_returned_(true, false), | 473 join_for_testing_returned_(true, false), |
| 474 #if DCHECK_IS_ON() |
| 475 threads_created_(true, false), |
| 476 #endif |
303 task_tracker_(task_tracker), | 477 task_tracker_(task_tracker), |
304 delayed_task_manager_(delayed_task_manager) { | 478 delayed_task_manager_(delayed_task_manager) { |
305 DCHECK(task_tracker_); | 479 DCHECK(task_tracker_); |
306 DCHECK(delayed_task_manager_); | 480 DCHECK(delayed_task_manager_); |
307 } | 481 } |
308 | 482 |
309 bool SchedulerThreadPoolImpl::Initialize( | 483 bool SchedulerThreadPoolImpl::Initialize( |
310 ThreadPriority thread_priority, | 484 ThreadPriority thread_priority, |
311 size_t max_threads, | 485 size_t max_threads, |
312 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) { | 486 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) { |
313 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | 487 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); |
314 | 488 |
315 DCHECK(worker_threads_.empty()); | 489 DCHECK(worker_threads_.empty()); |
316 | 490 |
317 for (size_t i = 0; i < max_threads; ++i) { | 491 for (size_t i = 0; i < max_threads; ++i) { |
318 std::unique_ptr<SchedulerWorkerThread> worker_thread = | 492 std::unique_ptr<SchedulerWorkerThread> worker_thread = |
319 SchedulerWorkerThread::Create( | 493 SchedulerWorkerThread::Create( |
320 thread_priority, WrapUnique(new SchedulerWorkerThreadDelegateImpl( | 494 thread_priority, |
321 this, re_enqueue_sequence_callback)), | 495 WrapUnique(new SchedulerWorkerThreadDelegateImpl( |
| 496 this, re_enqueue_sequence_callback, &shared_priority_queue_)), |
322 task_tracker_); | 497 task_tracker_); |
323 if (!worker_thread) | 498 if (!worker_thread) |
324 break; | 499 break; |
325 idle_worker_threads_stack_.Push(worker_thread.get()); | 500 idle_worker_threads_stack_.Push(worker_thread.get()); |
326 worker_threads_.push_back(std::move(worker_thread)); | 501 worker_threads_.push_back(std::move(worker_thread)); |
327 } | 502 } |
328 | 503 |
| 504 #if DCHECK_IS_ON() |
| 505 threads_created_.Signal(); |
| 506 #endif |
| 507 |
329 return !worker_threads_.empty(); | 508 return !worker_threads_.empty(); |
330 } | 509 } |
331 | 510 |
332 void SchedulerThreadPoolImpl::WakeUpOneThread() { | 511 void SchedulerThreadPoolImpl::WakeUpOneThread() { |
333 SchedulerWorkerThread* worker_thread; | 512 SchedulerWorkerThread* worker_thread; |
334 { | 513 { |
335 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | 514 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); |
336 worker_thread = idle_worker_threads_stack_.Pop(); | 515 worker_thread = idle_worker_threads_stack_.Pop(); |
337 } | 516 } |
338 if (worker_thread) | 517 if (worker_thread) |
339 worker_thread->WakeUp(); | 518 worker_thread->WakeUp(); |
340 } | 519 } |
341 | 520 |
342 void SchedulerThreadPoolImpl::AddToIdleWorkerThreadsStack( | 521 void SchedulerThreadPoolImpl::AddToIdleWorkerThreadsStack( |
343 SchedulerWorkerThread* worker_thread) { | 522 SchedulerWorkerThread* worker_thread) { |
344 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | 523 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); |
345 idle_worker_threads_stack_.Push(worker_thread); | 524 idle_worker_threads_stack_.Push(worker_thread); |
346 DCHECK_LE(idle_worker_threads_stack_.Size(), worker_threads_.size()); | 525 DCHECK_LE(idle_worker_threads_stack_.Size(), worker_threads_.size()); |
347 | 526 |
348 if (idle_worker_threads_stack_.Size() == worker_threads_.size()) | 527 if (idle_worker_threads_stack_.Size() == worker_threads_.size()) |
349 idle_worker_threads_stack_cv_for_testing_->Broadcast(); | 528 idle_worker_threads_stack_cv_for_testing_->Broadcast(); |
350 } | 529 } |
351 | 530 |
| 531 void SchedulerThreadPoolImpl::RemoveFromIdleWorkerThreadsStack( |
| 532 SchedulerWorkerThread* worker_thread) { |
| 533 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); |
| 534 idle_worker_threads_stack_.Remove(worker_thread); |
| 535 } |
| 536 |
352 } // namespace internal | 537 } // namespace internal |
353 } // namespace base | 538 } // namespace base |
OLD | NEW |