Chromium Code Reviews| 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); |
| +} |