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"

[jar (doing other things), 2011/12/09 19:01:48]

nit: alphabetize

+#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();
+
+ 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);

[jar (doing other things), 2011/12/09 19:01:48]

nit: explicit for constructor

+  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:
+  // 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<base::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(Worker* worker, const SequencedTask& task);
+  void DidRunWorkerTask(Worker* worker, 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 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_;

[jar (doing other things), 2011/12/09 19:01:48]

Is this const from construction time?

+
+  // 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 running tasks.
+  size_t running_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 the app is terminating and no further tasks should be allowed,
+  // though we may still be running existing tasks.
+  bool terminating_;
+
+  // Set when Shutdown is called to do some assertions.
+  bool shutdown_called_;
+
+  SequencedWorkerPool::TestingObserver* testing_observer_;
+};
+
+SequencedWorkerPool::Worker::Worker(SequencedWorkerPool::Inner* inner,
+                                    int thread_number)
+    : base::SimpleThread(
+          StringPrintf("BrowserWorker%d", thread_number).c_str()),

[jar (doing other things), 2011/12/09 19:01:48]

Since this class appears to be designed so that we can have several distinct
pools, you should probably have the prefix-name (here shown as "BrowserWorker")
as a parameter to the construction, which is held in inner.

+      inner_(inner),
+      current_shutdown_mode_(SequencedWorkerPool::CONTINUE_ON_SHUTDOWN) {
+  Start();
+}
+
+SequencedWorkerPool::Worker::~Worker() {
+}
+
+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),
+      thread_being_created_(false),
+      running_thread_count_(0),
+      blocking_shutdown_thread_count_(0),
+      pending_task_count_(0),
+      blocking_shutdown_pending_task_count_(0),
+      terminating_(false),
+      shutdown_called_(false) {
+}
+
+SequencedWorkerPool::Inner::~Inner() {
+  // You must call Shutdown() before destroying the pool.
+  DCHECK(shutdown_called_);
+
+  // Need to explicitly join with the threads before they're destroyed or else
+  // they will be running when our object is half torn down.
+  for (size_t i = 0; i < threads_.size(); i++)
+    threads_[i]->Join();
+  threads_.clear();
+}
+
+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) {
+  int create_thread_id = 0;
+  {
+    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;

[jar (doing other things), 2011/12/09 19:01:48]

Although not a big deal, lines 252-256 should as a matter of standard practice,
be performed outside the lock.

+
+    pending_tasks_.push_back(sequenced);
+    pending_task_count_++;
+    if (shutdown_behavior == BLOCK_SHUTDOWN)
+      blocking_shutdown_pending_task_count_++;
+
+    create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
+  }
+
+  // Actually start the additional thread or signal an existing one now that
+  // we're outside the lock.
+  if (create_thread_id)
+    FinishStartingAdditionalThread(create_thread_id);
+  else
+    cond_var_.Signal();
+
+  return true;
+}
+
+void SequencedWorkerPool::Inner::Shutdown() {
+  if (shutdown_called_)
+    return;
+  shutdown_called_ = true;
+
+  // 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;
+
+    // Tickle the threads. This will wake up a waiting one so it will know that
+    // it can exit, which in turn will wake up any other waiting ones.
+    cond_var_.Signal();
+
+    // There are no pending or running tasks blocking shutdown, we're done.
+    if (CanShutdown())
+      return;
+  }
+
+  // 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();
+
+  // Wait for no more tasks.
+  {
+    base::AutoLock lock(lock_);
+    while (!CanShutdown())
+      cond_var_.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 (true) {
+      // See GetWork for what delete_these_outside_lock is doing.
+      SequencedTask task;
+      std::vector<base::Closure> delete_these_outside_lock;
+      if (GetWork(&task, &delete_these_outside_lock)) {
+        int new_thread_id = WillRunWorkerTask(this_worker, task);
+        {
+          base::AutoUnlock unlock(lock_);

[jar (doing other things), 2011/12/09 19:01:48]

This inner loop looks excellent.  I'd suggest doin a signal() just after the
unlock.

+          delete_these_outside_lock.clear();
+
+          // Complete thread creation outside the lock if necessary.
+          if (new_thread_id)
+            FinishStartingAdditionalThread(new_thread_id);
+
+          task.task.Run();
+
+          // Make sure our task is erased outside the lock for the same reason
+          // we do this with delete_these_oustide_lock.
+          task.task = base::Closure();

[jar (doing other things), 2011/12/09 19:01:48]

:-) Perfect.

+        }
+        DidRunWorkerTask(this_worker, task);  // Must be done inside the lock.

[jar (doing other things), 2011/12/09 19:01:48]

Perhaps I'm a little confused here... was the task we just ran deleted while the
lock was released (just after it was run)?
If so, all we need to preserve from line 332 is the two bits of state that can
be fed in here to DidRunWorkerTask() (re: was this a BLOCKING task, so we need
to fix the count, and what was the sequence number that we have to pull from the
set<>).

+      } else {
+        // When we're terminating and there's no more work, we can shut down.
+        // You can't get more tasks posted once terminating_ is set. There may
+        // be some tasks stuck behind running ones with the same sequence
+        // token, but additional threads won't help this case.
+        if (terminating_)
+          break;
+        cond_var_.Wait();
+      }
+    }
+  }
+
+  // We noticed we should exit. Wake up the next worker so it knows it should
+  // exit as well (because the Shutdown() code only signals once).
+  cond_var_.Signal();

