| 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 <deque> | 7 #include <deque> |
| 8 #include <set> | 8 #include <set> |
| 9 | 9 |
| 10 #include "base/atomicops.h" | 10 #include "base/atomicops.h" |
| (...skipping 39 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 50 | 50 |
| 51 | 51 |
| 52 // Inner ---------------------------------------------------------------------- | 52 // Inner ---------------------------------------------------------------------- |
| 53 | 53 |
| 54 class SequencedWorkerPool::Inner | 54 class SequencedWorkerPool::Inner |
| 55 : public base::RefCountedThreadSafe<SequencedWorkerPool::Inner> { | 55 : public base::RefCountedThreadSafe<SequencedWorkerPool::Inner> { |
| 56 public: | 56 public: |
| 57 Inner(size_t max_threads, const std::string& thread_name_prefix); | 57 Inner(size_t max_threads, const std::string& thread_name_prefix); |
| 58 virtual ~Inner(); | 58 virtual ~Inner(); |
| 59 | 59 |
| 60 // Backends for SequenceWorkerPool. |
| 60 SequenceToken GetSequenceToken(); | 61 SequenceToken GetSequenceToken(); |
| 61 | |
| 62 SequenceToken GetNamedSequenceToken(const std::string& name); | 62 SequenceToken GetNamedSequenceToken(const std::string& name); |
| 63 | 63 bool PostTask(int sequence_token_id, |
| 64 // This function accepts a name and an ID. If the name is null, the | |
| 65 // token ID is used. This allows us to implement the optional name lookup | |
| 66 // from a single function without having to enter the lock a separate time. | |
| 67 bool PostTask(const std::string* optional_token_name, | |
| 68 int sequence_token_id, | |
| 69 SequencedWorkerPool::WorkerShutdown shutdown_behavior, | 64 SequencedWorkerPool::WorkerShutdown shutdown_behavior, |
| 70 const tracked_objects::Location& from_here, | 65 const tracked_objects::Location& from_here, |
| 71 const base::Closure& task); | 66 const base::Closure& task); |
| 72 | |
| 73 void Shutdown(); | 67 void Shutdown(); |
| 74 | |
| 75 void SetTestingObserver(SequencedWorkerPool::TestingObserver* observer); | 68 void SetTestingObserver(SequencedWorkerPool::TestingObserver* observer); |
| 76 | 69 |
| 77 // Runs the worker loop on the background thread. | 70 // Runs the worker loop on the background thread. |
| 78 void ThreadLoop(Worker* this_worker); | 71 void ThreadLoop(Worker* this_worker); |
| 79 | 72 |
| 80 private: | 73 private: |
| 81 // Called from within the lock, this converts the given token name into a | |
| 82 // token ID, creating a new one if necessary. | |
| 83 int LockedGetNamedTokenID(const std::string& name); | |
| 84 | |
| 85 // The calling code should clear the given delete_these_oustide_lock | 74 // The calling code should clear the given delete_these_oustide_lock |
| 86 // vector the next time the lock is released. See the implementation for | 75 // vector the next time the lock is released. See the implementation for |
| 87 // a more detailed description. | 76 // a more detailed description. |
| 88 bool GetWork(SequencedTask* task, | 77 bool GetWork(SequencedTask* task, |
| 89 std::vector<base::Closure>* delete_these_outside_lock); | 78 std::vector<base::Closure>* delete_these_outside_lock); |
| 90 | 79 |
| 91 // Peforms init and cleanup around running the given task. WillRun... | 80 // Peforms init and cleanup around running the given task. WillRun... |
| 92 // returns the value from PrepareToStartAdditionalThreadIfNecessary. | 81 // returns the value from PrepareToStartAdditionalThreadIfNecessary. |
| 93 // The calling code should call FinishStartingAdditionalThread once the | 82 // The calling code should call FinishStartingAdditionalThread once the |
| 94 // lock is released if the return values is nonzero. | 83 // lock is released if the return values is nonzero. |
| (...skipping 144 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 239 SequencedWorkerPool::Inner::GetSequenceToken() { | 228 SequencedWorkerPool::Inner::GetSequenceToken() { |
| 240 base::subtle::Atomic32 result = | 229 base::subtle::Atomic32 result = |
| 241 base::subtle::NoBarrier_AtomicIncrement(&last_sequence_number_, 1); | 230 base::subtle::NoBarrier_AtomicIncrement(&last_sequence_number_, 1); |
| 242 return SequenceToken(static_cast<int>(result)); | 231 return SequenceToken(static_cast<int>(result)); |
| 243 } | 232 } |
| 244 | 233 |
| 245 SequencedWorkerPool::SequenceToken | 234 SequencedWorkerPool::SequenceToken |
| 246 SequencedWorkerPool::Inner::GetNamedSequenceToken( | 235 SequencedWorkerPool::Inner::GetNamedSequenceToken( |
| 247 const std::string& name) { | 236 const std::string& name) { |
| 248 base::AutoLock lock(lock_); | 237 base::AutoLock lock(lock_); |
| 249 return SequenceToken(LockedGetNamedTokenID(name)); | 238 std::map<std::string, int>::const_iterator found = |
| 239 named_sequence_tokens_.find(name); |
| 240 if (found != named_sequence_tokens_.end()) |
| 241 return SequenceToken(found->second); // Got an existing one. |
| 242 |
| 243 // Create a new one for this name. |
| 244 SequenceToken result = GetSequenceToken(); |
| 245 named_sequence_tokens_.insert(std::make_pair(name, result.id_)); |
| 246 return result; |
| 250 } | 247 } |
| 251 | 248 |
| 252 bool SequencedWorkerPool::Inner::PostTask( | 249 bool SequencedWorkerPool::Inner::PostTask( |
| 253 const std::string* optional_token_name, | |
| 254 int sequence_token_id, | 250 int sequence_token_id, |
| 255 SequencedWorkerPool::WorkerShutdown shutdown_behavior, | 251 SequencedWorkerPool::WorkerShutdown shutdown_behavior, |
| 256 const tracked_objects::Location& from_here, | 252 const tracked_objects::Location& from_here, |
| 257 const base::Closure& task) { | 253 const base::Closure& task) { |
| 258 SequencedTask sequenced; | 254 SequencedTask sequenced; |
| 259 sequenced.sequence_token_id = sequence_token_id; | 255 sequenced.sequence_token_id = sequence_token_id; |
| 260 sequenced.shutdown_behavior = shutdown_behavior; | 256 sequenced.shutdown_behavior = shutdown_behavior; |
| 261 sequenced.location = from_here; | 257 sequenced.location = from_here; |
| 262 sequenced.task = task; | 258 sequenced.task = task; |
| 263 | 259 |
| 264 int create_thread_id = 0; | 260 int create_thread_id = 0; |
| 265 { | 261 { |
| 266 base::AutoLock lock(lock_); | 262 base::AutoLock lock(lock_); |
| 267 if (terminating_) | 263 if (terminating_) |
| 268 return false; | 264 return false; |
| 269 | 265 |
| 270 // Now that we have the lock, apply the named token rules. | |
| 271 if (optional_token_name) | |
| 272 sequenced.sequence_token_id = LockedGetNamedTokenID(*optional_token_name); | |
| 273 | |
| 274 pending_tasks_.push_back(sequenced); | 266 pending_tasks_.push_back(sequenced); |
| 275 pending_task_count_++; | 267 pending_task_count_++; |
| 276 if (shutdown_behavior == BLOCK_SHUTDOWN) | 268 if (shutdown_behavior == BLOCK_SHUTDOWN) |
| 277 blocking_shutdown_pending_task_count_++; | 269 blocking_shutdown_pending_task_count_++; |
| 278 | 270 |
| 279 create_thread_id = PrepareToStartAdditionalThreadIfHelpful(); | 271 create_thread_id = PrepareToStartAdditionalThreadIfHelpful(); |
| 280 } | 272 } |
| 281 | 273 |
| 282 // Actually start the additional thread or signal an existing one now that | 274 // Actually start the additional thread or signal an existing one now that |
| 283 // we're outside the lock. | 275 // we're outside the lock. |
| (...skipping 94 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 378 waiting_thread_count_--; | 370 waiting_thread_count_--; |
| 379 } | 371 } |
| 380 } | 372 } |
| 381 } | 373 } |
| 382 | 374 |
| 383 // We noticed we should exit. Wake up the next worker so it knows it should | 375 // We noticed we should exit. Wake up the next worker so it knows it should |
| 384 // exit as well (because the Shutdown() code only signals once). | 376 // exit as well (because the Shutdown() code only signals once). |
| 385 cond_var_.Signal(); | 377 cond_var_.Signal(); |
| 386 } | 378 } |
| 387 | 379 |
| 388 int SequencedWorkerPool::Inner::LockedGetNamedTokenID( | |
| 389 const std::string& name) { | |
| 390 lock_.AssertAcquired(); | |
| 391 DCHECK(!name.empty()); | |
| 392 | |
| 393 std::map<std::string, int>::const_iterator found = | |
| 394 named_sequence_tokens_.find(name); | |
| 395 if (found != named_sequence_tokens_.end()) | |
| 396 return found->second; // Got an existing one. | |
| 397 | |
| 398 // Create a new one for this name. | |
| 399 SequenceToken result = GetSequenceToken(); | |
| 400 named_sequence_tokens_.insert(std::make_pair(name, result.id_)); | |
| 401 return result.id_; | |
| 402 } | |
| 403 | |
| 404 bool SequencedWorkerPool::Inner::GetWork( | 380 bool SequencedWorkerPool::Inner::GetWork( |
| 405 SequencedTask* task, | 381 SequencedTask* task, |
| 406 std::vector<base::Closure>* delete_these_outside_lock) { | 382 std::vector<base::Closure>* delete_these_outside_lock) { |
| 407 lock_.AssertAcquired(); | 383 lock_.AssertAcquired(); |
| 408 | 384 |
| 409 DCHECK_EQ(pending_tasks_.size(), pending_task_count_); | 385 DCHECK_EQ(pending_tasks_.size(), pending_task_count_); |
| 410 UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount", | 386 UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount", |
| 411 static_cast<int>(pending_task_count_)); | 387 static_cast<int>(pending_task_count_)); |
| 412 | 388 |
| 413 // Find the next task with a sequence token that's not currently in use. | 389 // Find the next task with a sequence token that's not currently in use. |
| (...skipping 196 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 610 } | 586 } |
| 611 | 587 |
| 612 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken( | 588 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken( |
| 613 const std::string& name) { | 589 const std::string& name) { |
| 614 return inner_->GetNamedSequenceToken(name); | 590 return inner_->GetNamedSequenceToken(name); |
| 615 } | 591 } |
| 616 | 592 |
| 617 bool SequencedWorkerPool::PostWorkerTask( | 593 bool SequencedWorkerPool::PostWorkerTask( |
| 618 const tracked_objects::Location& from_here, | 594 const tracked_objects::Location& from_here, |
| 619 const base::Closure& task) { | 595 const base::Closure& task) { |
| 620 return inner_->PostTask(NULL, 0, BLOCK_SHUTDOWN, from_here, task); | 596 return inner_->PostTask(0, BLOCK_SHUTDOWN, from_here, task); |
| 621 } | 597 } |
| 622 | 598 |
| 623 bool SequencedWorkerPool::PostWorkerTaskWithShutdownBehavior( | 599 bool SequencedWorkerPool::PostWorkerTaskWithShutdownBehavior( |
| 624 const tracked_objects::Location& from_here, | 600 const tracked_objects::Location& from_here, |
| 625 const base::Closure& task, | 601 const base::Closure& task, |
| 626 WorkerShutdown shutdown_behavior) { | 602 WorkerShutdown shutdown_behavior) { |
| 627 return inner_->PostTask(NULL, 0, shutdown_behavior, from_here, task); | 603 return inner_->PostTask(0, shutdown_behavior, from_here, task); |
| 628 } | 604 } |
| 629 | 605 |
| 630 bool SequencedWorkerPool::PostSequencedWorkerTask( | 606 bool SequencedWorkerPool::PostSequencedWorkerTask( |
| 631 SequenceToken sequence_token, | 607 SequenceToken sequence_token, |
| 632 const tracked_objects::Location& from_here, | 608 const tracked_objects::Location& from_here, |
| 633 const base::Closure& task) { | 609 const base::Closure& task) { |
| 634 return inner_->PostTask(NULL, sequence_token.id_, BLOCK_SHUTDOWN, | 610 return inner_->PostTask(sequence_token.id_, BLOCK_SHUTDOWN, |
| 635 from_here, task); | 611 from_here, task); |
| 636 } | 612 } |
| 637 | 613 |
| 638 bool SequencedWorkerPool::PostNamedSequencedWorkerTask( | |
| 639 const std::string& token_name, | |
| 640 const tracked_objects::Location& from_here, | |
| 641 const base::Closure& task) { | |
| 642 DCHECK(!token_name.empty()); | |
| 643 return inner_->PostTask(&token_name, 0, BLOCK_SHUTDOWN, from_here, task); | |
| 644 } | |
| 645 | |
| 646 bool SequencedWorkerPool::PostSequencedWorkerTaskWithShutdownBehavior( | 614 bool SequencedWorkerPool::PostSequencedWorkerTaskWithShutdownBehavior( |
| 647 SequenceToken sequence_token, | 615 SequenceToken sequence_token, |
| 648 const tracked_objects::Location& from_here, | 616 const tracked_objects::Location& from_here, |
| 649 const base::Closure& task, | 617 const base::Closure& task, |
| 650 WorkerShutdown shutdown_behavior) { | 618 WorkerShutdown shutdown_behavior) { |
| 651 return inner_->PostTask(NULL, sequence_token.id_, shutdown_behavior, | 619 return inner_->PostTask(sequence_token.id_, shutdown_behavior, |
| 652 from_here, task); | 620 from_here, task); |
| 653 } | 621 } |
| 654 | 622 |
| 655 void SequencedWorkerPool::Shutdown() { | 623 void SequencedWorkerPool::Shutdown() { |
| 656 inner_->Shutdown(); | 624 inner_->Shutdown(); |
| 657 } | 625 } |
| 658 | 626 |
| 659 void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) { | 627 void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) { |
| 660 inner_->SetTestingObserver(observer); | 628 inner_->SetTestingObserver(observer); |
| 661 } | 629 } |
| 662 | 630 |
| 663 } // namespace base | 631 } // namespace base |
| OLD | NEW |