Pass Callback to TaskRunner by value and consume it on invocation (1)

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 <stdint.h>
 
 #include <list>
 #include <map>
 #include <memory>
 #include <set>
 #include <unordered_map>
 #include <utility>
 #include <vector>
 
 #include "base/atomic_sequence_num.h"
 #include "base/callback.h"
+#include "base/callback_helpers.h"
 #include "base/compiler_specific.h"
 #include "base/critical_closure.h"
 #include "base/debug/dump_without_crashing.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/stl_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/synchronization/condition_variable.h"
@@ skipping 104 matching lines @@
 //
 // Note that this class is RefCountedThreadSafe (inherited from TaskRunner).
 class SequencedWorkerPoolTaskRunner : public TaskRunner {
  public:
   SequencedWorkerPoolTaskRunner(
       scoped_refptr<SequencedWorkerPool> pool,
       SequencedWorkerPool::WorkerShutdown shutdown_behavior);
 
   // TaskRunner implementation
   bool PostDelayedTask(const tracked_objects::Location& from_here,
-                       const Closure& task,
+                       Closure task,
                        TimeDelta delay) override;
   bool RunsTasksOnCurrentThread() const override;
 
  private:
   ~SequencedWorkerPoolTaskRunner() override;
 
   const scoped_refptr<SequencedWorkerPool> pool_;
 
   const SequencedWorkerPool::WorkerShutdown shutdown_behavior_;
 
   DISALLOW_COPY_AND_ASSIGN(SequencedWorkerPoolTaskRunner);
 };
 
 SequencedWorkerPoolTaskRunner::SequencedWorkerPoolTaskRunner(
     scoped_refptr<SequencedWorkerPool> pool,
     SequencedWorkerPool::WorkerShutdown shutdown_behavior)
     : pool_(std::move(pool)), shutdown_behavior_(shutdown_behavior) {}
 
 SequencedWorkerPoolTaskRunner::~SequencedWorkerPoolTaskRunner() {
 }
 
 bool SequencedWorkerPoolTaskRunner::PostDelayedTask(
     const tracked_objects::Location& from_here,
-    const Closure& task,
+    Closure task,
     TimeDelta delay) {
   if (delay.is_zero()) {
-    return pool_->PostWorkerTaskWithShutdownBehavior(
-        from_here, task, shutdown_behavior_);
+    return pool_->PostWorkerTaskWithShutdownBehavior(from_here, std::move(task),
+                                                     shutdown_behavior_);
   }
-  return pool_->PostDelayedWorkerTask(from_here, task, delay);
+  return pool_->PostDelayedWorkerTask(from_here, std::move(task), delay);
 }
 
 bool SequencedWorkerPoolTaskRunner::RunsTasksOnCurrentThread() const {
   return pool_->RunsTasksOnCurrentThread();
 }
 
 }  // namespace
 
 // SequencedWorkerPool::PoolSequencedTaskRunner ------------------------------
 // A SequencedTaskRunner which posts tasks to a SequencedWorkerPool with a
 // fixed sequence token.
 //
 // Note that this class is RefCountedThreadSafe (inherited from TaskRunner).
 class SequencedWorkerPool::PoolSequencedTaskRunner
     : public SequencedTaskRunner {
  public:
   PoolSequencedTaskRunner(
       scoped_refptr<SequencedWorkerPool> pool,
       SequencedWorkerPool::SequenceToken token,
       SequencedWorkerPool::WorkerShutdown shutdown_behavior);
 
   // TaskRunner implementation
   bool PostDelayedTask(const tracked_objects::Location& from_here,
-                       const Closure& task,
+                       Closure task,
                        TimeDelta delay) override;
   bool RunsTasksOnCurrentThread() const override;
 
   // SequencedTaskRunner implementation
   bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
-                                  const Closure& task,
+                                  Closure task,
                                   TimeDelta delay) override;
 
  private:
   ~PoolSequencedTaskRunner() override;
 
   const scoped_refptr<SequencedWorkerPool> pool_;
 
   const SequencedWorkerPool::SequenceToken token_;
 
   const SequencedWorkerPool::WorkerShutdown shutdown_behavior_;
 
   DISALLOW_COPY_AND_ASSIGN(PoolSequencedTaskRunner);
 };
 
 SequencedWorkerPool::PoolSequencedTaskRunner::
     PoolSequencedTaskRunner(
         scoped_refptr<SequencedWorkerPool> pool,
         SequencedWorkerPool::SequenceToken token,
         SequencedWorkerPool::WorkerShutdown shutdown_behavior)
     : pool_(std::move(pool)),
       token_(token),
       shutdown_behavior_(shutdown_behavior) {}
 
 SequencedWorkerPool::PoolSequencedTaskRunner::
     ~PoolSequencedTaskRunner() = default;
 
