Implementation of SequencedTaskRunner based on SequencedWorkerPool.

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>
 #include <vector>
 
 #include "base/atomicops.h"
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/memory/linked_ptr.h"
 #include "base/message_loop_proxy.h"
 #include "base/metrics/histogram.h"
 #include "base/stl_util.h"
 #include "base/stringprintf.h"
 #include "base/synchronization/condition_variable.h"
 #include "base/synchronization/lock.h"
 #include "base/threading/platform_thread.h"
+#include "base/threading/sequenced_task_runner_impl.h"
 #include "base/threading/simple_thread.h"
 #include "base/time.h"
 #include "base/tracked_objects.h"
 
 namespace base {
 
 namespace {
 
 struct SequencedTask {
   SequencedTask()
@@ skipping 38 matching lines @@
   // by it).
   Inner(SequencedWorkerPool* worker_pool, size_t max_threads,
         const std::string& thread_name_prefix);
 
   ~Inner();
 
   SequenceToken GetSequenceToken();
 
   SequenceToken GetNamedSequenceToken(const std::string& name);
 
+  scoped_refptr<SequencedTaskRunner> GetSequencedTaskRunner();
+
+  scoped_refptr<SequencedTaskRunner> GetSequencedTaskRunner(

[akalin, 2012/03/15 19:17:35]

Overloads are against style guide -- I think there should only be one function
that takes a SequenceToken; users can either query for a new one or map from a
token_name if they need to.

[Francois, 2012/03/18 16:28:29]

Done.

+      SequenceToken sequence_token);
+
+  scoped_refptr<SequencedTaskRunner> GetSequencedTaskRunner(
+      const std::string& token_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);
 
   bool RunsTasksOnCurrentThread() const;
@@ skipping 13 matching lines @@
 
  private:
   // Returns whether there are no more pending tasks and all threads
   // are idle.  Must be called under lock.
   bool IsIdle() const;
 
   // 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);
 
+  // Called from within the lock, this returns a reference to a
+  // SequencedTaskRunner which wraps the SequencedWorkerPool object.
+  scoped_refptr<SequencedTaskRunner> LockedGetSequencedTaskRunner(
+      SequenceToken token);
+
   // 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.
@@ skipping 26 matching lines @@
 
   // Signal |has_work_| and increment |has_work_signal_count_|.
   void SignalHasWork();
 
   // Checks whether there is work left that's blocking shutdown. Must be
   // called inside the lock.
   bool CanShutdown() const;
 
   SequencedWorkerPool* const worker_pool_;
 
+  WeakPtr<SequencedTaskRunner> sequenced_task_runner_;

[akalin, 2012/03/15 19:17:35]

no need for this member variable.  Creating a new STRImpl is cheap enough that
there's no need to reuse one.

[Francois, 2012/03/18 16:28:29]

Do you mean that it should do "return scoped_refptr<STRImpl>(new STRImpl)"? If
so, I did consider it, but I reckoned that it would be pointless to return a
scoped_refptr to a new instance every time. I thought it would be better to
return a scoped_ptr (with a new instance) instead, but I got the impression that
it was quite important for STR to be ref-counted. This is why I went with a
persistent instance inside SWP, which would keep a weak pointer to it. But I
didn't realise that weak_ptrs are not thread-safe. So it will have to be one of
the other two options, but which one?

+
   // 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.
   mutable Lock lock_;
 
@@ skipping 130 matching lines @@
       subtle::NoBarrier_AtomicIncrement(&last_sequence_number_, 1);
   return SequenceToken(static_cast<int>(result));
 }
 
 SequencedWorkerPool::SequenceToken
 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) {
   AutoLock lock(lock_);
   return SequenceToken(LockedGetNamedTokenID(name));
 }
 
+scoped_refptr<SequencedTaskRunner>
+SequencedWorkerPool::Inner::GetSequencedTaskRunner() {
+  AutoLock lock(lock_);
+  return LockedGetSequencedTaskRunner(GetSequenceToken());
+}
+
+scoped_refptr<SequencedTaskRunner>
+SequencedWorkerPool::Inner::GetSequencedTaskRunner(SequenceToken token) {
+  AutoLock lock(lock_);

[akalin, 2012/03/15 19:17:35]

per the comments above, shouldn't need a lock for this

+  return LockedGetSequencedTaskRunner(token);
+}
+
+scoped_refptr<SequencedTaskRunner>
+SequencedWorkerPool::Inner::GetSequencedTaskRunner(
+    const std::string& token_name) {
+  AutoLock lock(lock_);
+  return LockedGetSequencedTaskRunner(
+      SequenceToken(LockedGetNamedTokenID(token_name)));
+}
+
 bool SequencedWorkerPool::Inner::PostTask(
     const std::string* optional_token_name,
     SequenceToken sequence_token,
     WorkerShutdown shutdown_behavior,
     const tracked_objects::Location& from_here,
     const Closure& task) {
   SequencedTask sequenced;
   sequenced.sequence_token_id = sequence_token.id_;
   sequenced.shutdown_behavior = shutdown_behavior;
   sequenced.location = from_here;
@@ skipping 167 matching lines @@
       named_sequence_tokens_.find(name);
   if (found != named_sequence_tokens_.end())
     return 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.id_;
 }
 
+scoped_refptr<SequencedTaskRunner>
+SequencedWorkerPool::Inner::LockedGetSequencedTaskRunner(SequenceToken token) {
+  if (!sequenced_task_runner_) {
+    scoped_refptr<SequencedTaskRunner> ref = new SequencedTaskRunnerImpl(
+        scoped_refptr<SequencedWorkerPool>(worker_pool_), token);
+    sequenced_task_runner_ =
+        static_cast<SequencedTaskRunnerImpl*>(ref.get())->AsWeakPtr();
+    return ref;
+  }
+  return scoped_refptr<SequencedTaskRunner>(sequenced_task_runner_.get());
+}
+
 bool SequencedWorkerPool::Inner::GetWork(
     SequencedTask* task,
     std::vector<Closure>* delete_these_outside_lock) {
   lock_.AssertAcquired();
 
   DCHECK_EQ(pending_tasks_.size(), pending_task_count_);
   UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount",
                            static_cast<int>(pending_task_count_));
 
   // Find the next task with a sequence token that's not currently in use.
@@ skipping 216 matching lines @@
 
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() {
   return inner_->GetSequenceToken();
 }
 
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken(
     const std::string& name) {
   return inner_->GetNamedSequenceToken(name);
 }
 
+scoped_refptr<SequencedTaskRunner> SequencedWorkerPool::GetSequencedTaskRunner(
+) {
+  return inner_->GetSequencedTaskRunner();
+}
+
+scoped_refptr<SequencedTaskRunner> SequencedWorkerPool::GetSequencedTaskRunner(
+    SequenceToken token) {
+  return inner_->GetSequencedTaskRunner(token);
+}
+
+scoped_refptr<SequencedTaskRunner> SequencedWorkerPool::GetSequencedTaskRunner(
+    const std::string& token_name) {
+  return inner_->GetSequencedTaskRunner(token_name);
+}
+
 bool SequencedWorkerPool::PostWorkerTask(
     const tracked_objects::Location& from_here,
     const Closure& task) {
   return inner_->PostTask(NULL, SequenceToken(), BLOCK_SHUTDOWN,
                           from_here, task);
 }
 
 bool SequencedWorkerPool::PostWorkerTaskWithShutdownBehavior(
     const tracked_objects::Location& from_here,
     const Closure& task,
@@ skipping 65 matching lines @@
 void SequencedWorkerPool::Shutdown() {
   DCHECK(constructor_message_loop_->BelongsToCurrentThread());
   inner_->Shutdown();
 }
 
 void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) {
   inner_->SetTestingObserver(observer);
 }
 
 }  // namespace base