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);
+
+  // Mark the sequence token as in use.
+  bool success = current_sequences_.insert(sequence_token.id_).second;
+  DCHECK(success);
+}
+
 // 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 104 matching lines @@
               SequenceToken(task.sequence_token_id), task.shutdown_behavior);
 
           tracked_objects::TaskStopwatch stopwatch;
           stopwatch.Start();
           task.task.Run();
           stopwatch.Stop();
 
           tracked_objects::ThreadData::TallyRunOnNamedThreadIfTracking(
               task, stopwatch);
 
+          // Update the sequence token in case it has been set from within the
+          // task, so it can be removed from the set of currently running
+          // sequences in DidRunWorkerTask() below.
+          task.sequence_token_id = this_worker->task_sequence_token().id_;

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

Is this line captured in any unit tests?

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

Yes, SequencedWorkerPoolTest.GetSequencedTaskRunnerForCurrentThread will hang
without this line, because the next task with this sequence token could never
run (as it's never removed from the set).

+
           // 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.
           task.task = Closure();
 
           this_worker->reset_running_task_info();
         }
         DidRunWorkerTask(task);  // Must be done inside the lock.
@@ skipping 354 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();
+  if (!sequence_token.IsValid()) {
+    sequence_token = Inner::GetSequenceToken();
+    pool->inner_->SetRunningTaskInfoForCurrentThread(sequence_token,
+                                                     shutdown_behavior);
+  }
+
+  DCHECK(pool->IsRunningSequenceOnCurrentThread(sequence_token));
+  return new SequencedWorkerPoolSequencedTaskRunner(
+      std::move(pool), sequence_token, 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