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 3 // found in the LICENSE file. 4 5 // Multi-threaded tests of ConditionVariable class. 6 7 #include "base/synchronization/condition_variable.h" 8 9 #include <time.h> 10 11 #include <algorithm> 12 #include <memory> 13 #include <vector> 14 15 #include "base/bind.h" 16 #include "base/location.h" 17 #include "base/logging.h" 18 #include "base/single_thread_task_runner.h" 19 #include "base/synchronization/lock.h" 20 #include "base/synchronization/spin_wait.h" 21 #include "base/threading/platform_thread.h" 22 #include "base/threading/thread.h" 23 #include "base/threading/thread_collision_warner.h" 24 #include "base/time/time.h" 25 #include "build/build_config.h" 26 #include "testing/gtest/include/gtest/gtest.h" 27 #include "testing/platform_test.h" 28 29 namespace base { 30 31 namespace { 32 //------------------------------------------------------------------------------ 33 // Define our test class, with several common variables. 34 //------------------------------------------------------------------------------ 35 36 class ConditionVariableTest : public PlatformTest { 37 public: 38 const TimeDelta kZeroMs; 39 const TimeDelta kTenMs; 40 const TimeDelta kThirtyMs; 41 const TimeDelta kFortyFiveMs; 42 const TimeDelta kSixtyMs; 43 const TimeDelta kOneHundredMs; 44 45 ConditionVariableTest() 46 : kZeroMs(TimeDelta::FromMilliseconds(0)), 47 kTenMs(TimeDelta::FromMilliseconds(10)), 48 kThirtyMs(TimeDelta::FromMilliseconds(30)), 49 kFortyFiveMs(TimeDelta::FromMilliseconds(45)), 50 kSixtyMs(TimeDelta::FromMilliseconds(60)), 51 kOneHundredMs(TimeDelta::FromMilliseconds(100)) { 52 } 53 }; 54 55 //------------------------------------------------------------------------------ 56 // Define a class that will control activities an several multi-threaded tests. 57 // The general structure of multi-threaded tests is that a test case will 58 // construct an instance of a WorkQueue. The WorkQueue will spin up some 59 // threads and control them throughout their lifetime, as well as maintaining 60 // a central repository of the work thread's activity. Finally, the WorkQueue 61 // will command the the worker threads to terminate. At that point, the test 62 // cases will validate that the WorkQueue has records showing that the desired 63 // activities were performed. 64 //------------------------------------------------------------------------------ 65 66 // Callers are responsible for synchronizing access to the following class. 67 // The WorkQueue::lock_, as accessed via WorkQueue::lock(), should be used for 68 // all synchronized access. 69 class WorkQueue : public PlatformThread::Delegate { 70 public: 71 explicit WorkQueue(int thread_count); 72 ~WorkQueue() override; 73 74 // PlatformThread::Delegate interface. 75 void ThreadMain() override; 76 77 //---------------------------------------------------------------------------- 78 // Worker threads only call the following methods. 79 // They should use the lock to get exclusive access. 80 int GetThreadId(); // Get an ID assigned to a thread.. 81 bool EveryIdWasAllocated() const; // Indicates that all IDs were handed out. 82 TimeDelta GetAnAssignment(int thread_id); // Get a work task duration. 83 void WorkIsCompleted(int thread_id); 84 85 int task_count() const; 86 bool allow_help_requests() const; // Workers can signal more workers. 87 bool shutdown() const; // Check if shutdown has been requested. 88 89 void thread_shutting_down(); 90 91 92 //---------------------------------------------------------------------------- 93 // Worker threads can call them but not needed to acquire a lock. 94 Lock* lock(); 95 96 ConditionVariable* work_is_available(); 97 ConditionVariable* all_threads_have_ids(); 98 ConditionVariable* no_more_tasks(); 99 100 //---------------------------------------------------------------------------- 101 // The rest of the methods are for use by the controlling master thread (the 102 // test case code). 103 void ResetHistory(); 104 int GetMinCompletionsByWorkerThread() const; 105 int GetMaxCompletionsByWorkerThread() const; 106 int GetNumThreadsTakingAssignments() const; 107 int GetNumThreadsCompletingTasks() const; 108 int GetNumberOfCompletedTasks() const; 109 110 void SetWorkTime(TimeDelta delay); 111 void SetTaskCount(int count); 112 void SetAllowHelp(bool allow); 113 114 // The following must be called without locking, and will spin wait until the 115 // threads are all in a wait state. 116 void SpinUntilAllThreadsAreWaiting(); 117 void SpinUntilTaskCountLessThan(int task_count); 118 119 // Caller must acquire lock before calling. 120 void SetShutdown(); 121 122 // Compares the |shutdown_task_count_| to the |thread_count| and returns true 123 // if they are equal. This check will acquire the |lock_| so the caller 124 // should not hold the lock when calling this method. 125 bool ThreadSafeCheckShutdown(int thread_count); 126 127 private: 128 // Both worker threads and controller use the following to synchronize. 129 Lock lock_; 130 ConditionVariable work_is_available_; // To tell threads there is work. 131 132 // Conditions to notify the controlling process (if it is interested). 133 ConditionVariable all_threads_have_ids_; // All threads are running. 134 ConditionVariable no_more_tasks_; // Task count is zero. 135 136 const int thread_count_; 137 int waiting_thread_count_; 138 std::unique_ptr<PlatformThreadHandle[]> thread_handles_; 139 std::vector<int> assignment_history_; // Number of assignment per worker. 140 std::vector<int> completion_history_; // Number of completions per worker. 141 int thread_started_counter_; // Used to issue unique id to workers. 142 int shutdown_task_count_; // Number of tasks told to shutdown 143 int task_count_; // Number of assignment tasks waiting to be processed. 144 TimeDelta worker_delay_; // Time each task takes to complete. 145 bool allow_help_requests_; // Workers can signal more workers. 146 bool shutdown_; // Set when threads need to terminate. 147 148 DFAKE_MUTEX(locked_methods_); 149 }; 150 151 //------------------------------------------------------------------------------ 152 // The next section contains the actual tests. 153 //------------------------------------------------------------------------------ 154 155 TEST_F(ConditionVariableTest, StartupShutdownTest) { 156 Lock lock; 157 158 // First try trivial startup/shutdown. 159 { 160 ConditionVariable cv1(&lock); 161 } // Call for cv1 destruction. 162 163 // Exercise with at least a few waits. 164 ConditionVariable cv(&lock); 165 166 lock.Acquire(); 167 cv.TimedWait(kTenMs); // Wait for 10 ms. 168 cv.TimedWait(kTenMs); // Wait for 10 ms. 169 lock.Release(); 170 171 lock.Acquire(); 172 cv.TimedWait(kTenMs); // Wait for 10 ms. 173 cv.TimedWait(kTenMs); // Wait for 10 ms. 174 cv.TimedWait(kTenMs); // Wait for 10 ms. 175 lock.Release(); 176 } // Call for cv destruction. 177 178 TEST_F(ConditionVariableTest, TimeoutTest) { 179 Lock lock; 180 ConditionVariable cv(&lock); 181 lock.Acquire(); 182 183 TimeTicks start = TimeTicks::Now(); 184 const TimeDelta WAIT_TIME = TimeDelta::FromMilliseconds(300); 185 // Allow for clocking rate granularity. 186 const TimeDelta FUDGE_TIME = TimeDelta::FromMilliseconds(50); 187 188 cv.TimedWait(WAIT_TIME + FUDGE_TIME); 189 TimeDelta duration = TimeTicks::Now() - start; 190 // We can't use EXPECT_GE here as the TimeDelta class does not support the 191 // required stream conversion. 192 EXPECT_TRUE(duration >= WAIT_TIME); 193 194 lock.Release(); 195 } 196 197 #if defined(OS_POSIX) 198 const int kDiscontinuitySeconds = 2; 199 200 void BackInTime(Lock* lock) { 201 AutoLock auto_lock(*lock); 202 203 timeval tv; 204 gettimeofday(&tv, NULL); 205 tv.tv_sec -= kDiscontinuitySeconds; 206 settimeofday(&tv, NULL); 207 } 208 209 // Tests that TimedWait ignores changes to the system clock. 210 // Test is disabled by default, because it needs to run as root to muck with the 211 // system clock. 212 // http://crbug.com/293736 213 TEST_F(ConditionVariableTest, DISABLED_TimeoutAcrossSetTimeOfDay) { 214 timeval tv; 215 gettimeofday(&tv, NULL); 216 tv.tv_sec += kDiscontinuitySeconds; 217 if (settimeofday(&tv, NULL) < 0) { 218 PLOG(ERROR) << "Could not set time of day. Run as root?"; 219 return; 220 } 221 222 Lock lock; 223 ConditionVariable cv(&lock); 224 lock.Acquire(); 225 226 Thread thread("Helper"); 227 thread.Start(); 228 thread.task_runner()->PostTask(FROM_HERE, base::Bind(&BackInTime, &lock)); 229 230 TimeTicks start = TimeTicks::Now(); 231 const TimeDelta kWaitTime = TimeDelta::FromMilliseconds(300); 232 // Allow for clocking rate granularity. 233 const TimeDelta kFudgeTime = TimeDelta::FromMilliseconds(50); 234 235 cv.TimedWait(kWaitTime + kFudgeTime); 236 TimeDelta duration = TimeTicks::Now() - start; 237 238 thread.Stop(); 239 // We can't use EXPECT_GE here as the TimeDelta class does not support the 240 // required stream conversion. 241 EXPECT_TRUE(duration >= kWaitTime); 242 EXPECT_TRUE(duration <= TimeDelta::FromSeconds(kDiscontinuitySeconds)); 243 244 lock.Release(); 245 } 246 #endif 247 248 249 // Suddenly got flaky on Win, see http://crbug.com/10607 (starting at 250 // comment #15). 251 #if defined(OS_WIN) 252 #define MAYBE_MultiThreadConsumerTest DISABLED_MultiThreadConsumerTest 253 #else 254 #define MAYBE_MultiThreadConsumerTest MultiThreadConsumerTest 255 #endif 256 // Test serial task servicing, as well as two parallel task servicing methods. 257 TEST_F(ConditionVariableTest, MAYBE_MultiThreadConsumerTest) { 258 const int kThreadCount = 10; 259 WorkQueue queue(kThreadCount); // Start the threads. 260 261 const int kTaskCount = 10; // Number of tasks in each mini-test here. 262 263 Time start_time; // Used to time task processing. 264 265 { 266 base::AutoLock auto_lock(*queue.lock()); 267 while (!queue.EveryIdWasAllocated()) 268 queue.all_threads_have_ids()->Wait(); 269 } 270 271 // If threads aren't in a wait state, they may start to gobble up tasks in 272 // parallel, short-circuiting (breaking) this test. 273 queue.SpinUntilAllThreadsAreWaiting(); 274 275 { 276 // Since we have no tasks yet, all threads should be waiting by now. 277 base::AutoLock auto_lock(*queue.lock()); 278 EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments()); 279 EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks()); 280 EXPECT_EQ(0, queue.task_count()); 281 EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread()); 282 EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread()); 283 EXPECT_EQ(0, queue.GetNumberOfCompletedTasks()); 284 285 // Set up to make each task include getting help from another worker, so 286 // so that the work gets done in paralell. 287 queue.ResetHistory(); 288 queue.SetTaskCount(kTaskCount); 289 queue.SetWorkTime(kThirtyMs); 290 queue.SetAllowHelp(true); 291 292 start_time = Time::Now(); 293 } 294 295 queue.work_is_available()->Signal(); // But each worker can signal another. 296 // Wait till we at least start to handle tasks (and we're not all waiting). 297 queue.SpinUntilTaskCountLessThan(kTaskCount); 298 // Wait to allow the all workers to get done. 299 queue.SpinUntilAllThreadsAreWaiting(); 300 301 { 302 // Wait until all work tasks have at least been assigned. 303 base::AutoLock auto_lock(*queue.lock()); 304 while (queue.task_count()) 305 queue.no_more_tasks()->Wait(); 306 307 // To avoid racy assumptions, we'll just assert that at least 2 threads 308 // did work. We know that the first worker should have gone to sleep, and 309 // hence a second worker should have gotten an assignment. 310 EXPECT_LE(2, queue.GetNumThreadsTakingAssignments()); 311 EXPECT_EQ(kTaskCount, queue.GetNumberOfCompletedTasks()); 312 313 // Try to ask all workers to help, and only a few will do the work. 314 queue.ResetHistory(); 315 queue.SetTaskCount(3); 316 queue.SetWorkTime(kThirtyMs); 317 queue.SetAllowHelp(false); 318 } 319 queue.work_is_available()->Broadcast(); // Make them all try. 320 // Wait till we at least start to handle tasks (and we're not all waiting). 321 queue.SpinUntilTaskCountLessThan(3); 322 // Wait to allow the 3 workers to get done. 323 queue.SpinUntilAllThreadsAreWaiting(); 324 325 { 326 base::AutoLock auto_lock(*queue.lock()); 327 EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments()); 328 EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks()); 329 EXPECT_EQ(0, queue.task_count()); 330 EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread()); 331 EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread()); 332 EXPECT_EQ(3, queue.GetNumberOfCompletedTasks()); 333 334 // Set up to make each task get help from another worker. 335 queue.ResetHistory(); 336 queue.SetTaskCount(3); 337 queue.SetWorkTime(kThirtyMs); 338 queue.SetAllowHelp(true); // Allow (unnecessary) help requests. 339 } 340 queue.work_is_available()->Broadcast(); // Signal all threads. 341 // Wait till we at least start to handle tasks (and we're not all waiting). 342 queue.SpinUntilTaskCountLessThan(3); 343 // Wait to allow the 3 workers to get done. 344 queue.SpinUntilAllThreadsAreWaiting(); 345 346 { 347 base::AutoLock auto_lock(*queue.lock()); 348 EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments()); 349 EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks()); 350 EXPECT_EQ(0, queue.task_count()); 351 EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread()); 352 EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread()); 353 EXPECT_EQ(3, queue.GetNumberOfCompletedTasks()); 354 355 // Set up to make each task get help from another worker. 356 queue.ResetHistory(); 357 queue.SetTaskCount(20); // 2 tasks per thread. 358 queue.SetWorkTime(kThirtyMs); 359 queue.SetAllowHelp(true); 360 } 361 queue.work_is_available()->Signal(); // But each worker can signal another. 362 // Wait till we at least start to handle tasks (and we're not all waiting). 363 queue.SpinUntilTaskCountLessThan(20); 364 // Wait to allow the 10 workers to get done. 365 queue.SpinUntilAllThreadsAreWaiting(); // Should take about 60 ms. 366 367 { 368 base::AutoLock auto_lock(*queue.lock()); 369 EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments()); 370 EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks()); 371 EXPECT_EQ(0, queue.task_count()); 372 EXPECT_EQ(20, queue.GetNumberOfCompletedTasks()); 373 374 // Same as last test, but with Broadcast(). 375 queue.ResetHistory(); 376 queue.SetTaskCount(20); // 2 tasks per thread. 377 queue.SetWorkTime(kThirtyMs); 378 queue.SetAllowHelp(true); 379 } 380 queue.work_is_available()->Broadcast(); 381 // Wait till we at least start to handle tasks (and we're not all waiting). 382 queue.SpinUntilTaskCountLessThan(20); 383 // Wait to allow the 10 workers to get done. 384 queue.SpinUntilAllThreadsAreWaiting(); // Should take about 60 ms. 385 386 { 387 base::AutoLock auto_lock(*queue.lock()); 388 EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments()); 389 EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks()); 390 EXPECT_EQ(0, queue.task_count()); 391 EXPECT_EQ(20, queue.GetNumberOfCompletedTasks()); 392 393 queue.SetShutdown(); 394 } 395 queue.work_is_available()->Broadcast(); // Force check for shutdown. 396 397 SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(TimeDelta::FromMinutes(1), 398 queue.ThreadSafeCheckShutdown(kThreadCount)); 399 } 400 401 TEST_F(ConditionVariableTest, LargeFastTaskTest) { 402 const int kThreadCount = 200; 403 WorkQueue queue(kThreadCount); // Start the threads. 404 405 Lock private_lock; // Used locally for master to wait. 406 base::AutoLock private_held_lock(private_lock); 407 ConditionVariable private_cv(&private_lock); 408 409 { 410 base::AutoLock auto_lock(*queue.lock()); 411 while (!queue.EveryIdWasAllocated()) 412 queue.all_threads_have_ids()->Wait(); 413 } 414 415 // Wait a bit more to allow threads to reach their wait state. 416 queue.SpinUntilAllThreadsAreWaiting(); 417 418 { 419 // Since we have no tasks, all threads should be waiting by now. 420 base::AutoLock auto_lock(*queue.lock()); 421 EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments()); 422 EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks()); 423 EXPECT_EQ(0, queue.task_count()); 424 EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread()); 425 EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread()); 426 EXPECT_EQ(0, queue.GetNumberOfCompletedTasks()); 427 428 // Set up to make all workers do (an average of) 20 tasks. 429 queue.ResetHistory(); 430 queue.SetTaskCount(20 * kThreadCount); 431 queue.SetWorkTime(kFortyFiveMs); 432 queue.SetAllowHelp(false); 433 } 434 queue.work_is_available()->Broadcast(); // Start up all threads. 435 // Wait until we've handed out all tasks. 436 { 437 base::AutoLock auto_lock(*queue.lock()); 438 while (queue.task_count() != 0) 439 queue.no_more_tasks()->Wait(); 440 } 441 442 // Wait till the last of the tasks complete. 443 queue.SpinUntilAllThreadsAreWaiting(); 444 445 { 446 // With Broadcast(), every thread should have participated. 447 // but with racing.. they may not all have done equal numbers of tasks. 448 base::AutoLock auto_lock(*queue.lock()); 449 EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments()); 450 EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks()); 451 EXPECT_EQ(0, queue.task_count()); 452 EXPECT_LE(20, queue.GetMaxCompletionsByWorkerThread()); 453 EXPECT_EQ(20 * kThreadCount, queue.GetNumberOfCompletedTasks()); 454 455 // Set up to make all workers do (an average of) 4 tasks. 456 queue.ResetHistory(); 457 queue.SetTaskCount(kThreadCount * 4); 458 queue.SetWorkTime(kFortyFiveMs); 459 queue.SetAllowHelp(true); // Might outperform Broadcast(). 460 } 461 queue.work_is_available()->Signal(); // Start up one thread. 462 463 // Wait until we've handed out all tasks 464 { 465 base::AutoLock auto_lock(*queue.lock()); 466 while (queue.task_count() != 0) 467 queue.no_more_tasks()->Wait(); 468 } 469 470 // Wait till the last of the tasks complete. 471 queue.SpinUntilAllThreadsAreWaiting(); 472 473 { 474 // With Signal(), every thread should have participated. 475 // but with racing.. they may not all have done four tasks. 476 base::AutoLock auto_lock(*queue.lock()); 477 EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments()); 478 EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks()); 479 EXPECT_EQ(0, queue.task_count()); 480 EXPECT_LE(4, queue.GetMaxCompletionsByWorkerThread()); 481 EXPECT_EQ(4 * kThreadCount, queue.GetNumberOfCompletedTasks()); 482 483 queue.SetShutdown(); 484 } 485 queue.work_is_available()->Broadcast(); // Force check for shutdown. 486 487 // Wait for shutdowns to complete. 488 SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(TimeDelta::FromMinutes(1), 489 queue.ThreadSafeCheckShutdown(kThreadCount)); 490 } 491 492 //------------------------------------------------------------------------------ 493 // Finally we provide the implementation for the methods in the WorkQueue class. 494 //------------------------------------------------------------------------------ 495 496 WorkQueue::WorkQueue(int thread_count) 497 : lock_(), 498 work_is_available_(&lock_), 499 all_threads_have_ids_(&lock_), 500 no_more_tasks_(&lock_), 501 thread_count_(thread_count), 502 waiting_thread_count_(0), 503 thread_handles_(new PlatformThreadHandle[thread_count]), 504 assignment_history_(thread_count), 505 completion_history_(thread_count), 506 thread_started_counter_(0), 507 shutdown_task_count_(0), 508 task_count_(0), 509 allow_help_requests_(false), 510 shutdown_(false) { 511 EXPECT_GE(thread_count_, 1); 512 ResetHistory(); 513 SetTaskCount(0); 514 SetWorkTime(TimeDelta::FromMilliseconds(30)); 515 516 for (int i = 0; i < thread_count_; ++i) { 517 PlatformThreadHandle pth; 518 EXPECT_TRUE(PlatformThread::Create(0, this, &pth)); 519 thread_handles_[i] = pth; 520 } 521 } 522 523 WorkQueue::~WorkQueue() { 524 { 525 base::AutoLock auto_lock(lock_); 526 SetShutdown(); 527 } 528 work_is_available_.Broadcast(); // Tell them all to terminate. 529 530 for (int i = 0; i < thread_count_; ++i) { 531 PlatformThread::Join(thread_handles_[i]); 532 } 533 EXPECT_EQ(0, waiting_thread_count_); 534 } 535 536 int WorkQueue::GetThreadId() { 537 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 538 DCHECK(!EveryIdWasAllocated()); 539 return thread_started_counter_++; // Give out Unique IDs. 540 } 541 542 bool WorkQueue::EveryIdWasAllocated() const { 543 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 544 return thread_count_ == thread_started_counter_; 545 } 546 547 TimeDelta WorkQueue::GetAnAssignment(int thread_id) { 548 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 549 DCHECK_LT(0, task_count_); 550 assignment_history_[thread_id]++; 551 if (0 == --task_count_) { 552 no_more_tasks_.Signal(); 553 } 554 return worker_delay_; 555 } 556 557 void WorkQueue::WorkIsCompleted(int thread_id) { 558 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 559 completion_history_[thread_id]++; 560 } 561 562 int WorkQueue::task_count() const { 563 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 564 return task_count_; 565 } 566 567 bool WorkQueue::allow_help_requests() const { 568 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 569 return allow_help_requests_; 570 } 571 572 bool WorkQueue::shutdown() const { 573 lock_.AssertAcquired(); 574 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 575 return shutdown_; 576 } 577 578 // Because this method is called from the test's main thread we need to actually 579 // take the lock. Threads will call the thread_shutting_down() method with the 580 // lock already acquired. 581 bool WorkQueue::ThreadSafeCheckShutdown(int thread_count) { 582 bool all_shutdown; 583 base::AutoLock auto_lock(lock_); 584 { 585 // Declare in scope so DFAKE is guranteed to be destroyed before AutoLock. 586 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 587 all_shutdown = (shutdown_task_count_ == thread_count); 588 } 589 return all_shutdown; 590 } 591 592 void WorkQueue::thread_shutting_down() { 593 lock_.AssertAcquired(); 594 DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_); 595 shutdown_task_count_++; 596 } 597 598 Lock* WorkQueue::lock() { 599 return &lock_; 600 } 601 602 ConditionVariable* WorkQueue::work_is_available() { 603 return &work_is_available_; 604 } 605 606 ConditionVariable* WorkQueue::all_threads_have_ids() { 607 return &all_threads_have_ids_; 608 } 609 610 ConditionVariable* WorkQueue::no_more_tasks() { 611 return &no_more_tasks_; 612 } 613 614 void WorkQueue::ResetHistory() { 615 for (int i = 0; i < thread_count_; ++i) { 616 assignment_history_[i] = 0; 617 completion_history_[i] = 0; 618 } 619 } 620 621 int WorkQueue::GetMinCompletionsByWorkerThread() const { 622 int minumum = completion_history_[0]; 623 for (int i = 0; i < thread_count_; ++i) 624 minumum = std::min(minumum, completion_history_[i]); 625 return minumum; 626 } 627 628 int WorkQueue::GetMaxCompletionsByWorkerThread() const { 629 int maximum = completion_history_[0]; 630 for (int i = 0; i < thread_count_; ++i) 631 maximum = std::max(maximum, completion_history_[i]); 632 return maximum; 633 } 634 635 int WorkQueue::GetNumThreadsTakingAssignments() const { 636 int count = 0; 637 for (int i = 0; i < thread_count_; ++i) 638 if (assignment_history_[i]) 639 count++; 640 return count; 641 } 642 643 int WorkQueue::GetNumThreadsCompletingTasks() const { 644 int count = 0; 645 for (int i = 0; i < thread_count_; ++i) 646 if (completion_history_[i]) 647 count++; 648 return count; 649 } 650 651 int WorkQueue::GetNumberOfCompletedTasks() const { 652 int total = 0; 653 for (int i = 0; i < thread_count_; ++i) 654 total += completion_history_[i]; 655 return total; 656 } 657 658 void WorkQueue::SetWorkTime(TimeDelta delay) { 659 worker_delay_ = delay; 660 } 661 662 void WorkQueue::SetTaskCount(int count) { 663 task_count_ = count; 664 } 665 666 void WorkQueue::SetAllowHelp(bool allow) { 667 allow_help_requests_ = allow; 668 } 669 670 void WorkQueue::SetShutdown() { 671 lock_.AssertAcquired(); 672 shutdown_ = true; 673 } 674 675 void WorkQueue::SpinUntilAllThreadsAreWaiting() { 676 while (true) { 677 { 678 base::AutoLock auto_lock(lock_); 679 if (waiting_thread_count_ == thread_count_) 680 break; 681 } 682 PlatformThread::Sleep(TimeDelta::FromMilliseconds(30)); 683 } 684 } 685 686 void WorkQueue::SpinUntilTaskCountLessThan(int task_count) { 687 while (true) { 688 { 689 base::AutoLock auto_lock(lock_); 690 if (task_count_ < task_count) 691 break; 692 } 693 PlatformThread::Sleep(TimeDelta::FromMilliseconds(30)); 694 } 695 } 696 697 698 //------------------------------------------------------------------------------ 699 // Define the standard worker task. Several tests will spin out many of these 700 // threads. 701 //------------------------------------------------------------------------------ 702 703 // The multithread tests involve several threads with a task to perform as 704 // directed by an instance of the class WorkQueue. 705 // The task is to: 706 // a) Check to see if there are more tasks (there is a task counter). 707 // a1) Wait on condition variable if there are no tasks currently. 708 // b) Call a function to see what should be done. 709 // c) Do some computation based on the number of milliseconds returned in (b). 710 // d) go back to (a). 711 712 // WorkQueue::ThreadMain() implements the above task for all threads. 713 // It calls the controlling object to tell the creator about progress, and to 714 // ask about tasks. 715 716 void WorkQueue::ThreadMain() { 717 int thread_id; 718 { 719 base::AutoLock auto_lock(lock_); 720 thread_id = GetThreadId(); 721 if (EveryIdWasAllocated()) 722 all_threads_have_ids()->Signal(); // Tell creator we're ready. 723 } 724 725 Lock private_lock; // Used to waste time on "our work". 726 while (1) { // This is the main consumer loop. 727 TimeDelta work_time; 728 bool could_use_help; 729 { 730 base::AutoLock auto_lock(lock_); 731 while (0 == task_count() && !shutdown()) { 732 ++waiting_thread_count_; 733 work_is_available()->Wait(); 734 --waiting_thread_count_; 735 } 736 if (shutdown()) { 737 // Ack the notification of a shutdown message back to the controller. 738 thread_shutting_down(); 739 return; // Terminate. 740 } 741 // Get our task duration from the queue. 742 work_time = GetAnAssignment(thread_id); 743 could_use_help = (task_count() > 0) && allow_help_requests(); 744 } // Release lock 745 746 // Do work (outside of locked region. 747 if (could_use_help) 748 work_is_available()->Signal(); // Get help from other threads. 749 750 if (work_time > TimeDelta::FromMilliseconds(0)) { 751 // We could just sleep(), but we'll instead further exercise the 752 // condition variable class, and do a timed wait. 753 base::AutoLock auto_lock(private_lock); 754 ConditionVariable private_cv(&private_lock); 755 private_cv.TimedWait(work_time); // Unsynchronized waiting. 756 } 757 758 { 759 base::AutoLock auto_lock(lock_); 760 // Send notification that we completed our "work." 761 WorkIsCompleted(thread_id); 762 } 763 } 764 } 765 766 } // namespace 767 768 } // namespace base 769