blob: b4e8b58520c6e40d5a3a227d449f04511833ba85 [file] [log] [blame]
// 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.
#include "base/threading/worker_pool_posix.h"
#include <set>
#include "base/bind.h"
#include "base/callback.h"
#include "base/macros.h"
#include "base/synchronization/condition_variable.h"
#include "base/synchronization/lock.h"
#include "base/synchronization/waitable_event.h"
#include "base/threading/platform_thread.h"
#include "testing/gtest/include/gtest/gtest.h"
namespace base {
// Peer class to provide passthrough access to PosixDynamicThreadPool internals.
class PosixDynamicThreadPool::PosixDynamicThreadPoolPeer {
public:
explicit PosixDynamicThreadPoolPeer(PosixDynamicThreadPool* pool)
: pool_(pool) {}
Lock* lock() { return &pool_->lock_; }
ConditionVariable* pending_tasks_available_cv() {
return &pool_->pending_tasks_available_cv_;
}
const std::queue<PendingTask>& pending_tasks() const {
return pool_->pending_tasks_;
}
int num_idle_threads() const { return pool_->num_idle_threads_; }
ConditionVariable* num_idle_threads_cv() {
return pool_->num_idle_threads_cv_.get();
}
void set_num_idle_threads_cv(ConditionVariable* cv) {
pool_->num_idle_threads_cv_.reset(cv);
}
private:
PosixDynamicThreadPool* pool_;
DISALLOW_COPY_AND_ASSIGN(PosixDynamicThreadPoolPeer);
};
namespace {
// IncrementingTask's main purpose is to increment a counter. It also updates a
// set of unique thread ids, and signals a ConditionVariable on completion.
// Note that since it does not block, there is no way to control the number of
// threads used if more than one IncrementingTask is consecutively posted to the
// thread pool, since the first one might finish executing before the subsequent
// PostTask() calls get invoked.
void IncrementingTask(Lock* counter_lock,
int* counter,
Lock* unique_threads_lock,
std::set<PlatformThreadId>* unique_threads) {
{
base::AutoLock locked(*unique_threads_lock);
unique_threads->insert(PlatformThread::CurrentId());
}
base::AutoLock locked(*counter_lock);
(*counter)++;
}
// BlockingIncrementingTask is a simple wrapper around IncrementingTask that
// allows for waiting at the start of Run() for a WaitableEvent to be signalled.
struct BlockingIncrementingTaskArgs {
Lock* counter_lock;
int* counter;
Lock* unique_threads_lock;
std::set<PlatformThreadId>* unique_threads;
Lock* num_waiting_to_start_lock;
int* num_waiting_to_start;
ConditionVariable* num_waiting_to_start_cv;
base::WaitableEvent* start;
};
void BlockingIncrementingTask(const BlockingIncrementingTaskArgs& args) {
{
base::AutoLock num_waiting_to_start_locked(*args.num_waiting_to_start_lock);
(*args.num_waiting_to_start)++;
}
args.num_waiting_to_start_cv->Signal();
args.start->Wait();
IncrementingTask(args.counter_lock, args.counter, args.unique_threads_lock,
args.unique_threads);
}
class PosixDynamicThreadPoolTest : public testing::Test {
protected:
PosixDynamicThreadPoolTest()
: pool_(new base::PosixDynamicThreadPool("dynamic_pool", 60 * 60)),
peer_(pool_.get()),
counter_(0),
num_waiting_to_start_(0),
num_waiting_to_start_cv_(&num_waiting_to_start_lock_),
start_(WaitableEvent::ResetPolicy::MANUAL,
WaitableEvent::InitialState::NOT_SIGNALED) {}
void SetUp() override {
peer_.set_num_idle_threads_cv(new ConditionVariable(peer_.lock()));
}
void WaitForTasksToStart(int num_tasks) {
base::AutoLock num_waiting_to_start_locked(num_waiting_to_start_lock_);
while (num_waiting_to_start_ < num_tasks) {
num_waiting_to_start_cv_.Wait();
}
}
void WaitForIdleThreads(int num_idle_threads) {
base::AutoLock pool_locked(*peer_.lock());
while (peer_.num_idle_threads() < num_idle_threads) {
peer_.num_idle_threads_cv()->Wait();
}
}
base::Closure CreateNewIncrementingTaskCallback() {
return base::Bind(&IncrementingTask, &counter_lock_, &counter_,
&unique_threads_lock_, &unique_threads_);
}
base::Closure CreateNewBlockingIncrementingTaskCallback() {
BlockingIncrementingTaskArgs args = {
&counter_lock_, &counter_, &unique_threads_lock_, &unique_threads_,
&num_waiting_to_start_lock_, &num_waiting_to_start_,
&num_waiting_to_start_cv_, &start_
};
return base::Bind(&BlockingIncrementingTask, args);
}
scoped_refptr<base::PosixDynamicThreadPool> pool_;
base::PosixDynamicThreadPool::PosixDynamicThreadPoolPeer peer_;
Lock counter_lock_;
int counter_;
Lock unique_threads_lock_;
std::set<PlatformThreadId> unique_threads_;
Lock num_waiting_to_start_lock_;
int num_waiting_to_start_;
ConditionVariable num_waiting_to_start_cv_;
base::WaitableEvent start_;
};
} // namespace
TEST_F(PosixDynamicThreadPoolTest, Basic) {
EXPECT_EQ(0, peer_.num_idle_threads());
EXPECT_EQ(0U, unique_threads_.size());
EXPECT_EQ(0U, peer_.pending_tasks().size());
// Add one task and wait for it to be completed.
pool_->PostTask(FROM_HERE, CreateNewIncrementingTaskCallback());
WaitForIdleThreads(1);
EXPECT_EQ(1U, unique_threads_.size()) <<
"There should be only one thread allocated for one task.";
EXPECT_EQ(1, counter_);
}
TEST_F(PosixDynamicThreadPoolTest, ReuseIdle) {
// Add one task and wait for it to be completed.
pool_->PostTask(FROM_HERE, CreateNewIncrementingTaskCallback());
WaitForIdleThreads(1);
// Add another 2 tasks. One should reuse the existing worker thread.
pool_->PostTask(FROM_HERE, CreateNewBlockingIncrementingTaskCallback());
pool_->PostTask(FROM_HERE, CreateNewBlockingIncrementingTaskCallback());
WaitForTasksToStart(2);
start_.Signal();
WaitForIdleThreads(2);
EXPECT_EQ(2U, unique_threads_.size());
EXPECT_EQ(2, peer_.num_idle_threads());
EXPECT_EQ(3, counter_);
}
TEST_F(PosixDynamicThreadPoolTest, TwoActiveTasks) {
// Add two blocking tasks.
pool_->PostTask(FROM_HERE, CreateNewBlockingIncrementingTaskCallback());
pool_->PostTask(FROM_HERE, CreateNewBlockingIncrementingTaskCallback());
EXPECT_EQ(0, counter_) << "Blocking tasks should not have started yet.";
WaitForTasksToStart(2);
start_.Signal();
WaitForIdleThreads(2);
EXPECT_EQ(2U, unique_threads_.size());
EXPECT_EQ(2, peer_.num_idle_threads()) << "Existing threads are now idle.";
EXPECT_EQ(2, counter_);
}
TEST_F(PosixDynamicThreadPoolTest, Complex) {
// Add two non blocking tasks and wait for them to finish.
pool_->PostTask(FROM_HERE, CreateNewIncrementingTaskCallback());
WaitForIdleThreads(1);
// Add two blocking tasks, start them simultaneously, and wait for them to
// finish.
pool_->PostTask(FROM_HERE, CreateNewBlockingIncrementingTaskCallback());
pool_->PostTask(FROM_HERE, CreateNewBlockingIncrementingTaskCallback());
WaitForTasksToStart(2);
start_.Signal();
WaitForIdleThreads(2);
EXPECT_EQ(3, counter_);
EXPECT_EQ(2, peer_.num_idle_threads());
EXPECT_EQ(2U, unique_threads_.size());
// Wake up all idle threads so they can exit.
{
base::AutoLock locked(*peer_.lock());
while (peer_.num_idle_threads() > 0) {
peer_.pending_tasks_available_cv()->Signal();
peer_.num_idle_threads_cv()->Wait();
}
}
// Add another non blocking task. There are no threads to reuse.
pool_->PostTask(FROM_HERE, CreateNewIncrementingTaskCallback());
WaitForIdleThreads(1);
// The POSIX implementation of PlatformThread::CurrentId() uses pthread_self()
// which is not guaranteed to be unique after a thread joins. The OS X
// implemntation of pthread_self() returns the address of the pthread_t, which
// is merely a malloc()ed pointer stored in the first TLS slot. When a thread
// joins and that structure is freed, the block of memory can be put on the
// OS free list, meaning the same address could be reused in a subsequent
// allocation. This in fact happens when allocating in a loop as this test
// does.
//
// Because there are two concurrent threads, there's at least the guarantee
// of having two unique thread IDs in the set. But after those two threads are
// joined, the next-created thread can get a re-used ID if the allocation of
// the pthread_t structure is taken from the free list. Therefore, there can
// be either 2 or 3 unique thread IDs in the set at this stage in the test.
EXPECT_TRUE(unique_threads_.size() >= 2 && unique_threads_.size() <= 3)
<< "unique_threads_.size() = " << unique_threads_.size();
EXPECT_EQ(1, peer_.num_idle_threads());
EXPECT_EQ(4, counter_);
}
} // namespace base