condition_variable_unittest.cc 25 KB

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