Allow SequencedTaskRunnerHandle::Get() while running unsequenced tasks.

base/threading/sequenced_worker_pool.cc

 // 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.
 
 #include "base/threading/sequenced_worker_pool.h"
 
 #include <list>
 #include <map>
 #include <set>
 #include <utility>
@@ skipping 278 matching lines @@
 class SequencedWorkerPool::Inner {
  public:
   // Take a raw pointer to |worker| to avoid cycles (since we're owned
   // by it).
   Inner(SequencedWorkerPool* worker_pool, size_t max_threads,
         const std::string& thread_name_prefix,
         TestingObserver* observer);
 
   ~Inner();
 
-  SequenceToken GetSequenceToken();
+  static SequenceToken GetSequenceToken();
 
   SequenceToken GetNamedSequenceToken(const std::string& name);
 
   // 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 PostTask(const std::string* optional_token_name,
                 SequenceToken sequence_token,
                 WorkerShutdown shutdown_behavior,
                 const tracked_objects::Location& from_here,
                 const Closure& task,
                 TimeDelta delay);
 
   bool RunsTasksOnCurrentThread() const;
 
   bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const;
 
+  void SetRunningTaskInfoForCurrentThread(SequenceToken sequence_token,
+                                          WorkerShutdown shutdown_behavior);
+
   void CleanupForTesting();
 
   void SignalHasWorkForTesting();
 
   int GetWorkSignalCountForTesting() const;
 
   void Shutdown(int max_blocking_tasks_after_shutdown);
 
   bool IsShutdownInProgress();
 
@@ skipping 249 matching lines @@
   // 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 (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it)
     it->second->Join();
   threads_.clear();
 
   if (testing_observer_)
     testing_observer_->OnDestruct();
 }
 
+// static
 SequencedWorkerPool::SequenceToken
 SequencedWorkerPool::Inner::GetSequenceToken() {
   // Need to add one because StaticAtomicSequenceNumber starts at zero, which
   // is used as a sentinel value in SequenceTokens.
   return SequenceToken(g_last_sequence_number_.GetNext() + 1);
 }
 
 SequencedWorkerPool::SequenceToken
 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) {
   AutoLock lock(lock_);
@@ skipping 80 matching lines @@
 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   AutoLock lock(lock_);
   ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId());
   if (found == threads_.end())
     return false;
   return found->second->is_processing_task() &&
          sequence_token.Equals(found->second->task_sequence_token());
 }
 
