Fix up SequencedWorkerPool in preparation for making it a TaskRunner

base/threading/sequenced_worker_pool.h

Index: base/threading/sequenced_worker_pool.h
diff --git a/base/threading/sequenced_worker_pool.h b/base/threading/sequenced_worker_pool.h
index f5770a72b966e21bf56cfadc15de3edb0ea36877..2bd3d581f5fb52eeac54c9d8ca229df9f115c603 100644
--- a/base/threading/sequenced_worker_pool.h
+++ b/base/threading/sequenced_worker_pool.h
@@ -1,4 +1,4 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -6,13 +6,22 @@
 #define BASE_THREADING_SEQUENCED_WORKER_POOL_H_
 #pragma once
 
+#include <cstddef>
+#include <list>
+#include <map>
+#include <set>
 #include <string>
+#include <vector>
 
+#include "base/atomicops.h"
+#include "base/base_export.h"
+#include "base/basictypes.h"
 #include "base/callback.h"
 #include "base/memory/linked_ptr.h"
 #include "base/memory/ref_counted.h"
+#include "base/synchronization/condition_variable.h"

[brettw, 2012/02/24 05:43:34]

One of the reasons I liked the inner class was that lots of code will use this,
and it keeps the dependencies under control. This is quite a lot of big
dependencies.

+#include "base/synchronization/lock.h"
 #include "base/tracked_objects.h"
