123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759 |
- // Copyright (c) 2012 The Chromium Authors. All rights reserved.
- // Use of this source code is governed by a BSD-style license that can be
- // found in the LICENSE file.
- // Multi-threaded tests of ConditionVariable class.
- #include "base/synchronization/condition_variable.h"
- #include <time.h>
- #include <algorithm>
- #include <memory>
- #include <vector>
- #include "base/bind.h"
- #include "base/location.h"
- #include "base/logging.h"
- #include "base/synchronization/lock.h"
- #include "base/task/single_thread_task_runner.h"
- #include "base/test/spin_wait.h"
- #include "base/threading/platform_thread.h"
- #include "base/threading/thread.h"
- #include "base/threading/thread_collision_warner.h"
- #include "base/time/time.h"
- #include "build/build_config.h"
- #include "testing/gtest/include/gtest/gtest.h"
- #include "testing/platform_test.h"
- namespace base {
- namespace {
- //------------------------------------------------------------------------------
- // Define our test class, with several common variables.
- //------------------------------------------------------------------------------
- class ConditionVariableTest : public PlatformTest {
- public:
- const TimeDelta kZeroMs;
- const TimeDelta kTenMs;
- const TimeDelta kThirtyMs;
- const TimeDelta kFortyFiveMs;
- const TimeDelta kSixtyMs;
- const TimeDelta kOneHundredMs;
- ConditionVariableTest()
- : kZeroMs(Milliseconds(0)),
- kTenMs(Milliseconds(10)),
- kThirtyMs(Milliseconds(30)),
- kFortyFiveMs(Milliseconds(45)),
- kSixtyMs(Milliseconds(60)),
- kOneHundredMs(Milliseconds(100)) {}
- };
- //------------------------------------------------------------------------------
- // Define a class that will control activities an several multi-threaded tests.
- // The general structure of multi-threaded tests is that a test case will
- // construct an instance of a WorkQueue. The WorkQueue will spin up some
- // threads and control them throughout their lifetime, as well as maintaining
- // a central repository of the work thread's activity. Finally, the WorkQueue
- // will command the worker threads to terminate. At that point, the test
- // cases will validate that the WorkQueue has records showing that the desired
- // activities were performed.
- //------------------------------------------------------------------------------
- // Callers are responsible for synchronizing access to the following class.
- // The WorkQueue::lock_, as accessed via WorkQueue::lock(), should be used for
- // all synchronized access.
- class WorkQueue : public PlatformThread::Delegate {
- public:
- explicit WorkQueue(int thread_count);
- ~WorkQueue() override;
- // PlatformThread::Delegate interface.
- void ThreadMain() override;
- //----------------------------------------------------------------------------
- // Worker threads only call the following methods.
- // They should use the lock to get exclusive access.
- int GetThreadId(); // Get an ID assigned to a thread..
- bool EveryIdWasAllocated() const; // Indicates that all IDs were handed out.
- TimeDelta GetAnAssignment(int thread_id); // Get a work task duration.
- void WorkIsCompleted(int thread_id);
- int task_count() const;
- bool allow_help_requests() const; // Workers can signal more workers.
- bool shutdown() const; // Check if shutdown has been requested.
- void thread_shutting_down();
- //----------------------------------------------------------------------------
- // Worker threads can call them but not needed to acquire a lock.
- Lock* lock();
- ConditionVariable* work_is_available();
- ConditionVariable* all_threads_have_ids();
- ConditionVariable* no_more_tasks();
- //----------------------------------------------------------------------------
- // The rest of the methods are for use by the controlling master thread (the
- // test case code).
- void ResetHistory();
- int GetMinCompletionsByWorkerThread() const;
- int GetMaxCompletionsByWorkerThread() const;
- int GetNumThreadsTakingAssignments() const;
- int GetNumThreadsCompletingTasks() const;
- int GetNumberOfCompletedTasks() const;
- void SetWorkTime(TimeDelta delay);
- void SetTaskCount(int count);
- void SetAllowHelp(bool allow);
- // The following must be called without locking, and will spin wait until the
- // threads are all in a wait state.
- void SpinUntilAllThreadsAreWaiting();
- void SpinUntilTaskCountLessThan(int task_count);
- // Caller must acquire lock before calling.
- void SetShutdown();
- // Compares the |shutdown_task_count_| to the |thread_count| and returns true
- // if they are equal. This check will acquire the |lock_| so the caller
- // should not hold the lock when calling this method.
- bool ThreadSafeCheckShutdown(int thread_count);
- private:
- // Both worker threads and controller use the following to synchronize.
- Lock lock_;
- ConditionVariable work_is_available_; // To tell threads there is work.
- // Conditions to notify the controlling process (if it is interested).
- ConditionVariable all_threads_have_ids_; // All threads are running.
- ConditionVariable no_more_tasks_; // Task count is zero.
- const int thread_count_;
- int waiting_thread_count_;
- std::unique_ptr<PlatformThreadHandle[]> thread_handles_;
- std::vector<int> assignment_history_; // Number of assignment per worker.
- std::vector<int> completion_history_; // Number of completions per worker.
- int thread_started_counter_; // Used to issue unique id to workers.
- int shutdown_task_count_; // Number of tasks told to shutdown
- int task_count_; // Number of assignment tasks waiting to be processed.
- TimeDelta worker_delay_; // Time each task takes to complete.
- bool allow_help_requests_; // Workers can signal more workers.
- bool shutdown_; // Set when threads need to terminate.
- DFAKE_MUTEX(locked_methods_);
- };
- //------------------------------------------------------------------------------
- // The next section contains the actual tests.
- //------------------------------------------------------------------------------
- TEST_F(ConditionVariableTest, StartupShutdownTest) {
- Lock lock;
- // First try trivial startup/shutdown.
- {
- ConditionVariable cv1(&lock);
- } // Call for cv1 destruction.
- // Exercise with at least a few waits.
- ConditionVariable cv(&lock);
- lock.Acquire();
- cv.TimedWait(kTenMs); // Wait for 10 ms.
- cv.TimedWait(kTenMs); // Wait for 10 ms.
- lock.Release();
- lock.Acquire();
- cv.TimedWait(kTenMs); // Wait for 10 ms.
- cv.TimedWait(kTenMs); // Wait for 10 ms.
- cv.TimedWait(kTenMs); // Wait for 10 ms.
- lock.Release();
- } // Call for cv destruction.
- TEST_F(ConditionVariableTest, TimeoutTest) {
- Lock lock;
- ConditionVariable cv(&lock);
- lock.Acquire();
- TimeTicks start = TimeTicks::Now();
- const TimeDelta WAIT_TIME = Milliseconds(300);
- // Allow for clocking rate granularity.
- const TimeDelta FUDGE_TIME = Milliseconds(50);
- cv.TimedWait(WAIT_TIME + FUDGE_TIME);
- TimeDelta duration = TimeTicks::Now() - start;
- // We can't use EXPECT_GE here as the TimeDelta class does not support the
- // required stream conversion.
- EXPECT_TRUE(duration >= WAIT_TIME);
- lock.Release();
- }
- #if BUILDFLAG(IS_POSIX)
- const int kDiscontinuitySeconds = 2;
- void BackInTime(Lock* lock) {
- AutoLock auto_lock(*lock);
- timeval tv;
- gettimeofday(&tv, nullptr);
- tv.tv_sec -= kDiscontinuitySeconds;
- settimeofday(&tv, nullptr);
- }
- // Tests that TimedWait ignores changes to the system clock.
- // Test is disabled by default, because it needs to run as root to muck with the
- // system clock.
- // http://crbug.com/293736
- TEST_F(ConditionVariableTest, DISABLED_TimeoutAcrossSetTimeOfDay) {
- timeval tv;
- gettimeofday(&tv, nullptr);
- tv.tv_sec += kDiscontinuitySeconds;
- if (settimeofday(&tv, nullptr) < 0) {
- PLOG(ERROR) << "Could not set time of day. Run as root?";
- return;
- }
- Lock lock;
- ConditionVariable cv(&lock);
- lock.Acquire();
- Thread thread("Helper");
- thread.Start();
- thread.task_runner()->PostTask(FROM_HERE, base::BindOnce(&BackInTime, &lock));
- TimeTicks start = TimeTicks::Now();
- const TimeDelta kWaitTime = Milliseconds(300);
- // Allow for clocking rate granularity.
- const TimeDelta kFudgeTime = Milliseconds(50);
- cv.TimedWait(kWaitTime + kFudgeTime);
- TimeDelta duration = TimeTicks::Now() - start;
- thread.Stop();
- // We can't use EXPECT_GE here as the TimeDelta class does not support the
- // required stream conversion.
- EXPECT_TRUE(duration >= kWaitTime);
- EXPECT_TRUE(duration <= Seconds(kDiscontinuitySeconds));
- lock.Release();
- }
- #endif
- // Test serial task servicing, as well as two parallel task servicing methods.
- TEST_F(ConditionVariableTest, MultiThreadConsumerTest) {
- const int kThreadCount = 10;
- WorkQueue queue(kThreadCount); // Start the threads.
- const int kTaskCount = 10; // Number of tasks in each mini-test here.
- Time start_time; // Used to time task processing.
- {
- base::AutoLock auto_lock(*queue.lock());
- while (!queue.EveryIdWasAllocated())
- queue.all_threads_have_ids()->Wait();
- }
- // If threads aren't in a wait state, they may start to gobble up tasks in
- // parallel, short-circuiting (breaking) this test.
- queue.SpinUntilAllThreadsAreWaiting();
- {
- // Since we have no tasks yet, all threads should be waiting by now.
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
- EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
- EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());
- // Set up to make each task include getting help from another worker, so
- // so that the work gets done in paralell.
- queue.ResetHistory();
- queue.SetTaskCount(kTaskCount);
- queue.SetWorkTime(kThirtyMs);
- queue.SetAllowHelp(true);
- start_time = Time::Now();
- }
- queue.work_is_available()->Signal(); // But each worker can signal another.
- // Wait till we at least start to handle tasks (and we're not all waiting).
- queue.SpinUntilTaskCountLessThan(kTaskCount);
- // Wait to allow the all workers to get done.
- queue.SpinUntilAllThreadsAreWaiting();
- {
- // Wait until all work tasks have at least been assigned.
- base::AutoLock auto_lock(*queue.lock());
- while (queue.task_count())
- queue.no_more_tasks()->Wait();
- // To avoid racy assumptions, we'll just assert that at least 2 threads
- // did work. We know that the first worker should have gone to sleep, and
- // hence a second worker should have gotten an assignment.
- EXPECT_LE(2, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(kTaskCount, queue.GetNumberOfCompletedTasks());
- // Try to ask all workers to help, and only a few will do the work.
- queue.ResetHistory();
- queue.SetTaskCount(3);
- queue.SetWorkTime(kThirtyMs);
- queue.SetAllowHelp(false);
- }
- queue.work_is_available()->Broadcast(); // Make them all try.
- // Wait till we at least start to handle tasks (and we're not all waiting).
- queue.SpinUntilTaskCountLessThan(3);
- // Wait to allow the 3 workers to get done.
- queue.SpinUntilAllThreadsAreWaiting();
- {
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
- EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
- EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
- // Set up to make each task get help from another worker.
- queue.ResetHistory();
- queue.SetTaskCount(3);
- queue.SetWorkTime(kThirtyMs);
- queue.SetAllowHelp(true); // Allow (unnecessary) help requests.
- }
- queue.work_is_available()->Broadcast(); // Signal all threads.
- // Wait till we at least start to handle tasks (and we're not all waiting).
- queue.SpinUntilTaskCountLessThan(3);
- // Wait to allow the 3 workers to get done.
- queue.SpinUntilAllThreadsAreWaiting();
- {
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
- EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
- EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
- // Set up to make each task get help from another worker.
- queue.ResetHistory();
- queue.SetTaskCount(20); // 2 tasks per thread.
- queue.SetWorkTime(kThirtyMs);
- queue.SetAllowHelp(true);
- }
- queue.work_is_available()->Signal(); // But each worker can signal another.
- // Wait till we at least start to handle tasks (and we're not all waiting).
- queue.SpinUntilTaskCountLessThan(20);
- // Wait to allow the 10 workers to get done.
- queue.SpinUntilAllThreadsAreWaiting(); // Should take about 60 ms.
- {
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
- // Same as last test, but with Broadcast().
- queue.ResetHistory();
- queue.SetTaskCount(20); // 2 tasks per thread.
- queue.SetWorkTime(kThirtyMs);
- queue.SetAllowHelp(true);
- }
- queue.work_is_available()->Broadcast();
- // Wait till we at least start to handle tasks (and we're not all waiting).
- queue.SpinUntilTaskCountLessThan(20);
- // Wait to allow the 10 workers to get done.
- queue.SpinUntilAllThreadsAreWaiting(); // Should take about 60 ms.
- {
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
- queue.SetShutdown();
- }
- queue.work_is_available()->Broadcast(); // Force check for shutdown.
- SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(Minutes(1),
- queue.ThreadSafeCheckShutdown(kThreadCount));
- }
- TEST_F(ConditionVariableTest, LargeFastTaskTest) {
- const int kThreadCount = 200;
- WorkQueue queue(kThreadCount); // Start the threads.
- Lock private_lock; // Used locally for master to wait.
- base::AutoLock private_held_lock(private_lock);
- ConditionVariable private_cv(&private_lock);
- {
- base::AutoLock auto_lock(*queue.lock());
- while (!queue.EveryIdWasAllocated())
- queue.all_threads_have_ids()->Wait();
- }
- // Wait a bit more to allow threads to reach their wait state.
- queue.SpinUntilAllThreadsAreWaiting();
- {
- // Since we have no tasks, all threads should be waiting by now.
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
- EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
- EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());
- // Set up to make all workers do (an average of) 20 tasks.
- queue.ResetHistory();
- queue.SetTaskCount(20 * kThreadCount);
- queue.SetWorkTime(kFortyFiveMs);
- queue.SetAllowHelp(false);
- }
- queue.work_is_available()->Broadcast(); // Start up all threads.
- // Wait until we've handed out all tasks.
- {
- base::AutoLock auto_lock(*queue.lock());
- while (queue.task_count() != 0)
- queue.no_more_tasks()->Wait();
- }
- // Wait till the last of the tasks complete.
- queue.SpinUntilAllThreadsAreWaiting();
- {
- // With Broadcast(), every thread should have participated.
- // but with racing.. they may not all have done equal numbers of tasks.
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_LE(20, queue.GetMaxCompletionsByWorkerThread());
- EXPECT_EQ(20 * kThreadCount, queue.GetNumberOfCompletedTasks());
- // Set up to make all workers do (an average of) 4 tasks.
- queue.ResetHistory();
- queue.SetTaskCount(kThreadCount * 4);
- queue.SetWorkTime(kFortyFiveMs);
- queue.SetAllowHelp(true); // Might outperform Broadcast().
- }
- queue.work_is_available()->Signal(); // Start up one thread.
- // Wait until we've handed out all tasks
- {
- base::AutoLock auto_lock(*queue.lock());
- while (queue.task_count() != 0)
- queue.no_more_tasks()->Wait();
- }
- // Wait till the last of the tasks complete.
- queue.SpinUntilAllThreadsAreWaiting();
- {
- // With Signal(), every thread should have participated.
- // but with racing.. they may not all have done four tasks.
- base::AutoLock auto_lock(*queue.lock());
- EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
- EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
- EXPECT_EQ(0, queue.task_count());
- EXPECT_LE(4, queue.GetMaxCompletionsByWorkerThread());
- EXPECT_EQ(4 * kThreadCount, queue.GetNumberOfCompletedTasks());
- queue.SetShutdown();
- }
- queue.work_is_available()->Broadcast(); // Force check for shutdown.
- // Wait for shutdowns to complete.
- SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(Minutes(1),
- queue.ThreadSafeCheckShutdown(kThreadCount));
- }
- //------------------------------------------------------------------------------
- // Finally we provide the implementation for the methods in the WorkQueue class.
- //------------------------------------------------------------------------------
- WorkQueue::WorkQueue(int thread_count)
- : lock_(),
- work_is_available_(&lock_),
- all_threads_have_ids_(&lock_),
- no_more_tasks_(&lock_),
- thread_count_(thread_count),
- waiting_thread_count_(0),
- thread_handles_(new PlatformThreadHandle[thread_count]),
- assignment_history_(thread_count),
- completion_history_(thread_count),
- thread_started_counter_(0),
- shutdown_task_count_(0),
- task_count_(0),
- allow_help_requests_(false),
- shutdown_(false) {
- EXPECT_GE(thread_count_, 1);
- ResetHistory();
- SetTaskCount(0);
- SetWorkTime(Milliseconds(30));
- for (int i = 0; i < thread_count_; ++i) {
- PlatformThreadHandle pth;
- EXPECT_TRUE(PlatformThread::Create(0, this, &pth));
- thread_handles_[i] = pth;
- }
- }
- WorkQueue::~WorkQueue() {
- {
- base::AutoLock auto_lock(lock_);
- SetShutdown();
- }
- work_is_available_.Broadcast(); // Tell them all to terminate.
- for (int i = 0; i < thread_count_; ++i) {
- PlatformThread::Join(thread_handles_[i]);
- }
- EXPECT_EQ(0, waiting_thread_count_);
- }
- int WorkQueue::GetThreadId() {
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- DCHECK(!EveryIdWasAllocated());
- return thread_started_counter_++; // Give out Unique IDs.
- }
- bool WorkQueue::EveryIdWasAllocated() const {
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- return thread_count_ == thread_started_counter_;
- }
- TimeDelta WorkQueue::GetAnAssignment(int thread_id) {
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- DCHECK_LT(0, task_count_);
- assignment_history_[thread_id]++;
- if (0 == --task_count_) {
- no_more_tasks_.Signal();
- }
- return worker_delay_;
- }
- void WorkQueue::WorkIsCompleted(int thread_id) {
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- completion_history_[thread_id]++;
- }
- int WorkQueue::task_count() const {
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- return task_count_;
- }
- bool WorkQueue::allow_help_requests() const {
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- return allow_help_requests_;
- }
- bool WorkQueue::shutdown() const {
- lock_.AssertAcquired();
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- return shutdown_;
- }
- // Because this method is called from the test's main thread we need to actually
- // take the lock. Threads will call the thread_shutting_down() method with the
- // lock already acquired.
- bool WorkQueue::ThreadSafeCheckShutdown(int thread_count) {
- bool all_shutdown;
- base::AutoLock auto_lock(lock_);
- {
- // Declare in scope so DFAKE is guranteed to be destroyed before AutoLock.
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- all_shutdown = (shutdown_task_count_ == thread_count);
- }
- return all_shutdown;
- }
- void WorkQueue::thread_shutting_down() {
- lock_.AssertAcquired();
- DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
- shutdown_task_count_++;
- }
- Lock* WorkQueue::lock() {
- return &lock_;
- }
- ConditionVariable* WorkQueue::work_is_available() {
- return &work_is_available_;
- }
- ConditionVariable* WorkQueue::all_threads_have_ids() {
- return &all_threads_have_ids_;
- }
- ConditionVariable* WorkQueue::no_more_tasks() {
- return &no_more_tasks_;
- }
- void WorkQueue::ResetHistory() {
- for (int i = 0; i < thread_count_; ++i) {
- assignment_history_[i] = 0;
- completion_history_[i] = 0;
- }
- }
- int WorkQueue::GetMinCompletionsByWorkerThread() const {
- int minumum = completion_history_[0];
- for (int i = 0; i < thread_count_; ++i)
- minumum = std::min(minumum, completion_history_[i]);
- return minumum;
- }
- int WorkQueue::GetMaxCompletionsByWorkerThread() const {
- int maximum = completion_history_[0];
- for (int i = 0; i < thread_count_; ++i)
- maximum = std::max(maximum, completion_history_[i]);
- return maximum;
- }
- int WorkQueue::GetNumThreadsTakingAssignments() const {
- int count = 0;
- for (int i = 0; i < thread_count_; ++i)
- if (assignment_history_[i])
- count++;
- return count;
- }
- int WorkQueue::GetNumThreadsCompletingTasks() const {
- int count = 0;
- for (int i = 0; i < thread_count_; ++i)
- if (completion_history_[i])
- count++;
- return count;
- }
- int WorkQueue::GetNumberOfCompletedTasks() const {
- int total = 0;
- for (int i = 0; i < thread_count_; ++i)
- total += completion_history_[i];
- return total;
- }
- void WorkQueue::SetWorkTime(TimeDelta delay) {
- worker_delay_ = delay;
- }
- void WorkQueue::SetTaskCount(int count) {
- task_count_ = count;
- }
- void WorkQueue::SetAllowHelp(bool allow) {
- allow_help_requests_ = allow;
- }
- void WorkQueue::SetShutdown() {
- lock_.AssertAcquired();
- shutdown_ = true;
- }
- void WorkQueue::SpinUntilAllThreadsAreWaiting() {
- while (true) {
- {
- base::AutoLock auto_lock(lock_);
- if (waiting_thread_count_ == thread_count_)
- break;
- }
- PlatformThread::Sleep(Milliseconds(30));
- }
- }
- void WorkQueue::SpinUntilTaskCountLessThan(int task_count) {
- while (true) {
- {
- base::AutoLock auto_lock(lock_);
- if (task_count_ < task_count)
- break;
- }
- PlatformThread::Sleep(Milliseconds(30));
- }
- }
- //------------------------------------------------------------------------------
- // Define the standard worker task. Several tests will spin out many of these
- // threads.
- //------------------------------------------------------------------------------
- // The multithread tests involve several threads with a task to perform as
- // directed by an instance of the class WorkQueue.
- // The task is to:
- // a) Check to see if there are more tasks (there is a task counter).
- // a1) Wait on condition variable if there are no tasks currently.
- // b) Call a function to see what should be done.
- // c) Do some computation based on the number of milliseconds returned in (b).
- // d) go back to (a).
- // WorkQueue::ThreadMain() implements the above task for all threads.
- // It calls the controlling object to tell the creator about progress, and to
- // ask about tasks.
- void WorkQueue::ThreadMain() {
- int thread_id;
- {
- base::AutoLock auto_lock(lock_);
- thread_id = GetThreadId();
- if (EveryIdWasAllocated())
- all_threads_have_ids()->Signal(); // Tell creator we're ready.
- }
- Lock private_lock; // Used to waste time on "our work".
- while (true) { // This is the main consumer loop.
- TimeDelta work_time;
- bool could_use_help;
- {
- base::AutoLock auto_lock(lock_);
- while (0 == task_count() && !shutdown()) {
- ++waiting_thread_count_;
- work_is_available()->Wait();
- --waiting_thread_count_;
- }
- if (shutdown()) {
- // Ack the notification of a shutdown message back to the controller.
- thread_shutting_down();
- return; // Terminate.
- }
- // Get our task duration from the queue.
- work_time = GetAnAssignment(thread_id);
- could_use_help = (task_count() > 0) && allow_help_requests();
- } // Release lock
- // Do work (outside of locked region.
- if (could_use_help)
- work_is_available()->Signal(); // Get help from other threads.
- if (work_time > Milliseconds(0)) {
- // We could just sleep(), but we'll instead further exercise the
- // condition variable class, and do a timed wait.
- base::AutoLock auto_lock(private_lock);
- ConditionVariable private_cv(&private_lock);
- private_cv.TimedWait(work_time); // Unsynchronized waiting.
- }
- {
- base::AutoLock auto_lock(lock_);
- // Send notification that we completed our "work."
- WorkIsCompleted(thread_id);
- }
- }
- }
- } // namespace
- } // namespace base
|