+void SequencedWorkerPool::Inner::SetRunningTaskInfoForCurrentThread(
+    SequenceToken sequence_token,
+    WorkerShutdown shutdown_behavior) {
+  AutoLock lock(lock_);
+  ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId());
+  DCHECK(found != threads_.end());
+  DCHECK(found->second->is_processing_task());
+  DCHECK(!found->second->task_sequence_token().IsValid());
+  found->second->set_running_task_info(sequence_token, shutdown_behavior);
+}
+
 // See https://code.google.com/p/chromium/issues/detail?id=168415
 void SequencedWorkerPool::Inner::CleanupForTesting() {
   DCHECK(!RunsTasksOnCurrentThread());
   base::ThreadRestrictions::ScopedAllowWait allow_wait;
   AutoLock lock(lock_);
   CHECK_EQ(CLEANUP_DONE, cleanup_state_);
   if (shutdown_called_)
     return;
   if (pending_tasks_.empty() && waiting_thread_count_ == threads_.size())
     return;
@@ skipping 108 matching lines @@
           task.task.Run();
           stopwatch.Stop();
 
           tracked_objects::ThreadData::TallyRunOnNamedThreadIfTracking(
               task, stopwatch);
 
           // Make sure our task is erased outside the lock for the
           // same reason we do this with delete_these_oustide_lock.
           // Also, do it before calling reset_running_task_info() so
           // that sequence-checking from within the task's destructor
           // still works.

[danakj, 2015/11/17 22:03:39]

The task.sequence_token_id is no longer correct after calling run here
sometimes? What kind of problems might that make? Does that break this comment?
The SequencedTask destructor is empty, what checking is it talking about?

[Bernhard Bauer, 2015/11/18 13:15:50]

I think this is referring to the fact that we might drop the last reference to
the previous task (as in Closure, not SequencedTask), which would then destroy
it and its parameters, so if any of those would have a sequence check in its
destructor, it would continue to work.

[danakj, 2015/11/20 19:18:43]

Can you add a unit test that uses SequenceChecker in its destructor and is
posted from an unsequenced task? We should maintain this if people change this
class or SequenceChecker.

I'm trying to see why it would work, it looks like because it would have
previously seen there was no sequence, so the check in the destructor would use
thread_checker_ instead of comparing sequence_token_. And because we destroy the
closure after running the task on the same thread, the thread checker should
pass in the same circumstances it did before this patch. Is that what you see as
well?

[Bernhard Bauer, 2015/12/10 16:09:58]

Done! I verified that the test fails if the task info is reset before.
Yup. A SequenceChecker will fail on an unsequenced task, so changing it so be
sequenced will not break anything. It also won't cause false negatives, because
the sequence will be brand new, so a previously existing SequenceChecker can't
have used it.

           task.task = Closure();
 
           this_worker->reset_running_task_info();
         }
         DidRunWorkerTask(task);  // Must be done inside the lock.
       } else if (cleanup_state_ == CLEANUP_RUNNING) {
         switch (status) {
           case GET_WORK_WAIT: {
               AutoUnlock unlock(lock_);
               delete_these_outside_lock.clear();
@@ skipping 349 matching lines @@
 SequencedWorkerPool::SequenceToken
 SequencedWorkerPool::GetSequenceTokenForCurrentThread() {
   Worker* worker = Worker::GetForCurrentThread();
   if (!worker)
     return SequenceToken();
 
   return worker->task_sequence_token();
 }
 
 // static
-scoped_refptr<SequencedWorkerPool>
-SequencedWorkerPool::GetWorkerPoolForCurrentThread() {
+scoped_refptr<SequencedTaskRunner>
+SequencedWorkerPool::GetSequencedTaskRunnerForCurrentThread() {
   Worker* worker = Worker::GetForCurrentThread();
   if (!worker)
     return nullptr;
 
-  return worker->worker_pool();
+  scoped_refptr<SequencedWorkerPool> pool = worker->worker_pool();
+  SequenceToken sequence_token = worker->task_sequence_token();
+  WorkerShutdown shutdown_behavior = worker->task_shutdown_behavior();

[danakj, 2015/11/17 22:03:39]

Can you DCHECK that is_processing_task() is true?

[Bernhard Bauer, 2015/11/18 13:15:50]

task_shutdown_behavior() already does that. Alternatively, we could put the
sequence token and shutdown behavior into a struct and keep a scoped_ptr to it
that we reset in reset_running_task_info()?

[danakj, 2015/11/20 19:18:43]

Can you add a comment at the !worker early out explaining why if worker exists,
we know we're inside/processing a sequenced task? (Cuz worker is only non-null
inside WorkerPool::Run(), which will run sequenced tasks and does not call
GetSequencedTaskRunnerForCurrentThread itself?)

[Bernhard Bauer, 2015/12/10 16:09:58]

Done, modulo s/sequenced //, because a worker thread will run both sequenced and
unsequenced tasks. If it is running a sequenced task, we can just return a
SequencedWorkerPoolSequencedTaskRunner for the current sequence token; if it is
running an unsequenced task, we have to do the dance below.

+  if (!sequence_token.IsValid()) {
+    sequence_token = Inner::GetSequenceToken();
+    pool->inner_->SetRunningTaskInfoForCurrentThread(sequence_token,

[danakj, 2015/11/17 22:03:39]

Why do you do this? Now the thread is left in a state where it's bound to this
sequence after calling this. Normally we reset it to no sequence when you finish
a task. What will clean that up? What is going to use what you're setting here?
You're not going to Run() a task here.

[Bernhard Bauer, 2015/11/18 13:15:50]

I do this to make sure that a task posted to the new SequencedTaskRunner will
not start running immediately on another thread... but now that you mention it,
I did forget to mark the sequence token as in use, so I fixed that. We still
need to set these so that calling this twice in a row will return the same
sequence.

Once the task is done, the task info will be reset (the old sequence token will
stay around, but |is_processing_task_| will be false, which guards all accesses;
see also my comment above).

[danakj, 2015/11/20 19:18:43]

Can you encode this into a comment in here?

And can you add a unit test that would fail without your fix mentioned here?

[Bernhard Bauer, 2015/12/10 16:09:58]

Done.
I tried that, but it's tricky, because I can't really verify that a task posted
to the new SequencedTaskRunner will *not* immediately start running -- I can
fail if it does, but if the tasks happen to run in sequence, it doesn't
necessarily mean that wasn't just a coincidence.

I could post a task to the new SequencedTaskRunner and verify that it tries to
signal an existing thread that there is work instead of starting a new thread,
which indicates that the task isn't runnable yet, but that also seems like it
relies on a bunch of implementation details of the SWP... Do you have any better
ideas?

[danakj, 2015/12/16 21:53:32]

What about just calling it twice and making sure you get the same sequence token
and that IsSequenceTokenRunnable() is true?

+                                                     shutdown_behavior);
+  }
+
+  DCHECK(pool->IsRunningSequenceOnCurrentThread(sequence_token));
+  return new SequencedWorkerPoolSequencedTaskRunner(pool, sequence_token,

[danakj, 2015/11/17 22:03:39]

can you move() the pool smart pointer? no need to keep a ref here.

[Bernhard Bauer, 2015/11/18 13:15:50]

Done.

+                                                    shutdown_behavior);
 }
 
 SequencedWorkerPool::SequencedWorkerPool(size_t max_threads,
                                          const std::string& thread_name_prefix)
     : constructor_task_runner_(ThreadTaskRunnerHandle::Get()),
       inner_(new Inner(this, max_threads, thread_name_prefix, NULL)) {
 }
 
 SequencedWorkerPool::SequencedWorkerPool(size_t max_threads,
                                          const std::string& thread_name_prefix,
                                          TestingObserver* observer)
     : constructor_task_runner_(ThreadTaskRunnerHandle::Get()),
       inner_(new Inner(this, max_threads, thread_name_prefix, observer)) {
 }
 
 SequencedWorkerPool::~SequencedWorkerPool() {}
 
 void SequencedWorkerPool::OnDestruct() const {
-  // Avoid deleting ourselves on a worker thread (which would
-  // deadlock).
+  // Avoid deleting ourselves on a worker thread (which would deadlock).
   if (RunsTasksOnCurrentThread()) {
     constructor_task_runner_->DeleteSoon(FROM_HERE, this);
   } else {
     delete this;
   }
 }
 
+// static
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() {
-  return inner_->GetSequenceToken();
+  return Inner::GetSequenceToken();
 }
 
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken(
     const std::string& name) {
   return inner_->GetNamedSequenceToken(name);
 }
 
 scoped_refptr<SequencedTaskRunner> SequencedWorkerPool::GetSequencedTaskRunner(
     SequenceToken token) {
   return GetSequencedTaskRunnerWithShutdownBehavior(token, BLOCK_SHUTDOWN);
@@ skipping 101 matching lines @@
 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) {
   DCHECK(constructor_task_runner_->BelongsToCurrentThread());
   inner_->Shutdown(max_new_blocking_tasks_after_shutdown);
 }
 
 bool SequencedWorkerPool::IsShutdownInProgress() {
   return inner_->IsShutdownInProgress();
 }
 
 }  // namespace base