Add a sequenced worker pool

content/common/sequenced_worker_pool.cc

+// 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 best, I'd expect counts to be
maintained in this controlling singleton.

+
+  // 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 to wait for, even on a
non-worker thread.

+};
+
+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 the thread that was just
started to handle the unused sequence token will be entrusted to create another
thread if need be.  So long as you know there are threads waiting for your
signal, you should let them handle the work of starting a (single) new thead if
need be.

+  // 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, and let them form the
"conditions" that are checked, and there would be no new race, or complications.
 

When someone posts a task, if we have more threads running then we have sequence
tokens in the active set, then everything will sort itself out (no need to
figure out if you should start yet another thread or not... the extra running
thread will do that soon enough).

The only time to start a new thread is when all threads are busy, and the new
token is not in the active set.  There is then a race of sorts, as an active
thread can always race to finish against the startup of the new thread, but that
is inevitable with any implementation.  No matter who starts a new thread, it is
possible that instantly we'll have all the other threads finish their work (they
can even be waiting for the lock as your deciding!), leaving us wondering why we
started a new thread.  The difference is that with conditional variables you
notice this race (logic is simpler, and you can see it), whereas with the
previous implementation you started a message loop and task, and ignored the
startup time (which was actually much more extensive).

Bottom line: a race can always cause us to create one more thread than we "could
have gotten away with."  

+  // 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 the number of started threads.
 You then exit the lock, and start the thread.

+    // 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 threads. I think they should be
self owning.  As a result, I don't think you need this push inside the lock.

I think you need a count of the number of threads that need to be waited for at
shutdown.  That is what any caller that wants to effectivel join the threads
should check for as a "condition" when they are waiting for those urgent threads
to complete.

+  }
+
+  // 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 thread. This will also keep such
cleanup off the critical path of the thread that called shutdown, unless you
want to assert that cleanup (delete of discarded tasks) must be waited for.  If
so, then worker threads should each take it upon themselves to delete one
discarded task at a time... and they'd all run in parallel.

[brettw, 2011/11/26 01:54:40]

How can you guarantee that there is a worker thread available to run the
discarded event? They may all be running CONTINUE_ON_SHUTDOWN tasks and will
never run more code from us again.

[jar (doing other things), 2011/11/26 06:45:36]

Interesting question.   I hadn't thought about that as a requirement.  It is a
great question.   Do you want the semantics to guarantee destruction of discard
tasks?  I'm surprised to see you suggest that you're willing to wait for them to
get destroyed, but yet you're not waiting for them to complete a run :-/.

IMO, given that you're willing to leave some non-critical tasks running, it is
no big deal to leave some tasks undeleted if you get "too busy."  I'd expect
that more typically, the non-critical tasks would finish, and you'd soon be down
to a few critical tasks, and a bunch of idle threads... just waiting to do your
deletions.  

Your question is very similar to the question of what to do if you have all your
threads "stuck" running non-critical tasks, but you  need to run critical tasks
at shutdown.  What do you think the right thing is to do?

Do you want the semantics to be that the "must run" tasks will be run only on a
restricted set of worker threads?  Are you willing to sacrifice the thread count
limit?  ...or is that firm, and you'll wait indefinitely for a time-slot on a
busy worker thread?   ... or do you want to try to run the "must run" tasks on
the thread that calls for a shutdown, since it was going to wait indefinitely
anyway... hmm.

My guess is that this is a great question... and we should settle on the
semantics. 

WDYT?  Hang?

At the heart of this, the current official release has a shutdown problem that
revolves around a non-critical(?) task hanging, and blocking all availability of
"the thread pool" (a single File Thread) from running a critical task.

I think that during shutdown, you have to make some extra threads available to
ensure that you can service the critical tasks.  I suspect that otherwise, we'll
sometimes get a bunch of hung threads, and have no thread power to handle the
shutdown.  This "reserve tank" approach is commonly used in memory allocators to
facilitate more controlled crashes when we're "out of memory."  Maybe we need a
reserve number of threads for such emergencies.

On 2011/11/26 01:54:40, brettw wrote:

+        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 on the condition variable,
testing to see that cleanup and runs are complete.

[brettw, 2011/11/26 01:54:40]

You want me to wait on the same condition variable I use for the workers? That
seems super confusing to me.

[jar (doing other things), 2011/11/26 06:45:36]

Clearly the goal is to wait until the system achives some "state" (all required
threads have run). It seems really natural to wait for such an event.

CVs are really good about avoiding the race between checking state, and waiting.
 CVs can be built atop waitable events, and vice versa.

+  }
+
+  // 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 you take on work, to have some
other worker check to see if there is more work.  You could spend the time in
this thread seeing if you "need help" servicing everything, or you could just
signal, and let some other thread check (in parallel) to see if there is more
work to do.

That missing line might be a part of your deadlock also.

Since you did go with a lazy start of threads, you'll also need a boolean to
indicate if you should spawn another worker just before you start to do your
"real" work (the call of Run()).  

The entity that decides to spawn a new thread should bump up the count on active
threads (inside the lock), so there is no race or confusion about the count of
active threads.

[brettw, 2011/11/26 01:54:40]

You're right, that's what was causing my problem. I think this adds to my
argument that condition variables are more difficult to understand than the
normal Chrome constructs.
I'm going to check the queue every time I take something out. It's difficult to
figure this out ahead of time since there may be multiple threads required.

[jar (doing other things), 2011/11/26 06:45:36]

The trick is to just move incrementally in the right direction.  You don't need
to start a lot of threads. At most, one more... and that thread, when it starts,
may choose to start another.

IMO, the simple approach is to always have one extra on hand.  Any time you run
a task, if you think your are the last idle thread, and you're belowe the limit,
you should start another.  As I said, you should try to make it simple, and just
look at the few state variables (the conditions) to see what you should do.

If you want to be fancy, when you call the function that finds the next task
that is eligible to run, that function can return a bool indicating if there are
other tasks that could have been run as well.  That would give you the minimal
thread creation.

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