Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(1751)

Side by Side Diff: base/threading/sequenced_worker_pool.cc

Issue 18650006: base: Make SequencedWorkerPool issue globally unique SequenceTokens. (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@master
Patch Set: base: SequencedWorkerPool Globally Unique Tokens with unused-variable fix. Created 7 years, 5 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
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>
11 #include <vector> 11 #include <vector>
12 12
13 #include "base/atomicops.h" 13 #include "base/atomic_sequence_num.h"
14 #include "base/callback.h" 14 #include "base/callback.h"
15 #include "base/compiler_specific.h" 15 #include "base/compiler_specific.h"
16 #include "base/critical_closure.h" 16 #include "base/critical_closure.h"
17 #include "base/debug/trace_event.h" 17 #include "base/debug/trace_event.h"
18 #include "base/lazy_instance.h"
18 #include "base/logging.h" 19 #include "base/logging.h"
19 #include "base/memory/linked_ptr.h" 20 #include "base/memory/linked_ptr.h"
20 #include "base/message_loop/message_loop_proxy.h" 21 #include "base/message_loop/message_loop_proxy.h"
21 #include "base/metrics/histogram.h"
22 #include "base/stl_util.h" 22 #include "base/stl_util.h"
23 #include "base/strings/stringprintf.h" 23 #include "base/strings/stringprintf.h"
24 #include "base/synchronization/condition_variable.h" 24 #include "base/synchronization/condition_variable.h"
25 #include "base/synchronization/lock.h" 25 #include "base/synchronization/lock.h"
26 #include "base/threading/platform_thread.h" 26 #include "base/threading/platform_thread.h"
27 #include "base/threading/simple_thread.h" 27 #include "base/threading/simple_thread.h"
28 #include "base/threading/thread_local.h"
28 #include "base/threading/thread_restrictions.h" 29 #include "base/threading/thread_restrictions.h"
29 #include "base/time/time.h" 30 #include "base/time/time.h"
30 #include "base/tracked_objects.h" 31 #include "base/tracked_objects.h"
31 32
32 #if defined(OS_MACOSX) 33 #if defined(OS_MACOSX)
33 #include "base/mac/scoped_nsautorelease_pool.h" 34 #include "base/mac/scoped_nsautorelease_pool.h"
34 #endif 35 #endif
35 36
37 #if !defined(OS_NACL)
38 #include "base/metrics/histogram.h"
39 #endif
40
36 namespace base { 41 namespace base {
37 42
38 namespace { 43 namespace {
39 44
40 struct SequencedTask : public TrackingInfo { 45 struct SequencedTask : public TrackingInfo {
41 SequencedTask() 46 SequencedTask()
42 : sequence_token_id(0), 47 : sequence_token_id(0),
43 trace_id(0), 48 trace_id(0),
44 sequence_task_number(0), 49 sequence_task_number(0),
45 shutdown_behavior(SequencedWorkerPool::BLOCK_SHUTDOWN) {} 50 shutdown_behavior(SequencedWorkerPool::BLOCK_SHUTDOWN) {}
(...skipping 160 matching lines...) Expand 10 before | Expand all | Expand 10 after
206 211
207 // Create a process-wide unique ID to represent this task in trace events. This 212 // Create a process-wide unique ID to represent this task in trace events. This
208 // will be mangled with a Process ID hash to reduce the likelyhood of colliding 213 // will be mangled with a Process ID hash to reduce the likelyhood of colliding
209 // with MessageLoop pointers on other processes. 214 // with MessageLoop pointers on other processes.
210 uint64 GetTaskTraceID(const SequencedTask& task, 215 uint64 GetTaskTraceID(const SequencedTask& task,
211 void* pool) { 216 void* pool) {
212 return (static_cast<uint64>(task.trace_id) << 32) | 217 return (static_cast<uint64>(task.trace_id) << 32) |
213 static_cast<uint64>(reinterpret_cast<intptr_t>(pool)); 218 static_cast<uint64>(reinterpret_cast<intptr_t>(pool));
214 } 219 }
215 220
221 base::LazyInstance<base::ThreadLocalPointer<
222 SequencedWorkerPool::SequenceToken> > g_lazy_tls_ptr =
223 LAZY_INSTANCE_INITIALIZER;
224
216 } // namespace 225 } // namespace
217 226
218 // Worker --------------------------------------------------------------------- 227 // Worker ---------------------------------------------------------------------
219 228
220 class SequencedWorkerPool::Worker : public SimpleThread { 229 class SequencedWorkerPool::Worker : public SimpleThread {
221 public: 230 public:
222 // Hold a (cyclic) ref to |worker_pool|, since we want to keep it 231 // Hold a (cyclic) ref to |worker_pool|, since we want to keep it
223 // around as long as we are running. 232 // around as long as we are running.
224 Worker(const scoped_refptr<SequencedWorkerPool>& worker_pool, 233 Worker(const scoped_refptr<SequencedWorkerPool>& worker_pool,
225 int thread_number, 234 int thread_number,
(...skipping 149 matching lines...) Expand 10 before | Expand all | Expand 10 after
375 void SignalHasWork(); 384 void SignalHasWork();
376 385
377 // Checks whether there is work left that's blocking shutdown. Must be 386 // Checks whether there is work left that's blocking shutdown. Must be
378 // called inside the lock. 387 // called inside the lock.
379 bool CanShutdown() const; 388 bool CanShutdown() const;
380 389
381 SequencedWorkerPool* const worker_pool_; 390 SequencedWorkerPool* const worker_pool_;
382 391
383 // The last sequence number used. Managed by GetSequenceToken, since this 392 // The last sequence number used. Managed by GetSequenceToken, since this
384 // only does threadsafe increment operations, you do not need to hold the 393 // only does threadsafe increment operations, you do not need to hold the
385 // lock. 394 // lock. This is class-static to make SequenceTokens issued by
386 volatile subtle::Atomic32 last_sequence_number_; 395 // GetSequenceToken unique across SequencedWorkerPool instances.
396 static base::StaticAtomicSequenceNumber g_last_sequence_number_;
387 397
388 // This lock protects |everything in this class|. Do not read or modify 398 // This lock protects |everything in this class|. Do not read or modify
389 // anything without holding this lock. Do not block while holding this 399 // anything without holding this lock. Do not block while holding this
390 // lock. 400 // lock.
391 mutable Lock lock_; 401 mutable Lock lock_;
392 402
393 // Condition variable that is waited on by worker threads until new 403 // Condition variable that is waited on by worker threads until new
394 // tasks are posted or shutdown starts. 404 // tasks are posted or shutdown starts.
395 ConditionVariable has_work_cv_; 405 ConditionVariable has_work_cv_;
396 406
(...skipping 75 matching lines...) Expand 10 before | Expand all | Expand 10 after
472 prefix + StringPrintf("Worker%d", thread_number).c_str()), 482 prefix + StringPrintf("Worker%d", thread_number).c_str()),
473 worker_pool_(worker_pool), 483 worker_pool_(worker_pool),
474 running_shutdown_behavior_(CONTINUE_ON_SHUTDOWN) { 484 running_shutdown_behavior_(CONTINUE_ON_SHUTDOWN) {
475 Start(); 485 Start();
476 } 486 }
477 487
478 SequencedWorkerPool::Worker::~Worker() { 488 SequencedWorkerPool::Worker::~Worker() {
479 } 489 }
480 490
481 void SequencedWorkerPool::Worker::Run() { 491 void SequencedWorkerPool::Worker::Run() {
492 // Store a pointer to the running sequence in thread local storage for
493 // static function access.
494 g_lazy_tls_ptr.Get().Set(&running_sequence_);
495
482 // Just jump back to the Inner object to run the thread, since it has all the 496 // Just jump back to the Inner object to run the thread, since it has all the
483 // tracking information and queues. It might be more natural to implement 497 // tracking information and queues. It might be more natural to implement
484 // using DelegateSimpleThread and have Inner implement the Delegate to avoid 498 // using DelegateSimpleThread and have Inner implement the Delegate to avoid
485 // having these worker objects at all, but that method lacks the ability to 499 // having these worker objects at all, but that method lacks the ability to
486 // send thread-specific information easily to the thread loop. 500 // send thread-specific information easily to the thread loop.
487 worker_pool_->inner_->ThreadLoop(this); 501 worker_pool_->inner_->ThreadLoop(this);
488 // Release our cyclic reference once we're done. 502 // Release our cyclic reference once we're done.
489 worker_pool_ = NULL; 503 worker_pool_ = NULL;
490 } 504 }
491 505
492 // Inner definitions --------------------------------------------------------- 506 // Inner definitions ---------------------------------------------------------
493 507
494 SequencedWorkerPool::Inner::Inner( 508 SequencedWorkerPool::Inner::Inner(
495 SequencedWorkerPool* worker_pool, 509 SequencedWorkerPool* worker_pool,
496 size_t max_threads, 510 size_t max_threads,
497 const std::string& thread_name_prefix, 511 const std::string& thread_name_prefix,
498 TestingObserver* observer) 512 TestingObserver* observer)
499 : worker_pool_(worker_pool), 513 : worker_pool_(worker_pool),
500 last_sequence_number_(0),
501 lock_(), 514 lock_(),
502 has_work_cv_(&lock_), 515 has_work_cv_(&lock_),
503 can_shutdown_cv_(&lock_), 516 can_shutdown_cv_(&lock_),
504 max_threads_(max_threads), 517 max_threads_(max_threads),
505 thread_name_prefix_(thread_name_prefix), 518 thread_name_prefix_(thread_name_prefix),
506 thread_being_created_(false), 519 thread_being_created_(false),
507 waiting_thread_count_(0), 520 waiting_thread_count_(0),
508 blocking_shutdown_thread_count_(0), 521 blocking_shutdown_thread_count_(0),
509 next_sequence_task_number_(0), 522 next_sequence_task_number_(0),
510 blocking_shutdown_pending_task_count_(0), 523 blocking_shutdown_pending_task_count_(0),
(...skipping 14 matching lines...) Expand all
525 for (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it) 538 for (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it)
526 it->second->Join(); 539 it->second->Join();
527 threads_.clear(); 540 threads_.clear();
528 541
529 if (testing_observer_) 542 if (testing_observer_)
530 testing_observer_->OnDestruct(); 543 testing_observer_->OnDestruct();
531 } 544 }
532 545
533 SequencedWorkerPool::SequenceToken 546 SequencedWorkerPool::SequenceToken
534 SequencedWorkerPool::Inner::GetSequenceToken() { 547 SequencedWorkerPool::Inner::GetSequenceToken() {
535 subtle::Atomic32 result = 548 // Need to add one because StaticAtomicSequenceNumber starts at zero, which
536 subtle::NoBarrier_AtomicIncrement(&last_sequence_number_, 1); 549 // is used as a sentinel value in SequenceTokens.
537 return SequenceToken(static_cast<int>(result)); 550 return SequenceToken(g_last_sequence_number_.GetNext() + 1);
538 } 551 }
539 552
540 SequencedWorkerPool::SequenceToken 553 SequencedWorkerPool::SequenceToken
541 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) { 554 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) {
542 AutoLock lock(lock_); 555 AutoLock lock(lock_);
543 return SequenceToken(LockedGetNamedTokenID(name)); 556 return SequenceToken(LockedGetNamedTokenID(name));
544 } 557 }
545 558
546 bool SequencedWorkerPool::Inner::PostTask( 559 bool SequencedWorkerPool::Inner::PostTask(
547 const std::string* optional_token_name, 560 const std::string* optional_token_name,
(...skipping 60 matching lines...) Expand 10 before | Expand all | Expand 10 after
608 AutoLock lock(lock_); 621 AutoLock lock(lock_);
609 return ContainsKey(threads_, PlatformThread::CurrentId()); 622 return ContainsKey(threads_, PlatformThread::CurrentId());
610 } 623 }
611 624
612 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread( 625 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
613 SequenceToken sequence_token) const { 626 SequenceToken sequence_token) const {
614 AutoLock lock(lock_); 627 AutoLock lock(lock_);
615 ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId()); 628 ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId());
616 if (found == threads_.end()) 629 if (found == threads_.end())
617 return false; 630 return false;
618 return found->second->running_sequence().Equals(sequence_token); 631 return sequence_token.Equals(found->second->running_sequence());
619 } 632 }
620 633
621 // See https://code.google.com/p/chromium/issues/detail?id=168415 634 // See https://code.google.com/p/chromium/issues/detail?id=168415
622 void SequencedWorkerPool::Inner::CleanupForTesting() { 635 void SequencedWorkerPool::Inner::CleanupForTesting() {
623 DCHECK(!RunsTasksOnCurrentThread()); 636 DCHECK(!RunsTasksOnCurrentThread());
624 base::ThreadRestrictions::ScopedAllowWait allow_wait; 637 base::ThreadRestrictions::ScopedAllowWait allow_wait;
625 AutoLock lock(lock_); 638 AutoLock lock(lock_);
626 CHECK_EQ(CLEANUP_DONE, cleanup_state_); 639 CHECK_EQ(CLEANUP_DONE, cleanup_state_);
627 if (shutdown_called_) 640 if (shutdown_called_)
628 return; 641 return;
(...skipping 30 matching lines...) Expand all
659 if (CanShutdown()) 672 if (CanShutdown())
660 return; 673 return;
661 } 674 }
662 675
663 // If we're here, then something is blocking shutdown. So wait for 676 // If we're here, then something is blocking shutdown. So wait for
664 // CanShutdown() to go to true. 677 // CanShutdown() to go to true.
665 678
666 if (testing_observer_) 679 if (testing_observer_)
667 testing_observer_->WillWaitForShutdown(); 680 testing_observer_->WillWaitForShutdown();
668 681
682 #if !defined(OS_NACL)
669 TimeTicks shutdown_wait_begin = TimeTicks::Now(); 683 TimeTicks shutdown_wait_begin = TimeTicks::Now();
684 #endif
670 685
671 { 686 {
672 base::ThreadRestrictions::ScopedAllowWait allow_wait; 687 base::ThreadRestrictions::ScopedAllowWait allow_wait;
673 AutoLock lock(lock_); 688 AutoLock lock(lock_);
674 while (!CanShutdown()) 689 while (!CanShutdown())
675 can_shutdown_cv_.Wait(); 690 can_shutdown_cv_.Wait();
676 } 691 }
692 #if !defined(OS_NACL)
677 UMA_HISTOGRAM_TIMES("SequencedWorkerPool.ShutdownDelayTime", 693 UMA_HISTOGRAM_TIMES("SequencedWorkerPool.ShutdownDelayTime",
678 TimeTicks::Now() - shutdown_wait_begin); 694 TimeTicks::Now() - shutdown_wait_begin);
695 #endif
679 } 696 }
680 697
681 void SequencedWorkerPool::Inner::ThreadLoop(Worker* this_worker) { 698 void SequencedWorkerPool::Inner::ThreadLoop(Worker* this_worker) {
682 { 699 {
683 AutoLock lock(lock_); 700 AutoLock lock(lock_);
684 DCHECK(thread_being_created_); 701 DCHECK(thread_being_created_);
685 thread_being_created_ = false; 702 thread_being_created_ = false;
686 std::pair<ThreadMap::iterator, bool> result = 703 std::pair<ThreadMap::iterator, bool> result =
687 threads_.insert( 704 threads_.insert(
688 std::make_pair(this_worker->tid(), make_linked_ptr(this_worker))); 705 std::make_pair(this_worker->tid(), make_linked_ptr(this_worker)));
(...skipping 176 matching lines...) Expand 10 before | Expand all | Expand 10 after
865 return CONTINUE_ON_SHUTDOWN; 882 return CONTINUE_ON_SHUTDOWN;
866 return found->second->running_shutdown_behavior(); 883 return found->second->running_shutdown_behavior();
867 } 884 }
868 885
869 SequencedWorkerPool::Inner::GetWorkStatus SequencedWorkerPool::Inner::GetWork( 886 SequencedWorkerPool::Inner::GetWorkStatus SequencedWorkerPool::Inner::GetWork(
870 SequencedTask* task, 887 SequencedTask* task,
871 TimeDelta* wait_time, 888 TimeDelta* wait_time,
872 std::vector<Closure>* delete_these_outside_lock) { 889 std::vector<Closure>* delete_these_outside_lock) {
873 lock_.AssertAcquired(); 890 lock_.AssertAcquired();
874 891
892 #if !defined(OS_NACL)
875 UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount", 893 UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount",
876 static_cast<int>(pending_tasks_.size())); 894 static_cast<int>(pending_tasks_.size()));
895 #endif
877 896
878 // Find the next task with a sequence token that's not currently in use. 897 // Find the next task with a sequence token that's not currently in use.
879 // If the token is in use, that means another thread is running something 898 // If the token is in use, that means another thread is running something
880 // in that sequence, and we can't run it without going out-of-order. 899 // in that sequence, and we can't run it without going out-of-order.
881 // 900 //
882 // This algorithm is simple and fair, but inefficient in some cases. For 901 // This algorithm is simple and fair, but inefficient in some cases. For
883 // example, say somebody schedules 1000 slow tasks with the same sequence 902 // example, say somebody schedules 1000 slow tasks with the same sequence
884 // number. We'll have to go through all those tasks each time we feel like 903 // number. We'll have to go through all those tasks each time we feel like
885 // there might be work to schedule. If this proves to be a problem, we 904 // there might be work to schedule. If this proves to be a problem, we
886 // should make this more efficient. 905 // should make this more efficient.
(...skipping 64 matching lines...) Expand 10 before | Expand all | Expand 10 after
951 blocking_shutdown_pending_task_count_--; 970 blocking_shutdown_pending_task_count_--;
952 } 971 }
953 972
954 status = GET_WORK_FOUND; 973 status = GET_WORK_FOUND;
955 break; 974 break;
956 } 975 }
957 976
958 // Track the number of tasks we had to skip over to see if we should be 977 // Track the number of tasks we had to skip over to see if we should be
959 // making this more efficient. If this number ever becomes large or is 978 // making this more efficient. If this number ever becomes large or is
960 // frequently "some", we should consider the optimization above. 979 // frequently "some", we should consider the optimization above.
980 #if !defined(OS_NACL)
961 UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.UnrunnableTaskCount", 981 UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.UnrunnableTaskCount",
962 unrunnable_tasks); 982 unrunnable_tasks);
983 #endif
963 return status; 984 return status;
964 } 985 }
965 986
966 int SequencedWorkerPool::Inner::WillRunWorkerTask(const SequencedTask& task) { 987 int SequencedWorkerPool::Inner::WillRunWorkerTask(const SequencedTask& task) {
967 lock_.AssertAcquired(); 988 lock_.AssertAcquired();
968 989
969 // Mark the task's sequence number as in use. 990 // Mark the task's sequence number as in use.
970 if (task.sequence_token_id) 991 if (task.sequence_token_id)
971 current_sequences_.insert(task.sequence_token_id); 992 current_sequences_.insert(task.sequence_token_id);
972 993
(...skipping 108 matching lines...) Expand 10 before | Expand all | Expand 10 after
1081 } 1102 }
1082 1103
1083 bool SequencedWorkerPool::Inner::CanShutdown() const { 1104 bool SequencedWorkerPool::Inner::CanShutdown() const {
1084 lock_.AssertAcquired(); 1105 lock_.AssertAcquired();
1085 // See PrepareToStartAdditionalThreadIfHelpful for how thread creation works. 1106 // See PrepareToStartAdditionalThreadIfHelpful for how thread creation works.
1086 return !thread_being_created_ && 1107 return !thread_being_created_ &&
1087 blocking_shutdown_thread_count_ == 0 && 1108 blocking_shutdown_thread_count_ == 0 &&
1088 blocking_shutdown_pending_task_count_ == 0; 1109 blocking_shutdown_pending_task_count_ == 0;
1089 } 1110 }
1090 1111
1112 base::StaticAtomicSequenceNumber
1113 SequencedWorkerPool::Inner::g_last_sequence_number_;
1114
1091 // SequencedWorkerPool -------------------------------------------------------- 1115 // SequencedWorkerPool --------------------------------------------------------
1092 1116
1117 // static
1118 SequencedWorkerPool::SequenceToken
1119 SequencedWorkerPool::GetSequenceTokenForCurrentThread() {
1120 SequencedWorkerPool::SequenceToken* token = g_lazy_tls_ptr.Get().Get();
1121 if (!token)
1122 return SequenceToken();
1123 return *token;
1124 }
1125
1093 SequencedWorkerPool::SequencedWorkerPool( 1126 SequencedWorkerPool::SequencedWorkerPool(
1094 size_t max_threads, 1127 size_t max_threads,
1095 const std::string& thread_name_prefix) 1128 const std::string& thread_name_prefix)
1096 : constructor_message_loop_(MessageLoopProxy::current()), 1129 : constructor_message_loop_(MessageLoopProxy::current()),
1097 inner_(new Inner(this, max_threads, thread_name_prefix, NULL)) { 1130 inner_(new Inner(this, max_threads, thread_name_prefix, NULL)) {
1098 } 1131 }
1099 1132
1100 SequencedWorkerPool::SequencedWorkerPool( 1133 SequencedWorkerPool::SequencedWorkerPool(
1101 size_t max_threads, 1134 size_t max_threads,
1102 const std::string& thread_name_prefix, 1135 const std::string& thread_name_prefix,
(...skipping 127 matching lines...) Expand 10 before | Expand all | Expand 10 after
1230 void SequencedWorkerPool::SignalHasWorkForTesting() { 1263 void SequencedWorkerPool::SignalHasWorkForTesting() {
1231 inner_->SignalHasWorkForTesting(); 1264 inner_->SignalHasWorkForTesting();
1232 } 1265 }
1233 1266
1234 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) { 1267 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) {
1235 DCHECK(constructor_message_loop_->BelongsToCurrentThread()); 1268 DCHECK(constructor_message_loop_->BelongsToCurrentThread());
1236 inner_->Shutdown(max_new_blocking_tasks_after_shutdown); 1269 inner_->Shutdown(max_new_blocking_tasks_after_shutdown);
1237 } 1270 }
1238 1271
1239 } // namespace base 1272 } // namespace base
OLDNEW
« no previous file with comments | « base/threading/sequenced_worker_pool.h ('k') | base/threading/sequenced_worker_pool_unittest.cc » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698