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