-#include "base/base_export.h"
 
 namespace base {
 
@@ -46,7 +55,8 @@ namespace base {
 // 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.
-class BASE_EXPORT SequencedWorkerPool {
+class BASE_EXPORT SequencedWorkerPool
+    : public RefCountedThreadSafe<SequencedWorkerPool> {
  public:
   // Defines what should happen to a task posted to the worker pool on shutdown.
   enum WorkerShutdown {
@@ -87,10 +97,10 @@ class BASE_EXPORT SequencedWorkerPool {
   };
 
   // Opaque identifier that defines sequencing of tasks posted to the worker
-  // pool. See NewSequenceToken().
+  // pool.
   class SequenceToken {
    public:
-    explicit SequenceToken() : id_(0) {}
+    SequenceToken() : id_(0) {}
     ~SequenceToken() {}
 
     bool Equals(const SequenceToken& other) const {
@@ -100,7 +110,7 @@ class BASE_EXPORT SequencedWorkerPool {
    private:
     friend class SequencedWorkerPool;
 
-    SequenceToken(int id) : id_(id) {}
+    explicit SequenceToken(int id) : id_(id) {}
 
     int id_;
   };
@@ -116,7 +126,6 @@ class BASE_EXPORT SequencedWorkerPool {
   // and a prefix for the thread name to ad in debugging.
   SequencedWorkerPool(size_t max_threads,
                       const std::string& thread_name_prefix);
-  ~SequencedWorkerPool();
 
   // Returns a unique token that can be used to sequence tasks posted to
   // PostSequencedWorkerTask(). Valid tokens are alwys nonzero.
@@ -149,12 +158,12 @@ class BASE_EXPORT SequencedWorkerPool {
   // Returns true if the task was posted successfully. This may fail during
   // shutdown regardless of the specified ShutdownBehavior.
   bool PostWorkerTask(const tracked_objects::Location& from_here,
-                      const base::Closure& task);
+                      const Closure& task);
 
   // Same as PostWorkerTask but allows specification of the shutdown behavior.
   bool PostWorkerTaskWithShutdownBehavior(
       const tracked_objects::Location& from_here,
-      const base::Closure& task,
+      const Closure& task,
       WorkerShutdown shutdown_behavior);
 
   // Like PostWorkerTask above, but provides sequencing semantics. This means
@@ -170,20 +179,20 @@ class BASE_EXPORT SequencedWorkerPool {
   // shutdown regardless of the specified ShutdownBehavior.
   bool PostSequencedWorkerTask(SequenceToken sequence_token,
                                const tracked_objects::Location& from_here,
-                               const base::Closure& task);
+                               const Closure& task);
 
   // Like PostSequencedWorkerTask above, but allows you to specify a named
   // token, which saves an extra call to GetNamedSequenceToken.
   bool PostNamedSequencedWorkerTask(const std::string& token_name,
                                     const tracked_objects::Location& from_here,
-                                    const base::Closure& task);
+                                    const Closure& task);
 
   // Same as PostSequencedWorkerTask but allows specification of the shutdown
   // behavior.
   bool PostSequencedWorkerTaskWithShutdownBehavior(
       SequenceToken sequence_token,
       const tracked_objects::Location& from_here,
-      const base::Closure& task,
+      const Closure& task,
       WorkerShutdown shutdown_behavior);
 
   // Blocks until all pending tasks are complete. This should only be called in
@@ -206,13 +215,135 @@ class BASE_EXPORT SequencedWorkerPool {
   void SetTestingObserver(TestingObserver* observer);
 
  private:
-  class Inner;
+  friend class RefCountedThreadSafe<SequencedWorkerPool>;
   class Worker;
 
-  friend class Inner;
-  friend class Worker;
+  struct SequencedTask {
+    SequencedTask();
+    ~SequencedTask();
+
+    int sequence_token_id;
+    WorkerShutdown shutdown_behavior;
+    tracked_objects::Location location;
+    Closure task;
+  };
+
+  ~SequencedWorkerPool();
+
+  // This function accepts a name and an ID. If the name is null, the
+  // token ID is used. This allows us to implement the optional name lookup
+  // from a single function without having to enter the lock a separate time.
+  bool PostTaskHelper(const std::string* optional_token_name,
+                      SequenceToken sequence_token,
+                      WorkerShutdown shutdown_behavior,
+                      const tracked_objects::Location& from_here,
+                      const Closure& task);
+
+  // Runs the worker loop on the background thread.
+  void ThreadLoop(Worker* this_worker);
+
+  // Called from within the lock, this converts the given token name into a
+  // token ID, creating a new one if necessary.
+  int LockedGetNamedTokenID(const std::string& name);
+
+  // The calling code should clear the given delete_these_oustide_lock
+  // vector the next time the lock is released. See the implementation for
+  // a more detailed description.
+  bool GetWork(SequencedTask* task,
+               std::vector<Closure>* delete_these_outside_lock);
+
+  // Peforms init and cleanup around running the given task. WillRun...
+  // returns the value from PrepareToStartAdditionalThreadIfNecessary.
+  // The calling code should call FinishStartingAdditionalThread once the
+  // lock is released if the return values is nonzero.
+  int WillRunWorkerTask(const SequencedTask& task);
+  void DidRunWorkerTask(const SequencedTask& task);
+
+  // Returns true if there are no threads currently running the given
+  // sequence token.
+  bool IsSequenceTokenRunnable(int sequence_token_id) const;
+
+  // Checks if all threads are busy and the addition of one more could run an
+  // additional task waiting in the queue. This must be called from within
+  // the lock.
+  //
+  // If another thread is helpful, this will mark the thread as being in the
+  // process of starting and returns the index of the new thread which will be
+  // 0 or more. The caller should then call FinishStartingAdditionalThread to
+  // complete initialization once the lock is released.
+  //
+  // If another thread is not necessary, returne 0;
+  //
+  // See the implementedion for more.
+  int PrepareToStartAdditionalThreadIfHelpful();
+
+  // The second part of thread creation after
+  // PrepareToStartAdditionalThreadIfHelpful with the thread number it
+  // generated. This actually creates the thread and should be called outside
+  // the lock to avoid blocking important work starting a thread in the lock.
+  void FinishStartingAdditionalThread(int thread_number);
+
+  // Checks whether there is work left that's blocking shutdown. Must be
+  // called inside the lock.
+  bool CanShutdown() 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 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.
+  Lock lock_;
+
+  // Condition variable used to wake up worker threads when a task is runnable.
+  ConditionVariable cond_var_;
+
+  // The maximum number of worker threads we'll create.
+  size_t max_threads_;
+
+  std::string thread_name_prefix_;
+
+  // 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_;
+
+  // Set to true when we're in the process of creating another thread.
+  // See PrepareToStartAdditionalThreadIfHelpful for more.
+  bool thread_being_created_;
+
+  // Number of threads currently waiting for work.
+  size_t waiting_thread_count_;
+
+  // Number of threads currently running tasks that have the BLOCK_SHUTDOWN
+  // flag set.
+  size_t blocking_shutdown_thread_count_;
+
+  // 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_;
+
+  // Number of tasks in the pending_tasks_ list that are marked as blocking
+  // shutdown.
+  size_t blocking_shutdown_pending_task_count_;
+
+  // Lists all sequence tokens currently executing.
+  std::set<int> current_sequences_;
+
+  // Set when Shutdown is called and no further tasks should be
+  // allowed, though we may still be running existing tasks.
+  bool shutdown_called_;
 
-  scoped_refptr<Inner> inner_;
+  TestingObserver* testing_observer_;
 
   DISALLOW_COPY_AND_ASSIGN(SequencedWorkerPool);
 };