[jar (doing other things), 2011/12/09 19:01:48]

If you add one more scope to hold the
AutoLock() on line 326, so you'd have a place to place the signal() outside the
lock.

[brettw, 2011/12/10 05:42:27]

This is already done the way you are describing.

[jar (doing other things), 2011/12/10 17:08:03]

AutoLock() is on line 326, which is the outermost scope of the function.  As a
result, this Signal() is still done while holding the lock.  

This is not critical to functional correctness, but at times it can cause what
is termed the "box-car performance problem."  The problem can involve a thread
waking up due to a Signal(), only to find that it can't acquire the lock, and
then going back to sleep waiting on the lock.

I think Clank manifests this problem, and often does a full-blown context switch
(on a single core CPU!) when a signal() is provided.  That is guaranteed to land
on the other thread with the lock already in use.  I *think* that was the most
recent horror story I heard about this.


On 2011/12/10 05:42:27, brettw wrote:

[brettw, 2011/12/10 17:25:28]

You are incorrect, the AutoLock is not at the outermost scope.

[jar (doing other things), 2011/12/10 18:41:43]

You are correct.  I was incorrect.  Sorry... I kept looking at the curly on the
first line of the function <doh>.

On 2011/12/10 17:25:28, brettw wrote:

+}
+
+bool SequencedWorkerPool::Inner::GetWork(
+    SequencedTask* task,
+    std::vector<base::Closure>* delete_these_outside_lock) {
+  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;
+  std::list<SequencedTask>::iterator i = pending_tasks_.begin();
+  while (i != pending_tasks_.end()) {
+    if (!IsSequenceTokenRunnable(i->sequence_token_id)) {
+      unrunnable_tasks++;
+      ++i;
+      continue;
+    }
+
+    if (terminating_ && i->shutdown_behavior != BLOCK_SHUTDOWN) {
+      // We're shutting down and the task we just found isn't blocking
+      // shutdown. Delete it and get more work.
+      //
+      // Note that we do not want to delete unrunnable tasks. Deleting a task
+      // can have side effects (like freeing some objects) and deleting a
+      // task that's supposed to run after one that's currently running could
+      // cause an obscure crash.
+      //
+      // We really want to delete these tasks outside the lock in case the
+      // closures are holding refs to objects that want to post work from
+      // their destructorss (which would deadlock). The closures are
+      // internally refcounted, so we just need to keep a copy of them alive
+      // until the lock is exited. The calling code can just clear() the
+      // vector they passed to us once the lock is exited to make this
+      // happen.
+      delete_these_outside_lock->push_back(i->task);
+      i = pending_tasks_.erase(i);
+      pending_task_count_--;
+    } else {
+      // Found a runnable task.
+      *task = *i;
+      delete_these_outside_lock->push_back(i->task);  // See above.

[jar (doing other things), 2011/12/09 19:01:48]

I *think* you take care of deleting this yourself separately.  In addition, if
you want to go this route, you should do the clear of delete_these_outside_lock
*after* you run the task around line 345..

...or am I confused?

+      i = pending_tasks_.erase(i);
+      pending_task_count_--;
+      if (task->shutdown_behavior == BLOCK_SHUTDOWN)
+        blocking_shutdown_pending_task_count_--;
+
+      found_task = true;
+      break;
+    }
+  }
+
+  // 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;
+}
+
+int SequencedWorkerPool::Inner::WillRunWorkerTask(Worker* worker,
+                                                  const SequencedTask& task) {
+  lock_.AssertAcquired();
+
+  // Mark the task's sequence number as in use.
+  if (task.sequence_token_id)
+    current_sequences_.insert(task.sequence_token_id);
+
+  running_thread_count_++;
+
+  if (task.shutdown_behavior == SequencedWorkerPool::BLOCK_SHUTDOWN)
+    blocking_shutdown_thread_count_++;
+
+  // We just picked up a task. Since StartAdditionalThreadIfHelpful only
+  // creates a new thread if there is no free one, there is a race when posting
+  // tasks that many tasks could have been posted before a thread started
+  // running them, so only one thread would have been created. So we also check
+  // for more threads after removing our task from the queue, which also has
+  // the nice side effect of creating the workers from background threads
+  // rather than the main thread of the app.
+  //
+  // If another thread wasn't created, we want to wake up an existing thread
+  // if there is one waiting to pick up the next task.
+  int create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
+  if (!create_thread_id && !pending_tasks_.empty() &&
+      running_thread_count_ < threads_.size())
+    cond_var_.Signal();

[jar (doing other things), 2011/12/09 19:01:48]

As mentioned above... is is simpler to always signal when you release the lock
to do some work... so you won't need to signal here.  You *could* be more
precise about the signalling.... but it is much easier to see everything is
perfect if you don't bother (and you avoid checking for "more work" in one
thread, and checking for "some work to do" in the signalled thread).

