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.
+          task.sequence_token_id = this_worker->task_sequence_token().id_;

[gab, 2015/11/18 22:37:33]

Why does this matter? Isn't the task's sequence_token_id only useful for it to
be picked up by |this_worker| in this method while the task is running (and
therefore irrelevant after)?

[Bernhard Bauer, 2015/11/19 10:15:12]

DidRunWorkerTask() below removes the sequence token ID from the set of currently
running sequences that is used to prevent multiple sequences from running at the
same time. Without this, the new sequence token ID would stay in that set, so
the sequence would never run anymore.

[gab, 2015/11/19 14:34:36]

I see, can you highlight that in the comment as well?

[Bernhard Bauer, 2015/11/19 17:01:55]

Done.

+
           // 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