| OLD | NEW |
| 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2012 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/threading/sequenced_worker_pool.h" | 5 #include "base/threading/sequenced_worker_pool.h" |
| 6 | 6 |
| 7 #include <list> | 7 #include <list> |
| 8 #include <map> | 8 #include <map> |
| 9 #include <set> | 9 #include <set> |
| 10 #include <utility> | 10 #include <utility> |
| (...skipping 278 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 289 class SequencedWorkerPool::Inner { | 289 class SequencedWorkerPool::Inner { |
| 290 public: | 290 public: |
| 291 // Take a raw pointer to |worker| to avoid cycles (since we're owned | 291 // Take a raw pointer to |worker| to avoid cycles (since we're owned |
| 292 // by it). | 292 // by it). |
| 293 Inner(SequencedWorkerPool* worker_pool, size_t max_threads, | 293 Inner(SequencedWorkerPool* worker_pool, size_t max_threads, |
| 294 const std::string& thread_name_prefix, | 294 const std::string& thread_name_prefix, |
| 295 TestingObserver* observer); | 295 TestingObserver* observer); |
| 296 | 296 |
| 297 ~Inner(); | 297 ~Inner(); |
| 298 | 298 |
| 299 SequenceToken GetSequenceToken(); | 299 static SequenceToken GetSequenceToken(); |
| 300 | 300 |
| 301 SequenceToken GetNamedSequenceToken(const std::string& name); | 301 SequenceToken GetNamedSequenceToken(const std::string& name); |
| 302 | 302 |
| 303 // This function accepts a name and an ID. If the name is null, the | 303 // This function accepts a name and an ID. If the name is null, the |
| 304 // token ID is used. This allows us to implement the optional name lookup | 304 // token ID is used. This allows us to implement the optional name lookup |
| 305 // from a single function without having to enter the lock a separate time. | 305 // from a single function without having to enter the lock a separate time. |
| 306 bool PostTask(const std::string* optional_token_name, | 306 bool PostTask(const std::string* optional_token_name, |
| 307 SequenceToken sequence_token, | 307 SequenceToken sequence_token, |
| 308 WorkerShutdown shutdown_behavior, | 308 WorkerShutdown shutdown_behavior, |
| 309 const tracked_objects::Location& from_here, | 309 const tracked_objects::Location& from_here, |
| 310 const Closure& task, | 310 const Closure& task, |
| 311 TimeDelta delay); | 311 TimeDelta delay); |
| 312 | 312 |
| 313 bool RunsTasksOnCurrentThread() const; | 313 bool RunsTasksOnCurrentThread() const; |
| 314 | 314 |
| 315 bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const; | 315 bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const; |
| 316 | 316 |
| 317 void SetRunningTaskInfoForCurrentThread( |
| 318 SequenceToken sequence_token, WorkerShutdown shutdown_behavior); |
| 319 |
| 317 void CleanupForTesting(); | 320 void CleanupForTesting(); |
| 318 | 321 |
| 319 void SignalHasWorkForTesting(); | 322 void SignalHasWorkForTesting(); |
| 320 | 323 |
| 321 int GetWorkSignalCountForTesting() const; | 324 int GetWorkSignalCountForTesting() const; |
| 322 | 325 |
| 323 void Shutdown(int max_blocking_tasks_after_shutdown); | 326 void Shutdown(int max_blocking_tasks_after_shutdown); |
| 324 | 327 |
| 325 bool IsShutdownInProgress(); | 328 bool IsShutdownInProgress(); |
| 326 | 329 |
| (...skipping 249 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 576 // Need to explicitly join with the threads before they're destroyed or else | 579 // Need to explicitly join with the threads before they're destroyed or else |
| 577 // they will be running when our object is half torn down. | 580 // they will be running when our object is half torn down. |
| 578 for (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it) | 581 for (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it) |
| 579 it->second->Join(); | 582 it->second->Join(); |
| 580 threads_.clear(); | 583 threads_.clear(); |
| 581 | 584 |
| 582 if (testing_observer_) | 585 if (testing_observer_) |
| 583 testing_observer_->OnDestruct(); | 586 testing_observer_->OnDestruct(); |
| 584 } | 587 } |
| 585 | 588 |
| 589 // static |
| 586 SequencedWorkerPool::SequenceToken | 590 SequencedWorkerPool::SequenceToken |
| 587 SequencedWorkerPool::Inner::GetSequenceToken() { | 591 SequencedWorkerPool::Inner::GetSequenceToken() { |
| 588 // Need to add one because StaticAtomicSequenceNumber starts at zero, which | 592 // Need to add one because StaticAtomicSequenceNumber starts at zero, which |
| 589 // is used as a sentinel value in SequenceTokens. | 593 // is used as a sentinel value in SequenceTokens. |
| 590 return SequenceToken(g_last_sequence_number_.GetNext() + 1); | 594 return SequenceToken(g_last_sequence_number_.GetNext() + 1); |
| 591 } | 595 } |
| 592 | 596 |
| 593 SequencedWorkerPool::SequenceToken | 597 SequencedWorkerPool::SequenceToken |
| 594 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) { | 598 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) { |
| 595 AutoLock lock(lock_); | 599 AutoLock lock(lock_); |
| (...skipping 80 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 676 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread( | 680 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread( |
| 677 SequenceToken sequence_token) const { | 681 SequenceToken sequence_token) const { |
| 678 AutoLock lock(lock_); | 682 AutoLock lock(lock_); |
| 679 ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId()); | 683 ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId()); |
| 680 if (found == threads_.end()) | 684 if (found == threads_.end()) |
| 681 return false; | 685 return false; |
| 682 return found->second->is_processing_task() && | 686 return found->second->is_processing_task() && |
| 683 sequence_token.Equals(found->second->task_sequence_token()); | 687 sequence_token.Equals(found->second->task_sequence_token()); |
| 684 } | 688 } |
| 685 | 689 |
| 690 void SequencedWorkerPool::Inner::SetRunningTaskInfoForCurrentThread( |
| 691 SequenceToken sequence_token, WorkerShutdown shutdown_behavior) { |
| 692 AutoLock lock(lock_); |
| 693 ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId()); |
| 694 DCHECK(found != threads_.end()); |
| 695 DCHECK(found->second->is_processing_task()); |
| 696 DCHECK(!found->second->task_sequence_token().IsValid()); |
| 697 found->second->set_running_task_info(sequence_token, shutdown_behavior); |
| 698 } |
| 699 |
| 686 // See https://code.google.com/p/chromium/issues/detail?id=168415 | 700 // See https://code.google.com/p/chromium/issues/detail?id=168415 |
| 687 void SequencedWorkerPool::Inner::CleanupForTesting() { | 701 void SequencedWorkerPool::Inner::CleanupForTesting() { |
| 688 DCHECK(!RunsTasksOnCurrentThread()); | 702 DCHECK(!RunsTasksOnCurrentThread()); |
| 689 base::ThreadRestrictions::ScopedAllowWait allow_wait; | 703 base::ThreadRestrictions::ScopedAllowWait allow_wait; |
| 690 AutoLock lock(lock_); | 704 AutoLock lock(lock_); |
| 691 CHECK_EQ(CLEANUP_DONE, cleanup_state_); | 705 CHECK_EQ(CLEANUP_DONE, cleanup_state_); |
| 692 if (shutdown_called_) | 706 if (shutdown_called_) |
| 693 return; | 707 return; |
| 694 if (pending_tasks_.empty() && waiting_thread_count_ == threads_.size()) | 708 if (pending_tasks_.empty() && waiting_thread_count_ == threads_.size()) |
| 695 return; | 709 return; |
| (...skipping 478 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 1174 SequencedWorkerPool::SequenceToken | 1188 SequencedWorkerPool::SequenceToken |
| 1175 SequencedWorkerPool::GetSequenceTokenForCurrentThread() { | 1189 SequencedWorkerPool::GetSequenceTokenForCurrentThread() { |
| 1176 Worker* worker = Worker::GetForCurrentThread(); | 1190 Worker* worker = Worker::GetForCurrentThread(); |
| 1177 if (!worker) | 1191 if (!worker) |
| 1178 return SequenceToken(); | 1192 return SequenceToken(); |
| 1179 | 1193 |
| 1180 return worker->task_sequence_token(); | 1194 return worker->task_sequence_token(); |
| 1181 } | 1195 } |
| 1182 | 1196 |
| 1183 // static | 1197 // static |
| 1184 scoped_refptr<SequencedWorkerPool> | 1198 scoped_refptr<SequencedTaskRunner> |
| 1185 SequencedWorkerPool::GetWorkerPoolForCurrentThread() { | 1199 SequencedWorkerPool::GetSequencedTaskRunnerForCurrentThread() { |
| 1186 Worker* worker = Worker::GetForCurrentThread(); | 1200 Worker* worker = Worker::GetForCurrentThread(); |
| 1187 if (!worker) | 1201 if (!worker) |
| 1188 return nullptr; | 1202 return nullptr; |
| 1189 | 1203 |
| 1190 return worker->worker_pool(); | 1204 scoped_refptr<SequencedWorkerPool> pool = worker->worker_pool(); |
| 1205 SequenceToken sequence_token = worker->task_sequence_token(); |
| 1206 WorkerShutdown shutdown_behavior = worker->task_shutdown_behavior(); |
| 1207 if (!sequence_token.IsValid()) { |
| 1208 sequence_token = Inner::GetSequenceToken(); |
| 1209 pool->inner_->SetRunningTaskInfoForCurrentThread(sequence_token, |
| 1210 shutdown_behavior); |
| 1211 } |
| 1212 DCHECK(pool->IsRunningSequenceOnCurrentThread(sequence_token)); |
| 1213 return new SequencedWorkerPoolSequencedTaskRunner(pool, sequence_token, |
| 1214 shutdown_behavior); |
| 1191 } | 1215 } |
| 1192 | 1216 |
| 1193 SequencedWorkerPool::SequencedWorkerPool(size_t max_threads, | 1217 SequencedWorkerPool::SequencedWorkerPool(size_t max_threads, |
| 1194 const std::string& thread_name_prefix) | 1218 const std::string& thread_name_prefix) |
| 1195 : constructor_task_runner_(ThreadTaskRunnerHandle::Get()), | 1219 : constructor_task_runner_(ThreadTaskRunnerHandle::Get()), |
| 1196 inner_(new Inner(this, max_threads, thread_name_prefix, NULL)) { | 1220 inner_(new Inner(this, max_threads, thread_name_prefix, NULL)) { |
| 1197 } | 1221 } |
| 1198 | 1222 |
| 1199 SequencedWorkerPool::SequencedWorkerPool(size_t max_threads, | 1223 SequencedWorkerPool::SequencedWorkerPool(size_t max_threads, |
| 1200 const std::string& thread_name_prefix, | 1224 const std::string& thread_name_prefix, |
| 1201 TestingObserver* observer) | 1225 TestingObserver* observer) |
| 1202 : constructor_task_runner_(ThreadTaskRunnerHandle::Get()), | 1226 : constructor_task_runner_(ThreadTaskRunnerHandle::Get()), |
| 1203 inner_(new Inner(this, max_threads, thread_name_prefix, observer)) { | 1227 inner_(new Inner(this, max_threads, thread_name_prefix, observer)) { |
| 1204 } | 1228 } |
| 1205 | 1229 |
| 1206 SequencedWorkerPool::~SequencedWorkerPool() {} | 1230 SequencedWorkerPool::~SequencedWorkerPool() {} |
| 1207 | 1231 |
| 1208 void SequencedWorkerPool::OnDestruct() const { | 1232 void SequencedWorkerPool::OnDestruct() const { |
| 1209 // Avoid deleting ourselves on a worker thread (which would | 1233 // Avoid deleting ourselves on a worker thread (which would deadlock). |
| 1210 // deadlock). | |
| 1211 if (RunsTasksOnCurrentThread()) { | 1234 if (RunsTasksOnCurrentThread()) { |
| 1212 constructor_task_runner_->DeleteSoon(FROM_HERE, this); | 1235 constructor_task_runner_->DeleteSoon(FROM_HERE, this); |
| 1213 } else { | 1236 } else { |
| 1214 delete this; | 1237 delete this; |
| 1215 } | 1238 } |
| 1216 } | 1239 } |
| 1217 | 1240 |
| 1241 // static |
| 1218 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() { | 1242 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() { |
| 1219 return inner_->GetSequenceToken(); | 1243 return Inner::GetSequenceToken(); |
| 1220 } | 1244 } |
| 1221 | 1245 |
| 1222 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken( | 1246 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken( |
| 1223 const std::string& name) { | 1247 const std::string& name) { |
| 1224 return inner_->GetNamedSequenceToken(name); | 1248 return inner_->GetNamedSequenceToken(name); |
| 1225 } | 1249 } |
| 1226 | 1250 |
| 1227 scoped_refptr<SequencedTaskRunner> SequencedWorkerPool::GetSequencedTaskRunner( | 1251 scoped_refptr<SequencedTaskRunner> SequencedWorkerPool::GetSequencedTaskRunner( |
| 1228 SequenceToken token) { | 1252 SequenceToken token) { |
| 1229 return GetSequencedTaskRunnerWithShutdownBehavior(token, BLOCK_SHUTDOWN); | 1253 return GetSequencedTaskRunnerWithShutdownBehavior(token, BLOCK_SHUTDOWN); |
| (...skipping 101 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 1331 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) { | 1355 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) { |
| 1332 DCHECK(constructor_task_runner_->BelongsToCurrentThread()); | 1356 DCHECK(constructor_task_runner_->BelongsToCurrentThread()); |
| 1333 inner_->Shutdown(max_new_blocking_tasks_after_shutdown); | 1357 inner_->Shutdown(max_new_blocking_tasks_after_shutdown); |
| 1334 } | 1358 } |
| 1335 | 1359 |
| 1336 bool SequencedWorkerPool::IsShutdownInProgress() { | 1360 bool SequencedWorkerPool::IsShutdownInProgress() { |
| 1337 return inner_->IsShutdownInProgress(); | 1361 return inner_->IsShutdownInProgress(); |
| 1338 } | 1362 } |
| 1339 | 1363 |
| 1340 } // namespace base | 1364 } // namespace base |
| OLD | NEW |