// Copyright 2016 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "base/task/thread_pool/worker_thread.h"
#include <stddef.h>
#include <atomic>
#include <memory>
#include <utility>
#include <vector>
#include "base/compiler_specific.h"
#include "base/functional/bind.h"
#include "base/functional/callback_helpers.h"
#include "base/memory/ptr_util.h"
#include "base/memory/raw_ptr.h"
#include "base/memory/ref_counted.h"
#include "base/message_loop/message_pump_type.h"
#include "base/synchronization/condition_variable.h"
#include "base/task/common/checked_lock.h"
#include "base/task/thread_pool/environment_config.h"
#include "base/task/thread_pool/sequence.h"
#include "base/task/thread_pool/task.h"
#include "base/task/thread_pool/task_tracker.h"
#include "base/task/thread_pool/test_utils.h"
#include "base/task/thread_pool/worker_thread_observer.h"
#include "base/test/bind.h"
#include "base/test/test_timeouts.h"
#include "base/test/test_waitable_event.h"
#include "base/threading/platform_thread.h"
#include "base/threading/simple_thread.h"
#include "base/threading/thread.h"
#include "base/time/time.h"
#include "build/build_config.h"
#include "partition_alloc/buildflags.h"
#include "partition_alloc/partition_alloc_config.h"
#include "partition_alloc/shim/allocator_shim.h"
#include "partition_alloc/shim/allocator_shim_default_dispatch_to_partition_alloc.h"
#include "testing/gmock/include/gmock/gmock.h"
#include "testing/gtest/include/gtest/gtest.h"
#if PA_BUILDFLAG(USE_PARTITION_ALLOC_AS_MALLOC) && \
PA_CONFIG(THREAD_CACHE_SUPPORTED)
#include "partition_alloc/extended_api.h"
#include "partition_alloc/thread_cache.h"
#endif // PA_BUILDFLAG(USE_PARTITION_ALLOC_AS_MALLOC) &&
// PA_CONFIG(THREAD_CACHE_SUPPORTED)
using testing::_;
using testing::Mock;
using testing::Ne;
using testing::StrictMock;
namespace base::internal {
namespace {
const size_t kNumSequencesPerTest = 150;
class WorkerThreadDefaultDelegate : public WorkerThread::Delegate {
public:
WorkerThreadDefaultDelegate() = default;
WorkerThreadDefaultDelegate(const WorkerThreadDefaultDelegate&) = delete;
WorkerThreadDefaultDelegate& operator=(const WorkerThreadDefaultDelegate&) =
delete;
// WorkerThread::Delegate:
WorkerThread::ThreadLabel GetThreadLabel() const override {
return WorkerThread::ThreadLabel::DEDICATED;
}
void OnMainEntry(WorkerThread* worker) override {}
RegisteredTaskSource GetWork(WorkerThread* worker) override {
return nullptr;
}
RegisteredTaskSource SwapProcessedTask(RegisteredTaskSource task_source,
WorkerThread* worker) override {
ADD_FAILURE() << "Unexpected call to SwapProcessedTask()";
return nullptr;
}
TimeDelta GetSleepTimeout() override { return TimeDelta::Max(); }
};
// The test parameter is the number of Tasks per Sequence returned by GetWork().
class ThreadPoolWorkerTest : public testing::Test {
public:
ThreadPoolWorkerTest(const ThreadPoolWorkerTest&) = delete;
ThreadPoolWorkerTest& operator=(const ThreadPoolWorkerTest&) = delete;
protected:
ThreadPoolWorkerTest() {
Thread::Options service_thread_options;
service_thread_options.message_pump_type = MessagePumpType::IO;
service_thread_.StartWithOptions(std::move(service_thread_options));
}
Thread service_thread_ = Thread("ServiceThread");
};
// The test parameter is the number of Tasks per Sequence returned by GetWork().
class ThreadPoolWorkerTestParam : public testing::TestWithParam<int> {
public:
ThreadPoolWorkerTestParam(const ThreadPoolWorkerTestParam&) = delete;
ThreadPoolWorkerTestParam& operator=(const ThreadPoolWorkerTestParam&) =
delete;
protected:
ThreadPoolWorkerTestParam()
: num_get_work_cv_(lock_.CreateConditionVariable()) {
Thread::Options service_thread_options;
service_thread_options.message_pump_type = MessagePumpType::IO;
service_thread_.StartWithOptions(std::move(service_thread_options));
}
void SetUp() override {
worker_ = MakeRefCounted<WorkerThread>(
ThreadType::kDefault,
std::make_unique<TestWorkerThreadDelegate>(this),
task_tracker_.GetTrackedRef(), 0);
ASSERT_TRUE(worker_);
worker_->Start(service_thread_.task_runner());
worker_set_.Signal();
main_entry_called_.Wait();
}
void TearDown() override {
// |worker_| needs to be released before ~TaskTracker() as it holds a
// TrackedRef to it.
worker_->JoinForTesting();
worker_ = nullptr;
}
int TasksPerSequence() const { return GetParam(); }
// Wait until GetWork() has been called |num_get_work| times.
void WaitForNumGetWork(size_t num_get_work) {
CheckedAutoLock auto_lock(lock_);
while (num_get_work_ < num_get_work) {
num_get_work_cv_.Wait();
}
}
void SetMaxGetWork(size_t max_get_work) {
CheckedAutoLock auto_lock(lock_);
max_get_work_ = max_get_work;
}
void SetNumSequencesToCreate(size_t num_sequences_to_create) {
CheckedAutoLock auto_lock(lock_);
EXPECT_EQ(0U, num_sequences_to_create_);
num_sequences_to_create_ = num_sequences_to_create;
}
size_t NumRunTasks() {
CheckedAutoLock auto_lock(lock_);
return num_run_tasks_;
}
std::vector<scoped_refptr<TaskSource>> CreatedTaskSources() {
CheckedAutoLock auto_lock(lock_);
return created_sequences_;
}
std::vector<scoped_refptr<TaskSource>> DidProcessTaskSequences() {
CheckedAutoLock auto_lock(lock_);
return did_run_task_sources_;
}
scoped_refptr<WorkerThread> worker_;
Thread service_thread_ = Thread("ServiceThread");
private:
class TestWorkerThreadDelegate
: public WorkerThreadDefaultDelegate {
public:
explicit TestWorkerThreadDelegate(ThreadPoolWorkerTestParam* outer)
: outer_(outer) {}
TestWorkerThreadDelegate(
const TestWorkerThreadDelegate&) = delete;
TestWorkerThreadDelegate& operator=(
const TestWorkerThreadDelegate&) = delete;
~TestWorkerThreadDelegate() override {
EXPECT_FALSE(IsCallToDidProcessTaskExpected());
}
// WorkerThread::Delegate:
void OnMainEntry(WorkerThread* worker) override {
outer_->worker_set_.Wait();
EXPECT_EQ(outer_->worker_.get(), worker);
EXPECT_FALSE(IsCallToDidProcessTaskExpected());
// Without synchronization, OnMainEntry() could be called twice without
// generating an error.
CheckedAutoLock auto_lock(outer_->lock_);
EXPECT_FALSE(outer_->main_entry_called_.IsSignaled());
outer_->main_entry_called_.Signal();
}
RegisteredTaskSource GetWork(WorkerThread* worker) override {
EXPECT_FALSE(IsCallToDidProcessTaskExpected());
EXPECT_EQ(outer_->worker_.get(), worker);
{
CheckedAutoLock auto_lock(outer_->lock_);
// Increment the number of times that this method has been called.
++outer_->num_get_work_;
outer_->num_get_work_cv_.Signal();
// Verify that this method isn't called more times than expected.
EXPECT_LE(outer_->num_get_work_, outer_->max_get_work_);
// Check if a Sequence should be returned.
if (outer_->num_sequences_to_create_ == 0) {
return nullptr;
}
--outer_->num_sequences_to_create_;
}
// Create a Sequence with TasksPerSequence() Tasks.
scoped_refptr<Sequence> sequence = MakeRefCounted<Sequence>(
TaskTraits(), nullptr, TaskSourceExecutionMode::kParallel);
Sequence::Transaction sequence_transaction(sequence->BeginTransaction());
for (int i = 0; i < outer_->TasksPerSequence(); ++i) {
Task task(FROM_HERE,
BindOnce(&ThreadPoolWorkerTestParam::RunTaskCallback,
Unretained(outer_)),
TimeTicks::Now(), TimeDelta());
sequence_transaction.WillPushImmediateTask();
EXPECT_TRUE(outer_->task_tracker_.WillPostTask(
&task, sequence->shutdown_behavior()));
sequence_transaction.PushImmediateTask(std::move(task));
}
auto registered_task_source =
outer_->task_tracker_.RegisterTaskSource(sequence);
EXPECT_TRUE(registered_task_source);
ExpectCallToDidProcessTask();
{
// Add the Sequence to the vector of created Sequences.
CheckedAutoLock auto_lock(outer_->lock_);
outer_->created_sequences_.push_back(sequence);
}
auto run_status = registered_task_source.WillRunTask();
EXPECT_NE(run_status, TaskSource::RunStatus::kDisallowed);
return registered_task_source;
}
RegisteredTaskSource SwapProcessedTask(
RegisteredTaskSource registered_task_source,
WorkerThread* worker) override {
{
CheckedAutoLock auto_lock(expect_did_run_task_lock_);
EXPECT_TRUE(expect_did_run_task_);
expect_did_run_task_ = false;
}
// If TasksPerSequence() is 1, |registered_task_source| should be nullptr.
// Otherwise, |registered_task_source| should contain TasksPerSequence() -
// 1 Tasks.
if (outer_->TasksPerSequence() == 1) {
EXPECT_FALSE(registered_task_source);
} else {
EXPECT_TRUE(registered_task_source);
EXPECT_TRUE(registered_task_source.WillReEnqueue(TimeTicks::Now()));
// Verify the number of Tasks in |registered_task_source|.
for (int i = 0; i < outer_->TasksPerSequence() - 1; ++i) {
registered_task_source.WillRunTask();
IgnoreResult(registered_task_source.TakeTask());
if (i < outer_->TasksPerSequence() - 2) {
EXPECT_TRUE(registered_task_source.DidProcessTask());
EXPECT_TRUE(registered_task_source.WillReEnqueue(TimeTicks::Now()));
} else {
EXPECT_FALSE(registered_task_source.DidProcessTask());
}
}
scoped_refptr<TaskSource> task_source =
registered_task_source.Unregister();
{
// Add |task_source| to |did_run_task_sources_|.
CheckedAutoLock auto_lock(outer_->lock_);
outer_->did_run_task_sources_.push_back(std::move(task_source));
EXPECT_LE(outer_->did_run_task_sources_.size(),
outer_->created_sequences_.size());
}
}
return GetWork(worker);
}
private:
// Expect a call to DidProcessTask() before the next call to any other
// method of this delegate.
void ExpectCallToDidProcessTask() {
CheckedAutoLock auto_lock(expect_did_run_task_lock_);
expect_did_run_task_ = true;
}
bool IsCallToDidProcessTaskExpected() const {
CheckedAutoLock auto_lock(expect_did_run_task_lock_);
return expect_did_run_task_;
}
raw_ptr<ThreadPoolWorkerTestParam> outer_;
// Synchronizes access to |expect_did_run_task_|.
mutable CheckedLock expect_did_run_task_lock_;
// Whether the next method called on this delegate should be
// DidProcessTask().
bool expect_did_run_task_ = false;
};
void RunTaskCallback() {
CheckedAutoLock auto_lock(lock_);
++num_run_tasks_;
EXPECT_LE(num_run_tasks_, created_sequences_.size());
}
TaskTracker task_tracker_;
// Synchronizes access to all members below.
mutable CheckedLock lock_;
// Signaled once OnMainEntry() has been called.
TestWaitableEvent main_entry_called_;
// Number of Sequences that should be created by GetWork(). When this
// is 0, GetWork() returns nullptr.
size_t num_sequences_to_create_ = 0;
// Number of times that GetWork() has been called.
size_t num_get_work_ = 0;
// Maximum number of times that GetWork() can be called.
size_t max_get_work_ = 0;
// Condition variable signaled when |num_get_work_| is incremented.
ConditionVariable num_get_work_cv_;
// Sequences created by GetWork().
std::vector<scoped_refptr<TaskSource>> created_sequences_;
// Sequences passed to DidProcessTask().
std::vector<scoped_refptr<TaskSource>> did_run_task_sources_;
// Number of times that RunTaskCallback() has been called.
size_t num_run_tasks_ = 0;
// Signaled after |worker_| is set.
TestWaitableEvent worker_set_;
};
} // namespace
// Verify that when GetWork() continuously returns Sequences, all Tasks in these
// Sequences run successfully. The test wakes up the WorkerThread once.
TEST_P(ThreadPoolWorkerTestParam, ContinuousWork) {
// Set GetWork() to return |kNumSequencesPerTest| Sequences before starting to
// return nullptr.
SetNumSequencesToCreate(kNumSequencesPerTest);
// Expect |kNumSequencesPerTest| calls to GetWork() in which it returns a
// Sequence and one call in which its returns nullptr.
const size_t kExpectedNumGetWork = kNumSequencesPerTest + 1;
SetMaxGetWork(kExpectedNumGetWork);
// Wake up |worker_| and wait until GetWork() has been invoked the
// expected amount of times.
worker_->WakeUp();
WaitForNumGetWork(kExpectedNumGetWork);
// All tasks should have run.
EXPECT_EQ(kNumSequencesPerTest, NumRunTasks());
// If Sequences returned by GetWork() contain more than one Task, they aren't
// empty after the worker pops Tasks from them and thus should be returned to
// DidProcessTask().
if (TasksPerSequence() > 1) {
EXPECT_EQ(CreatedTaskSources(), DidProcessTaskSequences());
} else {
EXPECT_TRUE(DidProcessTaskSequences().empty());
}
}
// Verify that when GetWork() alternates between returning a Sequence and
// returning nullptr, all Tasks in the returned Sequences run successfully. The
// test wakes up the WorkerThread once for each Sequence.
TEST_P(ThreadPoolWorkerTestParam, IntermittentWork) {
for (size_t i = 0; i < kNumSequencesPerTest; ++i) {
// Set GetWork() to return 1 Sequence before starting to return
// nullptr.
SetNumSequencesToCreate(1);
// Expect |i + 1| calls to GetWork() in which it returns a Sequence and
// |i + 1| calls in which it returns nullptr.
const size_t expected_num_get_work = 2 * (i + 1);
SetMaxGetWork(expected_num_get_work);
// Wake up |worker_| and wait until GetWork() has been invoked
// the expected amount of times.
worker_->WakeUp();
WaitForNumGetWork(expected_num_get_work);
// The Task should have run
EXPECT_EQ(i + 1, NumRunTasks());
// If Sequences returned by GetWork() contain more than one Task, they
// aren't empty after the worker pops Tasks from them and thus should be
// returned to DidProcessTask().
if (TasksPerSequence() > 1) {
EXPECT_EQ(CreatedTaskSources(), DidProcessTaskSequences());
} else {
EXPECT_TRUE(DidProcessTaskSequences().empty());
}
}
}
INSTANTIATE_TEST_SUITE_P(OneTaskPerSequence,
ThreadPoolWorkerTestParam,
::testing::Values(1));
INSTANTIATE_TEST_SUITE_P(TwoTasksPerSequence,
ThreadPoolWorkerTestParam,
::testing::Values(2));
namespace {
class ControllableCleanupDelegate : public WorkerThreadDefaultDelegate {
public:
class Controls : public RefCountedThreadSafe<Controls> {
public:
Controls() = default;
Controls(const Controls&) = delete;
Controls& operator=(const Controls&) = delete;
void HaveWorkBlock() { work_running_.Reset(); }
void UnblockWork() { work_running_.Signal(); }
void WaitForWorkToRun() { work_processed_.Wait(); }
void WaitForCleanupRequest() { cleanup_requested_.Wait(); }
void WaitForDelegateDestroy() { destroyed_.Wait(); }
void WaitForMainExit() { exited_.Wait(); }
void set_expect_get_work(bool expect_get_work) {
expect_get_work_ = expect_get_work;
}
void ResetState() {
work_running_.Signal();
work_processed_.Reset();
cleanup_requested_.Reset();
exited_.Reset();
work_requested_ = false;
}
void set_can_cleanup(bool can_cleanup) { can_cleanup_ = can_cleanup; }
private:
friend class ControllableCleanupDelegate;
friend class RefCountedThreadSafe<Controls>;
~Controls() = default;
TestWaitableEvent work_running_{WaitableEvent::ResetPolicy::MANUAL,
WaitableEvent::InitialState::SIGNALED};
TestWaitableEvent work_processed_;
TestWaitableEvent cleanup_requested_;
TestWaitableEvent destroyed_;
TestWaitableEvent exited_;
bool expect_get_work_ = true;
bool can_cleanup_ = false;
bool work_requested_ = false;
};
explicit ControllableCleanupDelegate(TaskTracker* task_tracker)
: task_tracker_(task_tracker), controls_(new Controls()) {}
ControllableCleanupDelegate(const ControllableCleanupDelegate&) = delete;
ControllableCleanupDelegate& operator=(const ControllableCleanupDelegate&) =
delete;
~ControllableCleanupDelegate() override { controls_->destroyed_.Signal(); }
RegisteredTaskSource GetWork(WorkerThread* worker) override {
EXPECT_TRUE(controls_->expect_get_work_);
// Sends one item of work to signal |work_processed_|. On subsequent calls,
// sends nullptr to indicate there's no more work to be done.
if (controls_->work_requested_) {
if (CanCleanup(worker)) {
OnCleanup();
worker->Cleanup();
controls_->set_expect_get_work(false);
}
return nullptr;
}
controls_->work_requested_ = true;
scoped_refptr<Sequence> sequence = MakeRefCounted<Sequence>(
TaskTraits(WithBaseSyncPrimitives(),
TaskShutdownBehavior::CONTINUE_ON_SHUTDOWN),
nullptr, TaskSourceExecutionMode::kParallel);
Task task(FROM_HERE,
BindOnce(
[](TestWaitableEvent* work_processed,
TestWaitableEvent* work_running) {
work_processed->Signal();
work_running->Wait();
},
Unretained(&controls_->work_processed_),
Unretained(&controls_->work_running_)),
TimeTicks::Now(), TimeDelta());
auto transaction = sequence->BeginTransaction();
transaction.WillPushImmediateTask();
EXPECT_TRUE(
task_tracker_->WillPostTask(&task, sequence->shutdown_behavior()));
transaction.PushImmediateTask(std::move(task));
auto registered_task_source =
task_tracker_->RegisterTaskSource(std::move(sequence));
EXPECT_TRUE(registered_task_source);
registered_task_source.WillRunTask();
return registered_task_source;
}
RegisteredTaskSource SwapProcessedTask(RegisteredTaskSource task_source,
WorkerThread* worker) override {
return GetWork(worker);
}
void OnMainExit(WorkerThread* worker) override {
controls_->exited_.Signal();
}
bool CanCleanup(WorkerThread* worker) {
// Saving |can_cleanup_| now so that callers waiting on |cleanup_requested_|
// have the thread go to sleep and then allow timing out.
bool can_cleanup = controls_->can_cleanup_;
controls_->cleanup_requested_.Signal();
return can_cleanup;
}
void OnCleanup() {
EXPECT_TRUE(controls_->can_cleanup_);
EXPECT_TRUE(controls_->cleanup_requested_.IsSignaled());
}
// ControllableCleanupDelegate:
scoped_refptr<Controls> controls() { return controls_; }
private:
scoped_refptr<Sequence> work_sequence_;
const raw_ptr<TaskTracker> task_tracker_;
scoped_refptr<Controls> controls_;
};
class MockedControllableCleanupDelegate : public ControllableCleanupDelegate {
public:
explicit MockedControllableCleanupDelegate(TaskTracker* task_tracker)
: ControllableCleanupDelegate(task_tracker) {}
MockedControllableCleanupDelegate(const MockedControllableCleanupDelegate&) =
delete;
MockedControllableCleanupDelegate& operator=(
const MockedControllableCleanupDelegate&) = delete;
~MockedControllableCleanupDelegate() override = default;
// WorkerThread::Delegate:
MOCK_METHOD1(OnMainEntry, void(WorkerThread* worker));
};
} // namespace
// Verify that calling WorkerThread::Cleanup() from GetWork() causes
// the WorkerThread's thread to exit.
TEST_F(ThreadPoolWorkerTest, WorkerCleanupFromGetWork) {
TaskTracker task_tracker;
// Will be owned by WorkerThread.
MockedControllableCleanupDelegate* delegate =
new StrictMock<MockedControllableCleanupDelegate>(&task_tracker);
scoped_refptr<ControllableCleanupDelegate::Controls> controls =
delegate->controls();
controls->set_can_cleanup(true);
EXPECT_CALL(*delegate, OnMainEntry(_));
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, WrapUnique(delegate), task_tracker.GetTrackedRef(),
0);
worker->Start(service_thread_.task_runner());
worker->WakeUp();
controls->WaitForWorkToRun();
Mock::VerifyAndClear(delegate);
controls->WaitForMainExit();
// Join the worker to avoid leaks.
worker->JoinForTesting();
}
TEST_F(ThreadPoolWorkerTest, WorkerCleanupDuringWork) {
TaskTracker task_tracker;
// Will be owned by WorkerThread.
// No mock here as that's reasonably covered by other tests and the delegate
// may destroy on a different thread. Mocks aren't designed with that in mind.
std::unique_ptr<ControllableCleanupDelegate> delegate =
std::make_unique<ControllableCleanupDelegate>(&task_tracker);
scoped_refptr<ControllableCleanupDelegate::Controls> controls =
delegate->controls();
controls->HaveWorkBlock();
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
worker->Start(service_thread_.task_runner());
worker->WakeUp();
controls->WaitForWorkToRun();
worker->Cleanup();
worker = nullptr;
controls->UnblockWork();
controls->WaitForDelegateDestroy();
}
TEST_F(ThreadPoolWorkerTest, WorkerCleanupDuringWait) {
TaskTracker task_tracker;
// Will be owned by WorkerThread.
// No mock here as that's reasonably covered by other tests and the delegate
// may destroy on a different thread. Mocks aren't designed with that in mind.
std::unique_ptr<ControllableCleanupDelegate> delegate =
std::make_unique<ControllableCleanupDelegate>(&task_tracker);
scoped_refptr<ControllableCleanupDelegate::Controls> controls =
delegate->controls();
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
worker->Start(service_thread_.task_runner());
worker->WakeUp();
controls->WaitForCleanupRequest();
worker->Cleanup();
worker = nullptr;
controls->WaitForDelegateDestroy();
}
TEST_F(ThreadPoolWorkerTest, WorkerCleanupDuringShutdown) {
TaskTracker task_tracker;
// Will be owned by WorkerThread.
// No mock here as that's reasonably covered by other tests and the delegate
// may destroy on a different thread. Mocks aren't designed with that in mind.
std::unique_ptr<ControllableCleanupDelegate> delegate =
std::make_unique<ControllableCleanupDelegate>(&task_tracker);
scoped_refptr<ControllableCleanupDelegate::Controls> controls =
delegate->controls();
controls->HaveWorkBlock();
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
worker->Start(service_thread_.task_runner());
worker->WakeUp();
controls->WaitForWorkToRun();
test::ShutdownTaskTracker(&task_tracker);
worker->Cleanup();
worker = nullptr;
controls->UnblockWork();
controls->WaitForDelegateDestroy();
}
// Verify that Start() is a no-op after Cleanup().
TEST_F(ThreadPoolWorkerTest, CleanupBeforeStart) {
TaskTracker task_tracker;
// Will be owned by WorkerThread.
// No mock here as that's reasonably covered by other tests and the delegate
// may destroy on a different thread. Mocks aren't designed with that in mind.
std::unique_ptr<ControllableCleanupDelegate> delegate =
std::make_unique<ControllableCleanupDelegate>(&task_tracker);
scoped_refptr<ControllableCleanupDelegate::Controls> controls =
delegate->controls();
controls->set_expect_get_work(false);
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
worker->Cleanup();
worker->Start(service_thread_.task_runner());
EXPECT_FALSE(worker->ThreadAliveForTesting());
}
namespace {
class CallJoinFromDifferentThread : public SimpleThread {
public:
explicit CallJoinFromDifferentThread(
WorkerThread* worker_to_join)
: SimpleThread("WorkerThreadJoinThread"),
worker_to_join_(worker_to_join) {}
CallJoinFromDifferentThread(const CallJoinFromDifferentThread&) = delete;
CallJoinFromDifferentThread& operator=(const CallJoinFromDifferentThread&) =
delete;
~CallJoinFromDifferentThread() override = default;
void Run() override {
run_started_event_.Signal();
worker_to_join_.ExtractAsDangling()->JoinForTesting();
}
void WaitForRunToStart() { run_started_event_.Wait(); }
private:
raw_ptr<WorkerThread> worker_to_join_;
TestWaitableEvent run_started_event_;
};
} // namespace
TEST_F(ThreadPoolWorkerTest, WorkerCleanupDuringJoin) {
TaskTracker task_tracker;
// Will be owned by WorkerThread.
// No mock here as that's reasonably covered by other tests and the
// delegate may destroy on a different thread. Mocks aren't designed with that
// in mind.
std::unique_ptr<ControllableCleanupDelegate> delegate =
std::make_unique<ControllableCleanupDelegate>(&task_tracker);
scoped_refptr<ControllableCleanupDelegate::Controls> controls =
delegate->controls();
controls->HaveWorkBlock();
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
worker->Start(service_thread_.task_runner());
worker->WakeUp();
controls->WaitForWorkToRun();
CallJoinFromDifferentThread join_from_different_thread(worker.get());
join_from_different_thread.Start();
join_from_different_thread.WaitForRunToStart();
// Sleep here to give the other thread a chance to call JoinForTesting().
// Receiving a signal that Run() was called doesn't mean JoinForTesting() was
// necessarily called, and we can't signal after JoinForTesting() as
// JoinForTesting() blocks until we call UnblockWork().
PlatformThread::Sleep(TestTimeouts::tiny_timeout());
worker->Cleanup();
worker = nullptr;
controls->UnblockWork();
controls->WaitForDelegateDestroy();
join_from_different_thread.Join();
}
namespace {
class ExpectThreadTypeDelegate : public WorkerThreadDefaultDelegate {
public:
ExpectThreadTypeDelegate()
: thread_type_verified_in_get_work_event_(
WaitableEvent::ResetPolicy::AUTOMATIC) {}
ExpectThreadTypeDelegate(const ExpectThreadTypeDelegate&) = delete;
ExpectThreadTypeDelegate& operator=(const ExpectThreadTypeDelegate&) = delete;
void SetExpectedThreadType(ThreadType expected_thread_type) {
expected_thread_type_ = expected_thread_type;
}
void WaitForThreadTypeVerifiedInGetWork() {
thread_type_verified_in_get_work_event_.Wait();
}
// WorkerThread::Delegate:
void OnMainEntry(WorkerThread* worker) override { VerifyThreadType(); }
RegisteredTaskSource GetWork(WorkerThread* worker) override {
VerifyThreadType();
thread_type_verified_in_get_work_event_.Signal();
return nullptr;
}
private:
void VerifyThreadType() {
CheckedAutoLock auto_lock(expected_thread_type_lock_);
EXPECT_EQ(expected_thread_type_, PlatformThread::GetCurrentThreadType());
}
// Signaled after GetWork() has verified the thread type of the worker thread.
TestWaitableEvent thread_type_verified_in_get_work_event_;
// Synchronizes access to |expected_thread_type_|.
CheckedLock expected_thread_type_lock_;
// Expected thread type for the next call to OnMainEntry() or GetWork().
ThreadType expected_thread_type_ = ThreadType::kDefault;
};
} // namespace
TEST_F(ThreadPoolWorkerTest, BumpThreadTypeOfAliveThreadDuringShutdown) {
if (!CanUseBackgroundThreadTypeForWorkerThread()) {
return;
}
TaskTracker task_tracker;
// Block shutdown to ensure that the worker doesn't exit when StartShutdown()
// is called.
scoped_refptr<Sequence> sequence =
MakeRefCounted<Sequence>(TaskTraits{TaskShutdownBehavior::BLOCK_SHUTDOWN},
nullptr, TaskSourceExecutionMode::kParallel);
auto registered_task_source =
task_tracker.RegisterTaskSource(std::move(sequence));
std::unique_ptr<ExpectThreadTypeDelegate> delegate(
new ExpectThreadTypeDelegate);
ExpectThreadTypeDelegate* delegate_raw = delegate.get();
delegate_raw->SetExpectedThreadType(ThreadType::kBackground);
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kBackground, std::move(delegate),
task_tracker.GetTrackedRef(), 0);
worker->Start(service_thread_.task_runner());
// Verify that the initial thread type is kBackground (or kNormal if thread
// type can't be increased).
worker->WakeUp();
delegate_raw->WaitForThreadTypeVerifiedInGetWork();
// Verify that the thread type is bumped to kNormal during shutdown.
delegate_raw->SetExpectedThreadType(ThreadType::kDefault);
task_tracker.StartShutdown();
worker->WakeUp();
delegate_raw->WaitForThreadTypeVerifiedInGetWork();
worker->JoinForTesting();
}
namespace {
class VerifyCallsToObserverDelegate : public WorkerThreadDefaultDelegate {
public:
explicit VerifyCallsToObserverDelegate(
test::MockWorkerThreadObserver* observer)
: observer_(observer) {}
VerifyCallsToObserverDelegate(const VerifyCallsToObserverDelegate&) = delete;
VerifyCallsToObserverDelegate& operator=(
const VerifyCallsToObserverDelegate&) = delete;
// WorkerThread::Delegate:
void OnMainEntry(WorkerThread* worker) override {
Mock::VerifyAndClear(observer_);
}
void OnMainExit(WorkerThread* worker) override {
observer_->AllowCallsOnMainExit(1);
}
private:
const raw_ptr<test::MockWorkerThreadObserver> observer_;
};
} // namespace
// Verify that the WorkerThreadObserver is notified when the worker enters
// and exits its main function.
TEST_F(ThreadPoolWorkerTest, WorkerThreadObserver) {
StrictMock<test::MockWorkerThreadObserver> observer;
TaskTracker task_tracker;
auto delegate = std::make_unique<VerifyCallsToObserverDelegate>(&observer);
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
EXPECT_CALL(observer, OnWorkerThreadMainEntry());
worker->Start(service_thread_.task_runner(), &observer);
worker->Cleanup();
// Join the worker to avoid leaks.
worker->JoinForTesting();
Mock::VerifyAndClear(&observer);
}
#if PA_BUILDFLAG(USE_PARTITION_ALLOC_AS_MALLOC) && \
PA_CONFIG(THREAD_CACHE_SUPPORTED)
namespace {
NOINLINE void FreeForTest(void* data) {
free(data);
}
using ::testing::AllOf;
using ::testing::Ge;
using ::testing::Le;
} // namespace
class WorkerThreadThreadCacheDelegate : public WorkerThreadDefaultDelegate {
public:
void PrepareForTesting() {
TimeTicks now = TimeTicks::Now();
WorkerThreadDefaultDelegate::set_first_sleep_time_for_testing(now);
first_sleep_time_for_testing_ = now;
}
TimeDelta GetSleepDurationBeforePurge(TimeTicks) override {
// Expect `GetSleepDurationBeforePurge()` to return
// `kFirstSleepDurationBeforePurge` when invoked within
// `kFirstSleepDurationBeforePurge` of the first sleep (with
// `kPurgeThreadCacheIdleDelay` tolerance due to alignment).
EXPECT_THAT(
WorkerThreadDefaultDelegate::GetSleepDurationBeforePurge(
/* now=*/first_sleep_time_for_testing_),
AllOf(Ge(WorkerThread::Delegate::kFirstSleepDurationBeforePurge),
Le(WorkerThread::Delegate::kFirstSleepDurationBeforePurge +
WorkerThread::Delegate::kPurgeThreadCacheIdleDelay)));
// Expect `GetSleepDurationBeforePurge()` to return
// `WorkerThread::Delegate::kPurgeThreadCacheIdleDelay` when invoked later
// than `kFirstSleepDurationBeforePurge` after the first sleep (with
// `kPurgeThreadCacheIdleDelay` tolerance due to alignment).
constexpr base::TimeDelta kEpsilon = base::Microseconds(1);
EXPECT_THAT(
WorkerThreadDefaultDelegate::GetSleepDurationBeforePurge(
/* now=*/first_sleep_time_for_testing_ +
WorkerThread::Delegate::kFirstSleepDurationBeforePurge + kEpsilon),
AllOf(Ge(WorkerThread::Delegate::kPurgeThreadCacheIdleDelay),
Le(WorkerThread::Delegate::kPurgeThreadCacheIdleDelay +
WorkerThread::Delegate::kPurgeThreadCacheIdleDelay)));
// The output of this function is used to drive real sleep in tests.
// Return a constant so that the tests can validate the wakeups without
// timing out.
return GetSleepTimeout();
}
void WaitForWork() override {
// Fill several buckets before going to sleep.
for (size_t size = 8;
size < partition_alloc::ThreadCache::kDefaultSizeThreshold; size++) {
void* data = malloc(size);
// A simple malloc() / free() pair can be discarded by the compiler (and
// is), making the test fail. It is sufficient to make |FreeForTest()| a
// NOINLINE function for the call to not be eliminated, but it is
// required.
FreeForTest(data);
}
size_t cached_memory_before =
partition_alloc::ThreadCache::Get()->CachedMemory();
WorkerThreadDefaultDelegate::WaitForWork();
size_t cached_memory_after =
partition_alloc::ThreadCache::Get()->CachedMemory();
if (!test_done_) {
if (purge_expected_) {
EXPECT_LT(cached_memory_after, cached_memory_before / 2);
} else {
EXPECT_GT(cached_memory_after, cached_memory_before / 2);
}
// Unblock the test.
wakeup_done_.Signal();
test_done_ = true;
}
}
// Avoid using the default sleep timeout which is infinite and prevents the
// tests from completing.
TimeDelta GetSleepTimeout() override {
return WorkerThread::Delegate::kPurgeThreadCacheIdleDelay +
TestTimeouts::tiny_timeout();
}
void SetPurgeExpectation(bool purge_expected) {
purge_expected_ = purge_expected;
}
TimeTicks first_sleep_time_for_testing_;
bool test_done_ = false;
bool purge_expected_ = false;
base::WaitableEvent wakeup_done_;
};
TEST_F(ThreadPoolWorkerTest, WorkerThreadCacheNoPurgeOnSignal) {
// Make sure the thread cache is enabled in the main partition.
partition_alloc::internal::ThreadCacheProcessScopeForTesting scope(
allocator_shim::internal::PartitionAllocMalloc::Allocator());
TaskTracker task_tracker;
auto delegate = std::make_unique<WorkerThreadThreadCacheDelegate>();
auto* delegate_raw = delegate.get();
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
delegate_raw->PrepareForTesting();
// No purge is expected on waking up from a signal.
delegate_raw->SetPurgeExpectation(false);
// Wake up before the thread is started to make sure the first wakeup is
// caused by a signal.
worker->WakeUp();
worker->Start(service_thread_.task_runner(), nullptr);
// Wait until a wakeup has completed.
delegate_raw->wakeup_done_.Wait();
worker->JoinForTesting();
}
TEST_F(ThreadPoolWorkerTest, PurgeOnUninteruptedSleep) {
// Make sure the thread cache is enabled in the main partition.
partition_alloc::internal::ThreadCacheProcessScopeForTesting scope(
allocator_shim::internal::PartitionAllocMalloc::Allocator());
TaskTracker task_tracker;
auto delegate = std::make_unique<WorkerThreadThreadCacheDelegate>();
auto* delegate_raw = delegate.get();
auto worker = MakeRefCounted<WorkerThread>(
ThreadType::kDefault, std::move(delegate), task_tracker.GetTrackedRef(),
0);
delegate_raw->PrepareForTesting();
// A purge will take place
delegate_raw->SetPurgeExpectation(true);
worker->Start(service_thread_.task_runner(), nullptr);
// Wait until a wakeup has completed.
delegate_raw->wakeup_done_.Wait();
worker->JoinForTesting();
}
#endif // PA_BUILDFLAG(USE_PARTITION_ALLOC_AS_MALLOC) &&
// PA_CONFIG(THREAD_CACHE_SUPPORTED)
} // namespace base::internal
Codebase Browser
chromium