-bool SequencedWorkerPool::PoolSequencedTaskRunner::
-    PostDelayedTask(const tracked_objects::Location& from_here,
-                    const Closure& task,
-                    TimeDelta delay) {
+bool SequencedWorkerPool::PoolSequencedTaskRunner::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    Closure task,
+    TimeDelta delay) {
   if (delay.is_zero()) {
     return pool_->PostSequencedWorkerTaskWithShutdownBehavior(
-        token_, from_here, task, shutdown_behavior_);
+        token_, from_here, std::move(task), shutdown_behavior_);
   }
-  return pool_->PostDelayedSequencedWorkerTask(token_, from_here, task, delay);
+  return pool_->PostDelayedSequencedWorkerTask(token_, from_here,
+                                               std::move(task), delay);
 }
 
 bool SequencedWorkerPool::PoolSequencedTaskRunner::
     RunsTasksOnCurrentThread() const {
   return pool_->IsRunningSequenceOnCurrentThread(token_);
 }
 
-bool SequencedWorkerPool::PoolSequencedTaskRunner::
-    PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
-                               const Closure& task,
-                               TimeDelta delay) {
+bool SequencedWorkerPool::PoolSequencedTaskRunner::PostNonNestableDelayedTask(
+    const tracked_objects::Location& from_here,
+    Closure task,
+    TimeDelta delay) {
   // There's no way to run nested tasks, so simply forward to
   // PostDelayedTask.
-  return PostDelayedTask(from_here, task, delay);
+  return PostDelayedTask(from_here, std::move(task), delay);
 }
 
 // Worker ---------------------------------------------------------------------
 
 class SequencedWorkerPool::Worker : public SimpleThread {
  public:
   // Hold a (cyclic) ref to |worker_pool|, since we want to keep it
   // around as long as we are running.
   Worker(scoped_refptr<SequencedWorkerPool> worker_pool,
          int thread_number,
@@ skipping 78 matching lines @@
 
   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,
+                Closure task,
                 TimeDelta delay);
 
   bool RunsTasksOnCurrentThread() const;
 
   bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const;
 
   void CleanupForTesting();
 
   void SignalHasWorkForTesting();
 
@@ skipping 323 matching lines @@
 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) {
   AutoLock lock(lock_);
   return SequenceToken(LockedGetNamedTokenID(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,
+    Closure task,
     TimeDelta delay) {
   // TODO(fdoray): Uncomment this DCHECK. It is initially commented to avoid a
   // revert of the CL that adds debug::DumpWithoutCrashing() if it fails on the
   // waterfall. https://crbug.com/622400
   // DCHECK_NE(AllPoolsState::POST_TASK_DISABLED, g_all_pools_state);
   if (g_all_pools_state == AllPoolsState::POST_TASK_DISABLED)
     debug::DumpWithoutCrashing();
 
   DCHECK(delay.is_zero() || shutdown_behavior == SKIP_ON_SHUTDOWN);
   SequencedTask sequenced(from_here);
   sequenced.sequence_token_id = sequence_token.id_;
   sequenced.shutdown_behavior = shutdown_behavior;
   sequenced.posted_from = from_here;
-  sequenced.task =
-      shutdown_behavior == BLOCK_SHUTDOWN ?
-      base::MakeCriticalClosure(task) : task;
+  sequenced.task = shutdown_behavior == BLOCK_SHUTDOWN
+                       ? base::MakeCriticalClosure(std::move(task))
+                       : std::move(task);
   sequenced.time_to_run = TimeTicks::Now() + delay;
 
   int create_thread_id = 0;
   {
     AutoLock lock(lock_);
 
     if (shutdown_called_) {
       // Don't allow a new task to be posted if it doesn't block shutdown.
       if (shutdown_behavior != BLOCK_SHUTDOWN)
         return false;
@@ skipping 289 matching lines @@
 
           // Complete thread creation outside the lock if necessary.
           if (new_thread_id)
             FinishStartingAdditionalThread(new_thread_id);
 
           this_worker->set_running_task_info(
               SequenceToken(task.sequence_token_id), task.shutdown_behavior);
 
           tracked_objects::TaskStopwatch stopwatch;
           stopwatch.Start();
-          task.task.Run();
+          base::ResetAndReturn(&task.task).Run();

[gab, 2017/03/15 19:13:42]

What's the difference between this and std::move(task.task).Run()? The
documentation on base::ResetAndReturn doesn't make it obvious why this should be
preferred, should it? Should the documentation be updated?

[tzik, 2017/03/21 05:43:20]

std::move(cb).Run() for Callback and OnceCallback behaves differently until
https://codereview.chromium.org/2753953003/ lands.
It doesn't clear |cb| if it's Callback, and does clear it if it's OnceCallback.

[tzik, 2017/03/22 08:52:03]

And it landed. Updated the CL to replace ResetAndReturn with std::move.

           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.
-          task.task = Closure();
+          DCHECK(!task.task);
 
           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_);
               DeleteWithoutLock(&delete_these_outside_lock, this_worker);
             }
@@ skipping 482 matching lines @@
 }
 
 scoped_refptr<TaskRunner>
 SequencedWorkerPool::GetTaskRunnerWithShutdownBehavior(
     WorkerShutdown shutdown_behavior) {
   return new SequencedWorkerPoolTaskRunner(this, shutdown_behavior);
 }
 
 bool SequencedWorkerPool::PostWorkerTask(
     const tracked_objects::Location& from_here,
-    const Closure& task) {
-  return inner_->PostTask(NULL, SequenceToken(), BLOCK_SHUTDOWN,
-                          from_here, task, TimeDelta());
+    Closure task) {
+  return inner_->PostTask(NULL, SequenceToken(), BLOCK_SHUTDOWN, from_here,
+                          std::move(task), TimeDelta());
 }
 
 bool SequencedWorkerPool::PostDelayedWorkerTask(
     const tracked_objects::Location& from_here,
-    const Closure& task,
+    Closure task,
     TimeDelta delay) {
   WorkerShutdown shutdown_behavior =
       delay.is_zero() ? BLOCK_SHUTDOWN : SKIP_ON_SHUTDOWN;
-  return inner_->PostTask(NULL, SequenceToken(), shutdown_behavior,
-                          from_here, task, delay);
+  return inner_->PostTask(NULL, SequenceToken(), shutdown_behavior, from_here,
+                          std::move(task), delay);
 }
 
 bool SequencedWorkerPool::PostWorkerTaskWithShutdownBehavior(
     const tracked_objects::Location& from_here,
-    const Closure& task,
+    Closure task,
     WorkerShutdown shutdown_behavior) {
-  return inner_->PostTask(NULL, SequenceToken(), shutdown_behavior,
-                          from_here, task, TimeDelta());
+  return inner_->PostTask(NULL, SequenceToken(), shutdown_behavior, from_here,
+                          std::move(task), TimeDelta());
 }
 
 bool SequencedWorkerPool::PostSequencedWorkerTask(
     SequenceToken sequence_token,
     const tracked_objects::Location& from_here,
-    const Closure& task) {
-  return inner_->PostTask(NULL, sequence_token, BLOCK_SHUTDOWN,
-                          from_here, task, TimeDelta());
+    Closure task) {
+  return inner_->PostTask(NULL, sequence_token, BLOCK_SHUTDOWN, from_here,
+                          std::move(task), TimeDelta());
 }
 
 bool SequencedWorkerPool::PostDelayedSequencedWorkerTask(
     SequenceToken sequence_token,
     const tracked_objects::Location& from_here,
-    const Closure& task,
+    Closure task,
     TimeDelta delay) {
   WorkerShutdown shutdown_behavior =
       delay.is_zero() ? BLOCK_SHUTDOWN : SKIP_ON_SHUTDOWN;
-  return inner_->PostTask(NULL, sequence_token, shutdown_behavior,
-                          from_here, task, delay);
+  return inner_->PostTask(NULL, sequence_token, shutdown_behavior, from_here,
+                          std::move(task), delay);
 }
 
 bool SequencedWorkerPool::PostNamedSequencedWorkerTask(
     const std::string& token_name,
     const tracked_objects::Location& from_here,
-    const Closure& task) {
+    Closure task) {
   DCHECK(!token_name.empty());
   return inner_->PostTask(&token_name, SequenceToken(), BLOCK_SHUTDOWN,
-                          from_here, task, TimeDelta());
+                          from_here, std::move(task), TimeDelta());
 }
 
 bool SequencedWorkerPool::PostSequencedWorkerTaskWithShutdownBehavior(
     SequenceToken sequence_token,
     const tracked_objects::Location& from_here,
-    const Closure& task,
+    Closure task,
     WorkerShutdown shutdown_behavior) {
-  return inner_->PostTask(NULL, sequence_token, shutdown_behavior,
-                          from_here, task, TimeDelta());
+  return inner_->PostTask(NULL, sequence_token, shutdown_behavior, from_here,
+                          std::move(task), TimeDelta());
 }
 
 bool SequencedWorkerPool::PostDelayedTask(
     const tracked_objects::Location& from_here,
-    const Closure& task,
+    Closure task,
     TimeDelta delay) {
-  return PostDelayedWorkerTask(from_here, task, delay);
+  return PostDelayedWorkerTask(from_here, std::move(task), delay);
 }
 
 bool SequencedWorkerPool::RunsTasksOnCurrentThread() const {
   return inner_->RunsTasksOnCurrentThread();
 }
 
 void SequencedWorkerPool::FlushForTesting() {
   DCHECK(!RunsTasksOnCurrentThread());
   base::ThreadRestrictions::ScopedAllowWait allow_wait;
   if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
@@ skipping 16 matching lines @@
 bool SequencedWorkerPool::IsShutdownInProgress() {
   return inner_->IsShutdownInProgress();
 }
 
 bool SequencedWorkerPool::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   return inner_->IsRunningSequenceOnCurrentThread(sequence_token);
 }
 
 }  // namespace base