+
+  return create_thread_id;
+}
+
+void SequencedWorkerPool::Inner::DidRunWorkerTask(Worker* worker,
+                                                  const SequencedTask& task) {
+  lock_.AssertAcquired();
+
+  if (task.shutdown_behavior == SequencedWorkerPool::BLOCK_SHUTDOWN) {
+    DCHECK(blocking_shutdown_thread_count_ > 0);

[jar (doing other things), 2011/12/09 19:01:48]

nit: DCHECK_GT

+    blocking_shutdown_thread_count_--;
+  }
+
+  if (task.sequence_token_id)
+    current_sequences_.erase(task.sequence_token_id);
+
+  running_thread_count_--;
+}
+
+bool SequencedWorkerPool::Inner::IsSequenceTokenRunnable(
+    int sequence_token_id) const {
+  lock_.AssertAcquired();
+  return !sequence_token_id ||
+      current_sequences_.find(sequence_token_id) ==
+          current_sequences_.end();
+}
+
+int SequencedWorkerPool::Inner::PrepareToStartAdditionalThreadIfHelpful() {
+  // How thread creation works:
+  //
+  // We'de like to avoid creating threads with the lock held. However, we
+  // need to be sure that we have an accurate accounting of the threads for
+  // proper Joining and deltion on shutdown.
+  //
+  // We need to figure out if we need another thread with the lock held, which
+  // is what this function does. It then marks us as in the process of creating
+  // a thread. When we do shutdown, we wait until the thread_being_created_
+  // flag is cleared, which ensures that the new thread is properly added to
+  // all the data structures and we can't leak it. Once shutdown starts, we'll
+  // refuse to create more threads or they would be leaked.

[jar (doing other things), 2011/12/09 19:01:48]

IMO, when you do the preparation for starting a new thread, you should update
(here and now) all the counts you care about.  That way, you won't have to wait
for the thread to really start and do the additions to the state.

+  if (!terminating_ &&
+      !thread_being_created_ &&
+      threads_.size() < max_threads_ &&
+      running_thread_count_ == threads_.size()) {
+    // We could use an additional thread if there's work to be done.
+    for (std::list<SequencedTask>::iterator i = pending_tasks_.begin();
+         i != pending_tasks_.end(); ++i) {
+      if (terminating_ && i->shutdown_behavior != BLOCK_SHUTDOWN)
+        continue;  // Skip non-critical tasks on shutdown.
+      if (IsSequenceTokenRunnable(i->sequence_token_id)) {
+        // Found a runnable task, mark the thread as being started.
+        thread_being_created_ = true;

[jar (doing other things), 2011/12/09 19:01:48]

This is a nice flag.... as it can keep us from starting two thread when one will
suffice. 

+        return threads_.size() + 1;

[jar (doing other things), 2011/12/09 19:01:48]

This is where you should do all the state changes to account for the new
addition of a thread.  You should effectively account for the thread as though
it is asleep (waiting).  IT will soon startup, and will go through all the
checking that a thread would do when it wakes up.

[brettw, 2011/12/10 05:42:27]

It's not clear what you want me to move here.

[jar (doing other things), 2011/12/10 17:08:03]

IMO, the number of running (including to-be-created) threads should be a state
variable, and that is what I think should be updated here.  I applogize for
having that as my mental model, which is not the current code, making my
suggestion very confusing.

Currently, you rely on threads_.size() to supply this count.  That vector size
will only be correctly updated after the new thread has started, and has updated
the vector with a push().  If instead you used a state variable to keep track of
the count of threads, then you can update it here (showing intent to start a new
thread) and hence have it correctly set to while you still hold the lock.

That change would remove the concern about leaking expressed in line 512.  If we
don't want to leak, then we wait until the count of running threads goes to
zero, and that count is not based on the vector size.

+      }
+    }
+  }
+  return 0;
+}
+
+void SequencedWorkerPool::Inner::FinishStartingAdditionalThread(
+    int thread_number) {
+  // Called outside of the lock.
+  DCHECK(thread_number > 0);
+  linked_ptr<Worker> new_thread(new Worker(this, thread_number));

[willchan no longer on Chromium, 2011/12/09 21:41:47]

linked_ptr isn't threadsafe...is this code safe? Is it possible another thread
the linked_ptr from |threads_| at the same time |new_thread| is getting
destroyed in this thread? I think that would lead to concurrent modification of
a thread-unsafe data structure.

[jar (doing other things), 2011/12/09 22:13:58]

I thought the access to this ref-counted data structure was only done under the
protection of a lock.  You are looking at the creation of the pointer with
associated structure, prior to making it available.   No one else has access to
it until it is pushed into the list of threads (inside the lock).

The one place this is later manipulated is when the vector of threads is
deleted, by calling threads_.clean().  I believe that is done outside a lock,
and might be a problem (as you suspected).

The vector is only kept around to be able to do the join... and I've argued that
the join is not needed (we should just look for the state to indicate that all
critical threads are complete.

There will still be a question of how to delete the |inner| instance.... but I
think that is solvable.

On 2011/12/09 21:41:47, willchan wrote:

[brettw, 2011/12/10 05:42:27]

Will is correct, this is bad.
As I explained before, the requirement that the tread be joined is enforced by
the tread code in base. I didn't choose to add this. I think the existing
semantics of SimpleThread are outside the scope of this review.

[jar (doing other things), 2011/12/10 17:08:03]

IMO, the would-be-reusable code of SimpleThread is not as reusable as it could
be, due to this requirement.  

The requirement should be removed (if it is really not critical to avoiding
abuse), or the class should be modified to make it more reusable.  We could, for
instance, have a setter (or a constuctor arg) that turns off this CHECK().

On 2011/12/10 05:42:27, brettw wrote:

+
+  {
+    base::AutoLock lock(lock_);
+
+    DCHECK(thread_being_created_);
+    thread_being_created_ = false;
+
+    threads_.push_back(new_thread);
+  }
+
+  // Clearing the thread_being_created_ flag could be the last thing the
+  // shutdown code was waiting for, so we need to wake it up.
+  cond_var_.Signal();
+}
+
+bool SequencedWorkerPool::Inner::CanShutdown() const {
+  lock_.AssertAcquired();
+  // See PrepareToStartAdditionalThreadIfHelpful for how thread creation works.
+  return !thread_being_created_ &&
+         blocking_shutdown_thread_count_ == 0 &&
+         blocking_shutdown_pending_task_count_ == 0;
+}
+
+// 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);
+}