chromium/src
0
Fork 0
Code Activity

Remove base::WorkerPool.

Browse Source 
This class is deprecated and no longer used.

TBR=agl@chromium.org,holte@chromium.org

Bug: 659191, 251774
Change-Id: I111c334d3feac302abd14aa9d4d142aa237a00b4
Reviewed-on: https://chromium-review.googlesource.com/650368
Commit-Queue: François Doray <fdoray@chromium.org>
Reviewed-by: Matt Menke <mmenke@chromium.org>
Reviewed-by: Gabriel Charette <gab@chromium.org>
Cr-Commit-Position: refs/heads/master@{#504708}
This commit is contained in:
Francois Doray
2017-09-27 16:59:40 +00:00
committed by Commit Bot
parent 4d50fe026a
commit f3d1fac07f
13 changed files with 11 additions and 913 deletions

7
base/BUILD.gn

@ -896,11 +896,6 @@ component("base") {
"threading/thread_task_runner_handle.h", "threading/thread_task_runner_handle.h",
"threading/watchdog.cc", "threading/watchdog.cc",
"threading/watchdog.h", "threading/watchdog.h",
"threading/worker_pool.cc",
"threading/worker_pool.h",
"threading/worker_pool_posix.cc",
"threading/worker_pool_posix.h",
"threading/worker_pool_win.cc",
"time/clock.cc", "time/clock.cc",
"time/clock.h", "time/clock.h",
"time/default_clock.cc", "time/default_clock.cc",
@ -2195,8 +2190,6 @@ test("base_unittests") {
"threading/thread_task_runner_handle_unittest.cc", "threading/thread_task_runner_handle_unittest.cc",
"threading/thread_unittest.cc", "threading/thread_unittest.cc",
"threading/watchdog_unittest.cc", "threading/watchdog_unittest.cc",
"threading/worker_pool_posix_unittest.cc",
"threading/worker_pool_unittest.cc",
"time/pr_time_unittest.cc", "time/pr_time_unittest.cc",
"time/time_unittest.cc", "time/time_unittest.cc",
"time/time_win_unittest.cc", "time/time_win_unittest.cc",

7
base/threading/post_task_and_reply_impl.h

@ -2,8 +2,7 @@
// Use of this source code is governed by a BSD-style license that can be // Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. // found in the LICENSE file.
// This file contains the implementation shared by // This file contains the implementation for TaskRunner::PostTaskAndReply.
// TaskRunner::PostTaskAndReply and WorkerPool::PostTaskAndReply.
#ifndef BASE_THREADING_POST_TASK_AND_REPLY_IMPL_H_ #ifndef BASE_THREADING_POST_TASK_AND_REPLY_IMPL_H_
#define BASE_THREADING_POST_TASK_AND_REPLY_IMPL_H_ #define BASE_THREADING_POST_TASK_AND_REPLY_IMPL_H_
@ -19,7 +18,9 @@ namespace internal {
// custom execution context. // custom execution context.
// //
// If you're looking for a concrete implementation of PostTaskAndReply, you // If you're looking for a concrete implementation of PostTaskAndReply, you
// probably want base::TaskRunner, or you may want base::WorkerPool. // probably want base::TaskRunner.
//
// TODO(fdoray): Move this to the anonymous namespace of base/task_runner.cc.
class BASE_EXPORT PostTaskAndReplyImpl { class BASE_EXPORT PostTaskAndReplyImpl {
public: public:
virtual ~PostTaskAndReplyImpl() = default; virtual ~PostTaskAndReplyImpl() = default;

5
base/threading/sequenced_worker_pool.h

@ -65,11 +65,6 @@ template <class T> class DeleteHelper;
// for CONTINUE_ON_SHUTDOWN behavior and is required for BLOCK_SHUTDOWN // for CONTINUE_ON_SHUTDOWN behavior and is required for BLOCK_SHUTDOWN
// behavior. // behavior.
// //
// Implementation note: This does not use a base::WorkerPool since that does
// not enforce shutdown semantics or allow us to specify how many worker
// threads to run. For the typical use case of random background work, we don't
// necessarily want to be super aggressive about creating threads.
//
// Note that SequencedWorkerPool is RefCountedThreadSafe (inherited // Note that SequencedWorkerPool is RefCountedThreadSafe (inherited
// from TaskRunner). // from TaskRunner).
// //

121
base/threading/worker_pool.cc

@ -1,121 +0,0 @@
// 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.h"
#include <utility>
#include "base/bind.h"
#include "base/compiler_specific.h"
#include "base/debug/leak_annotations.h"
#include "base/macros.h"
#include "base/task_runner.h"
#include "base/threading/post_task_and_reply_impl.h"
namespace base {
namespace {
class PostTaskAndReplyWorkerPool : public internal::PostTaskAndReplyImpl {
public:
explicit PostTaskAndReplyWorkerPool(bool task_is_slow)
: task_is_slow_(task_is_slow) {
}
~PostTaskAndReplyWorkerPool() override = default;
private:
bool PostTask(const Location& from_here, OnceClosure task) override {
return WorkerPool::PostTask(from_here, std::move(task), task_is_slow_);
}
bool task_is_slow_;
};
// WorkerPoolTaskRunner ---------------------------------------------
// A TaskRunner which posts tasks to a WorkerPool with a
// fixed ShutdownBehavior.
//
// Note that this class is RefCountedThreadSafe (inherited from TaskRunner).
class WorkerPoolTaskRunner : public TaskRunner {
public:
explicit WorkerPoolTaskRunner(bool tasks_are_slow);
// TaskRunner implementation
bool PostDelayedTask(const Location& from_here,
OnceClosure task,
TimeDelta delay) override;
bool RunsTasksInCurrentSequence() const override;
private:
~WorkerPoolTaskRunner() override;
// Helper function for posting a delayed task. Asserts that the delay is
// zero because non-zero delays are not supported.
bool PostDelayedTaskAssertZeroDelay(const Location& from_here,
OnceClosure task,
base::TimeDelta delay);
const bool tasks_are_slow_;
DISALLOW_COPY_AND_ASSIGN(WorkerPoolTaskRunner);
};
WorkerPoolTaskRunner::WorkerPoolTaskRunner(bool tasks_are_slow)
: tasks_are_slow_(tasks_are_slow) {
}
WorkerPoolTaskRunner::~WorkerPoolTaskRunner() {
}
bool WorkerPoolTaskRunner::PostDelayedTask(const Location& from_here,
OnceClosure task,
TimeDelta delay) {
return PostDelayedTaskAssertZeroDelay(from_here, std::move(task), delay);
}
bool WorkerPoolTaskRunner::RunsTasksInCurrentSequence() const {
return WorkerPool::RunsTasksOnCurrentThread();
}
bool WorkerPoolTaskRunner::PostDelayedTaskAssertZeroDelay(
const Location& from_here,
OnceClosure task,
base::TimeDelta delay) {
DCHECK_EQ(delay.InMillisecondsRoundedUp(), 0)
<< "WorkerPoolTaskRunner does not support non-zero delays";
return WorkerPool::PostTask(from_here, std::move(task), tasks_are_slow_);
}
struct TaskRunnerHolder {
TaskRunnerHolder() {
taskrunners_[0] = new WorkerPoolTaskRunner(false);
taskrunners_[1] = new WorkerPoolTaskRunner(true);
}
scoped_refptr<TaskRunner> taskrunners_[2];
};
} // namespace
bool WorkerPool::PostTaskAndReply(const Location& from_here,
OnceClosure task,
OnceClosure reply,
bool task_is_slow) {
// Do not report PostTaskAndReplyRelay leaks in tests. There's nothing we can
// do about them because WorkerPool doesn't have a flushing API.
// http://crbug.com/248513
// http://crbug.com/290897
// Note: this annotation does not cover tasks posted through a TaskRunner.
ANNOTATE_SCOPED_MEMORY_LEAK;
return PostTaskAndReplyWorkerPool(task_is_slow)
.PostTaskAndReply(from_here, std::move(task), std::move(reply));
}
// static
const scoped_refptr<TaskRunner>&
WorkerPool::GetTaskRunner(bool tasks_are_slow) {
static auto* task_runner_holder = new TaskRunnerHolder();
return task_runner_holder->taskrunners_[tasks_are_slow];
}
} // namespace base

56
base/threading/worker_pool.h

@ -1,56 +0,0 @@
// 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.
#ifndef BASE_THREADING_WORKER_POOL_H_
#define BASE_THREADING_WORKER_POOL_H_
#include "base/base_export.h"
#include "base/callback.h"
#include "base/memory/ref_counted.h"
namespace base {
class Location;
class TaskRunner;
// This is a facility that runs tasks that don't require a specific thread or
// a message loop.
//
// WARNING: This shouldn't be used unless absolutely necessary. We don't wait
// for the worker pool threads to finish on shutdown, so the tasks running
// inside the pool must be extremely careful about other objects they access
// (MessageLoops, Singletons, etc). During shutdown these object may no longer
// exist.
class BASE_EXPORT WorkerPool {
public:
// This function posts |task| to run on a worker thread. |task_is_slow|
// should be used for tasks that will take a long time to execute. Returns
// false if |task| could not be posted to a worker thread. Regardless of
// return value, ownership of |task| is transferred to the worker pool.
static bool PostTask(const Location& from_here,
OnceClosure task,
bool task_is_slow);
// Just like TaskRunner::PostTaskAndReply, except the destination
// for |task| is a worker thread and you can specify |task_is_slow| just
// like you can for PostTask above.
static bool PostTaskAndReply(const Location& from_here,
OnceClosure task,
OnceClosure reply,
bool task_is_slow);
// Return true if the current thread is one that this WorkerPool runs tasks
// on. (Note that if the Windows worker pool is used without going through
// this WorkerPool interface, RunsTasksOnCurrentThread would return false on
// those threads.)
static bool RunsTasksOnCurrentThread();
// Get a TaskRunner wrapper which posts to the WorkerPool using the given
// |task_is_slow| behavior.
static const scoped_refptr<TaskRunner>& GetTaskRunner(bool task_is_slow);
};
} // namespace base
#endif // BASE_THREADING_WORKER_POOL_H_

183
base/threading/worker_pool_posix.cc

@ -1,183 +0,0 @@
// 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 <stddef.h>
#include <utility>
#include "base/bind.h"
#include "base/callback.h"
#include "base/lazy_instance.h"
#include "base/logging.h"
#include "base/macros.h"
#include "base/memory/ref_counted.h"
#include "base/strings/stringprintf.h"
#include "base/threading/platform_thread.h"
#include "base/threading/thread_local.h"
#include "base/threading/worker_pool.h"
#include "base/trace_event/trace_event.h"
namespace base {
namespace {
base::LazyInstance<ThreadLocalBoolean>::Leaky
g_worker_pool_running_on_this_thread = LAZY_INSTANCE_INITIALIZER;
const int kIdleSecondsBeforeExit = 10 * 60;
#if defined(OS_MACOSX)
// On Mac OS X a background thread's default stack size is 512Kb. We need at
// least 1MB for compilation tasks in V8, so increase this default.
const int kStackSize = 1 * 1024 * 1024;
#else
const int kStackSize = 0;
#endif
class WorkerPoolImpl {
public:
WorkerPoolImpl();
// WorkerPoolImpl is only instantiated as a leaky LazyInstance, so the
// destructor is never called.
~WorkerPoolImpl() = delete;
void PostTask(const Location& from_here,
base::OnceClosure task,
bool task_is_slow);
private:
scoped_refptr<base::PosixDynamicThreadPool> pool_;
};
WorkerPoolImpl::WorkerPoolImpl()
: pool_(new base::PosixDynamicThreadPool("WorkerPool",
kIdleSecondsBeforeExit)) {}
void WorkerPoolImpl::PostTask(const Location& from_here,
base::OnceClosure task,
bool task_is_slow) {
pool_->PostTask(from_here, std::move(task));
}
base::LazyInstance<WorkerPoolImpl>::Leaky g_lazy_worker_pool =
LAZY_INSTANCE_INITIALIZER;
class WorkerThread : public PlatformThread::Delegate {
public:
WorkerThread(const std::string& name_prefix,
base::PosixDynamicThreadPool* pool)
: name_prefix_(name_prefix), pool_(pool) {}
void ThreadMain() override;
private:
const std::string name_prefix_;
scoped_refptr<base::PosixDynamicThreadPool> pool_;
DISALLOW_COPY_AND_ASSIGN(WorkerThread);
};
void WorkerThread::ThreadMain() {
g_worker_pool_running_on_this_thread.Get().Set(true);
const std::string name = base::StringPrintf("%s/%d", name_prefix_.c_str(),
PlatformThread::CurrentId());
// Note |name.c_str()| must remain valid for for the whole life of the thread.
PlatformThread::SetName(name);
for (;;) {
PendingTask pending_task = pool_->WaitForTask();
if (pending_task.task.is_null())
break;
TRACE_TASK_EXECUTION("WorkerThread::ThreadMain::Run", pending_task);
std::move(pending_task.task).Run();
}
// The WorkerThread is non-joinable, so it deletes itself.
delete this;
}
} // namespace
// static
bool WorkerPool::PostTask(const Location& from_here,
base::OnceClosure task,
bool task_is_slow) {
g_lazy_worker_pool.Pointer()->PostTask(from_here, std::move(task),
task_is_slow);
return true;
}
// static
bool WorkerPool::RunsTasksOnCurrentThread() {
return g_worker_pool_running_on_this_thread.Get().Get();
}
PosixDynamicThreadPool::PosixDynamicThreadPool(const std::string& name_prefix,
int idle_seconds_before_exit)
: name_prefix_(name_prefix),
idle_seconds_before_exit_(idle_seconds_before_exit),
pending_tasks_available_cv_(&lock_),
num_idle_threads_(0) {}
PosixDynamicThreadPool::~PosixDynamicThreadPool() {
while (!pending_tasks_.empty())
pending_tasks_.pop();
}
void PosixDynamicThreadPool::PostTask(const Location& from_here,
base::OnceClosure task) {
PendingTask pending_task(from_here, std::move(task));
AddTask(&pending_task);
}
void PosixDynamicThreadPool::AddTask(PendingTask* pending_task) {
DCHECK(pending_task);
// Use CHECK instead of DCHECK to crash earlier. See http://crbug.com/711167
// for details.
CHECK(pending_task->task);
AutoLock locked(lock_);
pending_tasks_.push(std::move(*pending_task));
// We have enough worker threads.
if (static_cast<size_t>(num_idle_threads_) >= pending_tasks_.size()) {
pending_tasks_available_cv_.Signal();
} else {
// The new PlatformThread will take ownership of the WorkerThread object,
// which will delete itself on exit.
WorkerThread* worker = new WorkerThread(name_prefix_, this);
PlatformThread::CreateNonJoinable(kStackSize, worker);
}
}
PendingTask PosixDynamicThreadPool::WaitForTask() {
AutoLock locked(lock_);
if (pending_tasks_.empty()) { // No work available, wait for work.
num_idle_threads_++;
if (num_idle_threads_cv_.get())
num_idle_threads_cv_->Signal();
pending_tasks_available_cv_.TimedWait(
TimeDelta::FromSeconds(idle_seconds_before_exit_));
num_idle_threads_--;
if (num_idle_threads_cv_.get())
num_idle_threads_cv_->Signal();
if (pending_tasks_.empty()) {
// We waited for work, but there's still no work. Return NULL to signal
// the thread to terminate.
return PendingTask(FROM_HERE, base::Closure());
}
}
PendingTask pending_task = std::move(pending_tasks_.front());
pending_tasks_.pop();
return pending_task;
}
} // namespace base

89
base/threading/worker_pool_posix.h

@ -1,89 +0,0 @@
// 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.
//
// The thread pool used in the POSIX implementation of WorkerPool dynamically
// adds threads as necessary to handle all tasks. It keeps old threads around
// for a period of time to allow them to be reused. After this waiting period,
// the threads exit. This thread pool uses non-joinable threads, therefore
// worker threads are not joined during process shutdown. This means that
// potentially long running tasks (such as DNS lookup) do not block process
// shutdown, but also means that process shutdown may "leak" objects. Note that
// although PosixDynamicThreadPool spawns the worker threads and manages the
// task queue, it does not own the worker threads. The worker threads ask the
// PosixDynamicThreadPool for work and eventually clean themselves up. The
// worker threads all maintain scoped_refptrs to the PosixDynamicThreadPool
// instance, which prevents PosixDynamicThreadPool from disappearing before all
// worker threads exit. The owner of PosixDynamicThreadPool should likewise
// maintain a scoped_refptr to the PosixDynamicThreadPool instance.
//
// NOTE: The classes defined in this file are only meant for use by the POSIX
// implementation of WorkerPool. No one else should be using these classes.
// These symbols are exported in a header purely for testing purposes.
#ifndef BASE_THREADING_WORKER_POOL_POSIX_H_
#define BASE_THREADING_WORKER_POOL_POSIX_H_
#include <memory>
#include <queue>
#include <string>
#include "base/callback.h"
#include "base/location.h"
#include "base/macros.h"
#include "base/memory/ref_counted.h"
#include "base/pending_task.h"
#include "base/synchronization/condition_variable.h"
#include "base/synchronization/lock.h"
#include "base/threading/platform_thread.h"
namespace base {
class BASE_EXPORT PosixDynamicThreadPool
: public RefCountedThreadSafe<PosixDynamicThreadPool> {
public:
class PosixDynamicThreadPoolPeer;
// All worker threads will share the same |name_prefix|. They will exit after
// |idle_seconds_before_exit|.
PosixDynamicThreadPool(const std::string& name_prefix,
int idle_seconds_before_exit);
// Adds |task| to the thread pool.
void PostTask(const Location& from_here, OnceClosure task);
// Worker thread method to wait for up to |idle_seconds_before_exit| for more
// work from the thread pool. Returns NULL if no work is available.
PendingTask WaitForTask();
private:
friend class RefCountedThreadSafe<PosixDynamicThreadPool>;
friend class PosixDynamicThreadPoolPeer;
~PosixDynamicThreadPool();
// Adds pending_task to the thread pool. This function will clear
// |pending_task->task|.
void AddTask(PendingTask* pending_task);
const std::string name_prefix_;
const int idle_seconds_before_exit_;
Lock lock_; // Protects all the variables below.
// Signal()s worker threads to let them know more tasks are available.
// Also used for Broadcast()'ing to worker threads to let them know the pool
// is being deleted and they can exit.
ConditionVariable pending_tasks_available_cv_;
int num_idle_threads_;
TaskQueue pending_tasks_;
// Only used for tests to ensure correct thread ordering. It will always be
// NULL in non-test code.
std::unique_ptr<ConditionVariable> num_idle_threads_cv_;
DISALLOW_COPY_AND_ASSIGN(PosixDynamicThreadPool);
};
} // namespace base
#endif // BASE_THREADING_WORKER_POOL_POSIX_H_

251
base/threading/worker_pool_posix_unittest.cc

@ -1,251 +0,0 @@
// 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/containers/queue.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 base::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

117
base/threading/worker_pool_unittest.cc

@ -1,117 +0,0 @@
// 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.h"
#include "base/bind.h"
#include "base/bind_helpers.h"
#include "base/location.h"
#include "base/message_loop/message_loop.h"
#include "base/run_loop.h"
#include "base/synchronization/waitable_event.h"
#include "base/test/test_timeouts.h"
#include "base/threading/thread_checker_impl.h"
#include "base/time/time.h"
#include "build/build_config.h"
#include "testing/gtest/include/gtest/gtest.h"
#include "testing/platform_test.h"
typedef PlatformTest WorkerPoolTest;
namespace base {
namespace {
class PostTaskAndReplyTester
: public base::RefCountedThreadSafe<PostTaskAndReplyTester> {
public:
PostTaskAndReplyTester()
: finished_(false),
test_event_(WaitableEvent::ResetPolicy::AUTOMATIC,
WaitableEvent::InitialState::NOT_SIGNALED) {}
void RunTest() {
ASSERT_TRUE(thread_checker_.CalledOnValidThread());
WorkerPool::PostTaskAndReply(
FROM_HERE,
base::BindOnce(&PostTaskAndReplyTester::OnWorkerThread, this),
base::BindOnce(&PostTaskAndReplyTester::OnOriginalThread, this), false);
test_event_.Wait();
}
void OnWorkerThread() {
// We're not on the original thread.
EXPECT_FALSE(thread_checker_.CalledOnValidThread());
test_event_.Signal();
}
void OnOriginalThread() {
EXPECT_TRUE(thread_checker_.CalledOnValidThread());
finished_ = true;
}
bool finished() const {
return finished_;
}
private:
friend class base::RefCountedThreadSafe<PostTaskAndReplyTester>;
~PostTaskAndReplyTester() {}
bool finished_;
WaitableEvent test_event_;
// The Impl version performs its checks even in release builds.
ThreadCheckerImpl thread_checker_;
};
} // namespace
TEST_F(WorkerPoolTest, PostTask) {
WaitableEvent test_event(WaitableEvent::ResetPolicy::AUTOMATIC,
WaitableEvent::InitialState::NOT_SIGNALED);
WaitableEvent long_test_event(WaitableEvent::ResetPolicy::AUTOMATIC,
WaitableEvent::InitialState::NOT_SIGNALED);
WorkerPool::PostTask(
FROM_HERE,
base::BindOnce(&WaitableEvent::Signal, base::Unretained(&test_event)),
false);
WorkerPool::PostTask(FROM_HERE,
base::BindOnce(&WaitableEvent::Signal,
base::Unretained(&long_test_event)),
true);
test_event.Wait();
long_test_event.Wait();
}
#if defined(OS_WIN) || defined(OS_LINUX)
// Flaky on Windows and Linux (http://crbug.com/130337)
#define MAYBE_PostTaskAndReply DISABLED_PostTaskAndReply
#else
#define MAYBE_PostTaskAndReply PostTaskAndReply
#endif
TEST_F(WorkerPoolTest, MAYBE_PostTaskAndReply) {
MessageLoop message_loop;
scoped_refptr<PostTaskAndReplyTester> tester(new PostTaskAndReplyTester());
tester->RunTest();
const TimeDelta kMaxDuration = TestTimeouts::tiny_timeout();
TimeTicks start = TimeTicks::Now();
while (!tester->finished() && TimeTicks::Now() - start < kMaxDuration) {
#if defined(OS_IOS)
// Ensure that the other thread has a chance to run even on a single-core
// device.
pthread_yield_np();
#endif
RunLoop().RunUntilIdle();
}
EXPECT_TRUE(tester->finished());
}
} // namespace base

73
base/threading/worker_pool_win.cc

@ -1,73 +0,0 @@
// 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.h"
#include <utility>
#include "base/bind.h"
#include "base/callback.h"
#include "base/logging.h"
#include "base/pending_task.h"
#include "base/threading/thread_local.h"
#include "base/trace_event/trace_event.h"
namespace base {
namespace {
ThreadLocalBoolean* GetWorkerPoolRunningOnThisThread() {
static auto* thread_local_boolean = new ThreadLocalBoolean();
return thread_local_boolean;
}
DWORD CALLBACK WorkItemCallback(void* param) {
PendingTask* pending_task = static_cast<PendingTask*>(param);
TRACE_TASK_EXECUTION("WorkerThread::ThreadMain::Run", *pending_task);
GetWorkerPoolRunningOnThisThread()->Set(true);
std::move(pending_task->task).Run();
GetWorkerPoolRunningOnThisThread()->Set(false);
delete pending_task;
return 0;
}
// Takes ownership of |pending_task|
bool PostTaskInternal(PendingTask* pending_task, bool task_is_slow) {
// Use CHECK instead of DCHECK to crash earlier. See http://crbug.com/711167
// for details.
CHECK(pending_task->task);
ULONG flags = 0;
if (task_is_slow)
flags |= WT_EXECUTELONGFUNCTION;
if (!QueueUserWorkItem(WorkItemCallback, pending_task, flags)) {
DPLOG(ERROR) << "QueueUserWorkItem failed";
delete pending_task;
return false;
}
return true;
}
} // namespace
// static
bool WorkerPool::PostTask(const Location& from_here,
base::OnceClosure task,
bool task_is_slow) {
PendingTask* pending_task = new PendingTask(from_here, std::move(task));
return PostTaskInternal(pending_task, task_is_slow);
}
// static
bool WorkerPool::RunsTasksOnCurrentThread() {
return GetWorkerPoolRunningOnThisThread()->Get();
}
} // namespace base

8
components/metrics/file_metrics_provider.cc

@ -533,10 +533,10 @@ void FileMetricsProvider::RecordSourceAsRead(SourceInfo* source) {
void FileMetricsProvider::OnDidCreateMetricsLog() { void FileMetricsProvider::OnDidCreateMetricsLog() {
DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
// Schedule a check to see if there are new metrics to load. If so, they // Schedule a check to see if there are new metrics to load. If so, they will
// will be reported during the next collection run after this one. The // be reported during the next collection run after this one. The check is run
// check is run off of the worker-pool so as to not cause delays on the // off of a MayBlock() TaskRunner so as to not cause delays on the main UI
// main UI thread (which is currently where metric collection is done). // thread (which is currently where metric collection is done).
ScheduleSourcesCheck(); ScheduleSourcesCheck();
// Clear any data for initial metrics since they're always reported // Clear any data for initial metrics since they're always reported

6
content/browser/loader/resource_dispatcher_host_unittest.cc

@ -3496,9 +3496,9 @@ TEST_F(ResourceDispatcherHostTest, DownloadToFile) {
OnMessageReceived(request_msg, filter_.get()); OnMessageReceived(request_msg, filter_.get());
// Running the message loop until idle does not work because // Running the message loop until idle does not work because
// RedirectToFileResourceHandler posts things to base::WorkerPool. Instead, // ResourceDispatcherHostImpl posts tasks to TaskScheduler. Instead, wait for
// wait for the ResourceMsg_RequestComplete to go out. Then run the event loop // the ResourceMsg_RequestComplete to go out. Then, wait until no
// until idle so the loader is gone. // TaskScheduler or main thread tasks remain.
WaitForRequestComplete(); WaitForRequestComplete();
content::RunAllBlockingPoolTasksUntilIdle(); content::RunAllBlockingPoolTasksUntilIdle();
EXPECT_EQ(0, host_.pending_requests()); EXPECT_EQ(0, host_.pending_requests());

1
crypto/nss_util.cc

@ -34,7 +34,6 @@
#include "base/threading/thread_checker.h" #include "base/threading/thread_checker.h"
#include "base/threading/thread_restrictions.h" #include "base/threading/thread_restrictions.h"
#include "base/threading/thread_task_runner_handle.h" #include "base/threading/thread_task_runner_handle.h"
#include "base/threading/worker_pool.h"
#include "build/build_config.h" #include "build/build_config.h"
#include "crypto/nss_crypto_module_delegate.h" #include "crypto/nss_crypto_module_delegate.h"
#include "crypto/nss_util_internal.h" #include "crypto/nss_util_internal.h"