Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(1272)

Unified Diff: content/common/sequenced_worker_pool.cc

Issue 8416019: Add a sequenced worker pool (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/src/
Patch Set: '' Created 9 years, 1 month ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « content/common/sequenced_worker_pool.h ('k') | content/common/sequenced_worker_pool_unittest.cc » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: content/common/sequenced_worker_pool.cc
===================================================================
--- content/common/sequenced_worker_pool.cc (revision 0)
+++ content/common/sequenced_worker_pool.cc (revision 0)
@@ -0,0 +1,513 @@
+// Copyright (c) 2011 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 "content/common/sequenced_worker_pool.h"
+
+#include <deque>
+#include <set>
+
+#include "base/atomicops.h"
+#include "base/bind.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/metrics/histogram.h"
+#include "base/threading/thread.h"
+#include "base/stringprintf.h"
+#include "base/synchronization/condition_variable.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/simple_thread.h"
+
+namespace {
+
+struct SequencedTask {
+ int sequence_token_id;
+ SequencedWorkerPool::WorkerShutdown shutdown_behavior;
+ tracked_objects::Location location;
+ base::Closure task;
+};
+
+} // namespace
+
+// Worker ---------------------------------------------------------------------
+
+class SequencedWorkerPool::Worker : public base::SimpleThread {
+ public:
+ Worker(SequencedWorkerPool::Inner* inner, int thread_number);
+ ~Worker();
+
+ // SimpleThread implementation. This actually runs the background thread.
+ virtual void Run();
+
+ // When running a task, the shutdown mode is stored on the worker. It will
+ // be INTERRUPT_ON_SHUTDOWN if there is no running task.
+ //
+ // This must only be called when holding the Inner class' lock.
+ SequencedWorkerPool::WorkerShutdown current_shutdown_mode() const {
+ return current_shutdown_mode_;
+ }
+ void set_current_shutdown_mode(SequencedWorkerPool::WorkerShutdown mode) {
+ current_shutdown_mode_ = mode;
+ }
+ void reset_current_shutdown_mode() {
+ // Resets back to the default.
+ current_shutdown_mode_ = SequencedWorkerPool::CONTINUE_ON_SHUTDOWN;
+ }
+
+ private:
+ SequencedWorkerPool::Inner* inner_;
+ SequencedWorkerPool::WorkerShutdown current_shutdown_mode_;
+
+ DISALLOW_COPY_AND_ASSIGN(Worker);
+};
+
+
+// Inner ----------------------------------------------------------------------
+
+class SequencedWorkerPool::Inner
+ : public base::RefCountedThreadSafe<SequencedWorkerPool::Inner> {
+ public:
+ Inner(size_t max_threads);
+ virtual ~Inner();
+
+ // Backends for SequenceWorkerPool.
+ SequenceToken GetSequenceToken();
+ SequenceToken GetNamedSequenceToken(const std::string& name);
+ bool PostTask(int sequence_token_id,
+ SequencedWorkerPool::WorkerShutdown shutdown_behavior,
+ const tracked_objects::Location& from_here,
+ const base::Closure& task);
+ void Shutdown();
+ void SetTestingObserver(SequencedWorkerPool::TestingObserver* observer);
+
+ // Runs the worker loop on the background thread.
+ void ThreadLoop(Worker* this_worker);
+
+ private:
+ bool GetWork(SequencedTask* task);
+
+ void MarkWorkerBusy(Worker* worker, const SequencedTask& task);
+ void MarkWorkerFree(Worker* worker);
+
+ // Returns true if there are no threads currently running the given
+ // sequence token.
+ bool IsSequenceTokenRunnable(int sequence_token_id) const;
+
+ // Returns true if there are more currently runnable tasks than there are
+ // availble threads to run them. This function does a brute-force check of
+ // all tasks so avoid calling when possible.
+ //
+ // Must be called from within the lock.
+ bool RunnableTasksAreOverSubscribed() const;
jar (doing other things) 2011/11/25 23:06:09 I don't think you'll need this function.
+
+ // Returns true if any worker is running a shutdown-blocking task. This
+ // includes BLOCK_SHUTDOWN and also SKIP_ON_SHUTDOWN tasks that are already
+ // running (since they must be run to completion). It must be called with
+ // the lock held.
+ bool IsRunningBlockingTask() const;
+
+ // The last sequence number used. Managed by GetSequenceToken, since this
+ // only does threadsafe increment operations, you do not need to hold the
+ // lock.
+ volatile base::subtle::Atomic32 last_sequence_number_;
+
+ // This lock protects |everything in this class|. Do not read or modify
+ // anything without holding this lock. Do not block while holding this
+ // lock.
+ base::Lock lock_;
+
+ // Condition variable used to wake up worker threads when a task is runnable.
+ base::ConditionVariable cond_var_;
+
+ // The maximum number of worker threads we'll create.
+ size_t max_threads_;
+
+ // Associates all known sequence token names with their IDs.
+ std::map<std::string, int> named_sequence_tokens_;
+
+ // Owning pointers to all threads we've created so far. Since we lazily
+ // create threads, this may be less than max_threads_ and will be initially
+ // empty.
+ std::vector<linked_ptr<Worker> > threads_;
+
+ // Lists all workers currently running tasks. Contains non-owning pointers.
+ std::set<Worker*> running_threads_;
jar (doing other things) 2011/11/25 23:06:09 I'd expect each thread to hold its own status. At
+
+ // In-order list of all pending tasks. These are tasks waiting for a thread
+ // to run on or that are blocked on a previous task in their sequence.
+ //
+ // We maintain the pending_task_count_ separately for metrics because
+ // list.size() can be linear time.
+ std::list<SequencedTask> pending_tasks_;
+ size_t pending_task_count_;
+
+ // Lists all sequence tokens currently executing.
+ std::set<int> current_sequences_;
+
+ // Set when the app is terminating and no further tasks should be allowed,
+ // though we may still be running existing tasks.
+ bool terminating_;
+
+ // Set when the worker pool is being destroyed, and all worker threads should
+ // exit.
+ bool exit_workers_;
+
+ SequencedWorkerPool::TestingObserver* testing_observer_;
+
+ // Created lazily when terminating_ is set and there are pending tasks, this
+ // is signaled by ScheduleWork whel all blocking tasks have completed.
+ scoped_ptr<base::WaitableEvent> shutdown_complete_;
jar (doing other things) 2011/11/25 23:06:09 The condition variable would be the natural thing
+};
+
+SequencedWorkerPool::Worker::Worker(SequencedWorkerPool::Inner* inner,
+ int thread_number)
+ : base::SimpleThread(
+ StringPrintf("BrowserWorker%d", thread_number).c_str()),
+ inner_(inner),
+ current_shutdown_mode_(SequencedWorkerPool::CONTINUE_ON_SHUTDOWN) {
+ Start();
+}
+
+SequencedWorkerPool::Worker::~Worker() {
+ Join();
+}
+
+void SequencedWorkerPool::Worker::Run() {
+ // Just jump back to the Inner object to run the thread, since it has all the
+ // tracking information and queues. It might be more natural to implement
+ // using DelegateSimpleThread and have Inner implement the Delegate to avoid
+ // having these worker objects at all, but that method lacks the ability to
+ // send thread-specific information easily to the thread loop.
+ inner_->ThreadLoop(this);
+}
+
+SequencedWorkerPool::Inner::Inner(size_t max_threads)
+ : last_sequence_number_(0),
+ lock_(),
+ cond_var_(&lock_),
+ max_threads_(max_threads),
+ pending_task_count_(0),
+ terminating_(false),
+ exit_workers_(false) {
+}
+
+SequencedWorkerPool::Inner::~Inner() {
+ // Tell all workers to exit. The Worker destructor will actually join with
+ // each corresponding thread when the object is torn down.
+ exit_workers_ = true;
+ cond_var_.Broadcast();
+}
+
+SequencedWorkerPool::SequenceToken
+SequencedWorkerPool::Inner::GetSequenceToken() {
+ base::subtle::Atomic32 result =
+ base::subtle::NoBarrier_AtomicIncrement(&last_sequence_number_, 1);
+ return SequenceToken(static_cast<int>(result));
+}
+
+SequencedWorkerPool::SequenceToken
+SequencedWorkerPool::Inner::GetNamedSequenceToken(
+ const std::string& name) {
+ base::AutoLock lock(lock_);
+ std::map<std::string, int>::const_iterator found =
+ named_sequence_tokens_.find(name);
+ if (found != named_sequence_tokens_.end())
+ return SequenceToken(found->second); // Got an existing one.
+
+ // Create a new one for this name.
+ SequenceToken result = GetSequenceToken();
+ named_sequence_tokens_.insert(std::make_pair(name, result.id_));
+ return result;
+}
+
+bool SequencedWorkerPool::Inner::PostTask(
+ int sequence_token_id,
+ SequencedWorkerPool::WorkerShutdown shutdown_behavior,
+ const tracked_objects::Location& from_here,
+ const base::Closure& task) {
+ base::AutoLock lock(lock_);
+
+ if (terminating_)
+ return false;
+
+ SequencedTask sequenced;
+ sequenced.sequence_token_id = sequence_token_id;
+ sequenced.shutdown_behavior = shutdown_behavior;
+ sequenced.location = from_here;
+ sequenced.task = task;
+ pending_tasks_.push_back(sequenced);
+ pending_task_count_++;
+
+ // Start a new thread if it would be useful for this task.
+ //
+ // There is a race condition between posting tasks and having the workers
+ // pick them up. The main thread could post two tasks with the same sequence
+ // token before a worker picks the first one up. When the second task is
+ // posted, the sequence token will not be marked as running and a quick check
+ // of task and free thread counts would indicate that a new thread should be
jar (doing other things) 2011/11/25 23:06:09 I think what you left out of this argument is that
+ // started. However, this may not be the case since there could be runnable
+ // tasks and free threads just haven't picked them up yet.
+ //
+ // Therefore, we do a more thorough brute-force check of runnable tasks to
jar (doing other things) 2011/11/25 23:06:09 I think we should just maintain a few counters, an
+ // avoid creating extra threads. This should generally be fast, and most
+ // cases will be caught by the first two checks of the if statement, meaning
+ // the slow RunnableTasksAreOverSubscribed() is only run occationally.
+ if (threads_.size() < max_threads_ &&
+ IsSequenceTokenRunnable(sequence_token_id) &&
+ RunnableTasksAreOverSubscribed()) {
+ // Start a new thread. This is done from within the lock which sucks (since
+ // it's not necessarily fast and it could block other work), but avoiding
jar (doing other things) 2011/11/25 23:06:09 All you do inside the lock is bump the counter on
+ // this would be a lot more code.
+ threads_.push_back(linked_ptr<Worker>(new Worker(this, threads_.size())));
jar (doing other things) 2011/11/25 23:06:09 I don't think you need to maintain a list of threa
+ }
+
+ // Wake up a worker thread to run this.
+ cond_var_.Signal();
+
+ return true;
+}
+
+void SequencedWorkerPool::Inner::Shutdown() {
+ // Lists all tasks that we're dropping rather than running. The closure's
+ // destructor may do complicated things like release refcounted objects,
+ // which we don't want to do in the lock. Since the closures are internally
+ // refcounted, all we have to do is keep a ref to the closure alive longer
+ // than the lock is held. That's what this vector does, so the cleanup of
+ // the tasks will magically happen in the vector's destructor.
+ std::vector<SequencedTask> dropped_tasks;
+
+ // Mark us as terminated and go through and drop all tasks that aren't
+ // required to run on shutdown. Since no new tasks will get posted once the
+ // terminated flag is set, this ensures that all remaining tasks are required
+ // for shutdown whenever the termianted_ flag is set.
+ {
+ base::AutoLock lock(lock_);
+ DCHECK(!terminating_);
+ terminating_ = true;
+
+ std::list<SequencedTask>::iterator i = pending_tasks_.begin();
+ while (i != pending_tasks_.end()) {
+ if (i->shutdown_behavior == BLOCK_SHUTDOWN) {
+ i++;
+ } else {
+ dropped_tasks.push_back(*i);
jar (doing other things) 2011/11/25 23:06:09 This cleanup action should happen in a worker thre
brettw 2011/11/26 01:54:40 How can you guarantee that there is a worker threa
jar (doing other things) 2011/11/26 06:45:36 Interesting question. I hadn't thought about tha
+ i = pending_tasks_.erase(i);
+ pending_task_count_--;
+ }
+ }
+ DCHECK_EQ(pending_tasks_.size(), pending_task_count_);
+
+ if (pending_tasks_.empty() && !IsRunningBlockingTask()) {
+ // There are no pending or running tasks blocking shutdown, we're done.
+ return;
+ }
+
+ // Need to wait for some tasks, create the event.
+ DCHECK(!shutdown_complete_.get());
+ shutdown_complete_.reset(new base::WaitableEvent(false, false));
jar (doing other things) 2011/11/25 23:06:09 Instead of using a waitable event, you should wait
brettw 2011/11/26 01:54:40 You want me to wait on the same condition variable
jar (doing other things) 2011/11/26 06:45:36 Clearly the goal is to wait until the system achiv
+ }
+
+ // If we get here, we know we're either waiting on a blocking task that's
+ // currently running, waiting on a blocking task that hasn't been scheduled
+ // yet, or both. Block on the "queue empty" event to know when all tasks are
+ // complete. This must be done outside the lock.
+ if (testing_observer_)
+ testing_observer_->WillWaitForShutdown();
+
+ base::TimeTicks shutdown_wait_begin = base::TimeTicks::Now();
+ shutdown_complete_->Wait();
+ UMA_HISTOGRAM_TIMES("SequencedWorkerPool.ShutdownDelayTime",
+ base::TimeTicks::Now() - shutdown_wait_begin);
+}
+
+void SequencedWorkerPool::Inner::SetTestingObserver(
+ SequencedWorkerPool::TestingObserver* observer) {
+ base::AutoLock lock(lock_);
+ testing_observer_ = observer;
+}
+
+void SequencedWorkerPool::Inner::ThreadLoop(Worker* this_worker) {
+ base::AutoLock lock(lock_);
+ while (!exit_workers_) {
+ SequencedTask task;
+ if (GetWork(&task)) {
+ MarkWorkerBusy(this_worker, task); // Must be done inside the lock.
+ {
+ base::AutoUnlock unlock(lock_);
+ task.task.Run();
jar (doing other things) 2011/11/26 01:25:59 One other thing... it is common to signal after yo
brettw 2011/11/26 01:54:40 You're right, that's what was causing my problem.
jar (doing other things) 2011/11/26 06:45:36 The trick is to just move incrementally in the rig
+ }
+ MarkWorkerFree(this_worker); // Must be done inside the lock.
+ }
+
+ cond_var_.Wait();
+ }
+}
+
+bool SequencedWorkerPool::Inner::GetWork(SequencedTask* task) {
+ lock_.AssertAcquired();
+
+ DCHECK_EQ(pending_tasks_.size(), pending_task_count_);
+ UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount",
+ pending_task_count_);
+
+ // Find the next task with a sequence token that's not currently in use.
+ // If the token is in use, that means another thread is running something
+ // in that sequence, and we can't run it without going out-of-order.
+ //
+ // This algorithm is simple and fair, but inefficient in some cases. For
+ // example, say somebody schedules 1000 slow tasks with the same sequence
+ // number. We'll have to go through all those tasks each time we feel like
+ // there might be work to schedule. If this proves to be a problem, we
+ // should make this more efficient.
+ //
+ // One possible enhancement would be to keep a map from sequence ID to a
+ // list of pending but currently blocked SequencedTasks for that ID.
+ // When a worker finishes a task of one sequence token, it can pick up the
+ // next one from that token right away.
+ //
+ // This may lead to starvation if there are sufficient numbers of sequences
+ // in use. To alleviate this, we could add an incrementing priority counter
+ // to each SequencedTask. Then maintain a priority_queue of all runnable
+ // tasks, sorted by priority counter. When a sequenced task is completed
+ // we would pop the head element off of that tasks pending list and add it
+ // to the priority queue. Then we would run the first item in the priority
+ // queue.
+ bool found_task = false;
+ int unrunnable_tasks = 0;
+ for (std::list<SequencedTask>::iterator i = pending_tasks_.begin();
+ i != pending_tasks_.end(); ++i) {
+ if (IsSequenceTokenRunnable(i->sequence_token_id)) {
+ // Found a runnable task.
+ *task = *i;
+ pending_tasks_.erase(i);
+ pending_task_count_--;
+
+ // Mark the task's sequence number as in use.
+ if (task->sequence_token_id)
+ current_sequences_.insert(task->sequence_token_id);
+
+ found_task = true;
+ break;
+ }
+ unrunnable_tasks++;
+ }
+
+ // Track the number of tasks we had to skip over to see if we should be
+ // making this more efficient. If this number ever becomes large or is
+ // frequently "some", we should consider the optimization above.
+ UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.UnrunnableTaskCount",
+ unrunnable_tasks);
+ return found_task;
+}
+
+void SequencedWorkerPool::Inner::MarkWorkerBusy(Worker* worker,
+ const SequencedTask& task) {
+ lock_.AssertAcquired();
+ DCHECK(running_threads_.find(worker) == running_threads_.end());
+
+ worker->set_current_shutdown_mode(task.shutdown_behavior);
+ running_threads_.insert(worker);
+}
+
+void SequencedWorkerPool::Inner::MarkWorkerFree(Worker* worker) {
+ lock_.AssertAcquired();
+ DCHECK(running_threads_.find(worker) != running_threads_.end());
+
+ worker->reset_current_shutdown_mode();
+ running_threads_.erase(worker);
+
+ if (terminating_ && shutdown_complete_.get()) {
+ // When the app is terminating, check for "no more blocking work" and
+ // signal if shutdown tasks are complete.
+ if (pending_tasks_.empty() && !IsRunningBlockingTask())
+ shutdown_complete_->Signal();
+ }
+}
+
+bool SequencedWorkerPool::Inner::IsSequenceTokenRunnable(
+ int sequence_token_id) const {
+ lock_.AssertAcquired();
+ return !sequence_token_id ||
+ current_sequences_.find(sequence_token_id) ==
+ current_sequences_.end();
+}
+
+bool SequencedWorkerPool::Inner::RunnableTasksAreOverSubscribed() const {
+ lock_.AssertAcquired();
+
+ size_t available_threads = threads_.size() - running_threads_.size();
+
+ // Fast path the (relatively common) case where there are more threads
+ // available than there are tasks to run.
+ if (pending_task_count_ <= available_threads)
+ return false;
+
+ size_t runnable_task_count = 0;
+ for (std::list<SequencedTask>::const_iterator i = pending_tasks_.begin();
+ i != pending_tasks_.end(); ++i) {
+ if (IsSequenceTokenRunnable(i->sequence_token_id)) {
+ runnable_task_count++;
+
+ // We can return as soon as we've found more runnable tasks than
+ // available threads.
+ if (runnable_task_count > available_threads)
+ return true;
+ }
+ }
+
+ // Did not fund more runnable tasks than threads.
+ return false;
+}
+
+bool SequencedWorkerPool::Inner::IsRunningBlockingTask() const {
+ lock_.AssertAcquired();
+
+ for (std::set<Worker*>::const_iterator i = running_threads_.begin();
+ i != running_threads_.end(); ++i) {
+ if ((*i)->current_shutdown_mode() == SequencedWorkerPool::BLOCK_SHUTDOWN ||
+ (*i)->current_shutdown_mode() == SequencedWorkerPool::SKIP_ON_SHUTDOWN)
+ return true;
+ }
+ return false;
+}
+
+// SequencedWorkerPool --------------------------------------------------------
+
+SequencedWorkerPool::SequencedWorkerPool(size_t max_threads)
+ : inner_(new Inner(max_threads)) {
+}
+
+SequencedWorkerPool::~SequencedWorkerPool() {
+}
+
+SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() {
+ return inner_->GetSequenceToken();
+}
+
+SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken(
+ const std::string& name) {
+ return inner_->GetNamedSequenceToken(name);
+}
+
+bool SequencedWorkerPool::PostWorkerTask(
+ WorkerShutdown shutdown_behavior,
+ const tracked_objects::Location& from_here,
+ const base::Closure& task) {
+ return inner_->PostTask(0, shutdown_behavior, from_here, task);
+}
+
+bool SequencedWorkerPool::PostSequencedWorkerTask(
+ SequenceToken sequence_token,
+ WorkerShutdown shutdown_behavior,
+ const tracked_objects::Location& from_here,
+ const base::Closure& task) {
+ return inner_->PostTask(sequence_token.id_, shutdown_behavior,
+ from_here, task);
+}
+
+void SequencedWorkerPool::Shutdown() {
+ inner_->Shutdown();
+}
+
+void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) {
+ inner_->SetTestingObserver(observer);
+}
« no previous file with comments | « content/common/sequenced_worker_pool.h ('k') | content/common/sequenced_worker_pool_unittest.cc » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698