WILL BE MERGED Change Ownership of Sequence to the Single Thread SchedulerWorker Delegate

base/task_scheduler/scheduler_single_thread_task_runner_manager.cc

 // Copyright 2017 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/task_scheduler/scheduler_single_thread_task_runner_manager.h"
 
 #include <algorithm>
 #include <memory>
 #include <string>
 
@@ skipping 46 matching lines @@
       : thread_name_(thread_name) {}
 
   // SchedulerWorker::Delegate:
   void OnMainEntry(SchedulerWorker* worker) override {
     thread_ref_checker_.Set();
     PlatformThread::SetName(thread_name_);
   }
 
   scoped_refptr<Sequence> GetWork(SchedulerWorker* worker) override {
     AutoSchedulerLock auto_lock(sequence_lock_);
-    return std::move(sequence_);
+    bool has_work = has_work_;
+    has_work_ = false;
+    return has_work ? sequence_ : nullptr;
   }
 
   void DidRunTask() override {}
 
   void ReEnqueueSequence(scoped_refptr<Sequence> sequence) override {
     AutoSchedulerLock auto_lock(sequence_lock_);
-    DCHECK(!sequence_);
-    sequence_ = std::move(sequence);
+    DCHECK_EQ(sequence, sequence_);
+    has_work_ = true;
   }
 
   TimeDelta GetSleepTimeout() override { return TimeDelta::Max(); }
 
   bool CanDetach(SchedulerWorker* worker) override { return false; }
 
   void OnDetach() override { NOTREACHED(); }
 
   bool RunsTasksOnCurrentThread() {
     // We check the thread ref instead of the sequence for the benefit of COM
     // callbacks which may execute without a sequence context.
     return thread_ref_checker_.IsCurrentThreadSameAsSetThread();
   }
 
+  void OnMainExit() override {
+    // To reclaim skipped tasks on shutdown, we null out the sequence to allow
+    // the tasks to destroy themselves.

[fdoray, 2017/03/02 14:37:48]

DCHECK(sequence->HasOneRef());

[robliao, 2017/03/03 04:24:17]

In general for refcounted things, it's a dangerous practice to assert OneRef
remaining at a certain point.

If one ref is remaining and we fail to cleanup, the ConcurrentJoin tests will
fail. If someone in the future legitimately needs a ref to survive past the
delegate and they do cleanup correctly, they'll DCHECK here.

+    sequence_ = nullptr;
+  }
+
+  // SchedulerWorkerDelegate:
+  Sequence* sequence() { return sequence_.get(); }
+
  private:
   const std::string thread_name_;
 
-  // Synchronizes access to |sequence_| and handles the fact that
-  // ReEnqueueSequence() is called on both the worker thread for reenqueuing
-  // the sequence and off of the worker thread to seed the sequence for
-  // GetWork().
+  // Synchronizes access to |sequence_| and |has_work_|.
   SchedulerLock sequence_lock_;
-  scoped_refptr<Sequence> sequence_;
+  scoped_refptr<Sequence> sequence_ = new Sequence;
+  bool has_work_ = false;
 
   AtomicThreadRefChecker thread_ref_checker_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerDelegate);
 };
 
 }  // namespace
 
 class SchedulerSingleThreadTaskRunnerManager::SchedulerSingleThreadTaskRunner
     : public SingleThreadTaskRunner {
  public:
   // Constructs a SchedulerSingleThreadTaskRunner that indirectly controls the
   // lifetime of a dedicated |worker| for |traits|.
   SchedulerSingleThreadTaskRunner(
       SchedulerSingleThreadTaskRunnerManager* const outer,
       const TaskTraits& traits,
       SchedulerWorker* worker)
       : outer_(outer), traits_(traits), worker_(worker) {
     DCHECK(outer_);
     DCHECK(worker_);
   }
 
   // SingleThreadTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
-                       TimeDelta delay) override;
+                       TimeDelta delay) override {
+    auto task = MakeUnique<Task>(from_here, closure, traits_, delay);
+    task->single_thread_task_runner_ref = this;
+
+    if (!outer_->task_tracker_->WillPostTask(task.get()))
+      return false;
+
+    if (task->delayed_run_time.is_null()) {
+      PostTaskNow(std::move(task));
+    } else {
+      outer_->delayed_task_manager_->AddDelayedTask(
+          std::move(task), Bind(&SchedulerSingleThreadTaskRunner::PostTaskNow,
+                                Unretained(this)));
+    }
+    return true;
+  }
 
   bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
                                   const Closure& closure,
                                   base::TimeDelta delay) override {
     // Tasks are never nested within the task scheduler.
     return PostDelayedTask(from_here, closure, delay);
   }
 
   bool RunsTasksOnCurrentThread() const override {
-    auto* delegate = static_cast<SchedulerWorkerDelegate*>(worker_->delegate());
-    return delegate->RunsTasksOnCurrentThread();
+    return GetDelegate()->RunsTasksOnCurrentThread();
   }
 
  private:
   ~SchedulerSingleThreadTaskRunner() override {
     outer_->UnregisterSchedulerWorker(worker_);
   }
 
-  void PostTaskNow(std::unique_ptr<Task> task);
+  void PostTaskNow(std::unique_ptr<Task> task) {
+    Sequence* sequence = GetDelegate()->sequence();

[fdoray, 2017/03/02 14:37:48]

When shutdown completes, the thread exits and the sequence is destroyed. If a
task is posted after that, this will crash. I don't think this is right. Instead
of dropping the reference to the Sequence in OnMainExit(), you could Terminate()
the Sequence (destroy all its tasks and prevent new tasks from being added).

[robliao, 2017/03/03 04:24:17]

Handled by making this a scoped_refptr. At this point, the caller of the
PostTask will clean up the task (which if there are thread checkers, is probably
a violation of the Shutdown + PostTask contract by the caller).

+    const bool sequence_was_empty = sequence->PushTask(std::move(task));
+    if (sequence_was_empty) {
+      GetDelegate()->ReEnqueueSequence(sequence);
+      worker_->WakeUp();
+    }
+  }
 
-  // Sequence for all Tasks posted through this TaskRunner.
-  const scoped_refptr<Sequence> sequence_ = new Sequence;
+  SchedulerWorkerDelegate* GetDelegate() const {
+    return static_cast<SchedulerWorkerDelegate*>(worker_->delegate());
+  }
 
   SchedulerSingleThreadTaskRunnerManager* const outer_;
   const TaskTraits traits_;
   SchedulerWorker* const worker_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
 };
 
-bool SchedulerSingleThreadTaskRunnerManager::SchedulerSingleThreadTaskRunner::
-    PostDelayedTask(const tracked_objects::Location& from_here,
-                    const Closure& closure,
-                    TimeDelta delay) {
-  auto task = MakeUnique<Task>(from_here, closure, traits_, delay);
-  task->single_thread_task_runner_ref = this;
-
-  if (!outer_->task_tracker_->WillPostTask(task.get()))
-    return false;
-
-  if (task->delayed_run_time.is_null()) {
-    PostTaskNow(std::move(task));
-  } else {
-    outer_->delayed_task_manager_->AddDelayedTask(
-        std::move(task),
-        Bind(&SchedulerSingleThreadTaskRunner::PostTaskNow, Unretained(this)));
-  }
-  return true;
-}
-
-void SchedulerSingleThreadTaskRunnerManager::SchedulerSingleThreadTaskRunner::
-    PostTaskNow(std::unique_ptr<Task> task) {
-  const bool sequence_was_empty = sequence_->PushTask(std::move(task));
-  if (sequence_was_empty) {
-    auto* delegate = static_cast<SchedulerWorkerDelegate*>(worker_->delegate());
-    delegate->ReEnqueueSequence(sequence_);
-    worker_->WakeUp();
-  }
-}
-
 SchedulerSingleThreadTaskRunnerManager::SchedulerSingleThreadTaskRunnerManager(
     const std::vector<SchedulerWorkerPoolParams>& worker_pool_params_vector,
     const TaskScheduler::WorkerPoolIndexForTraitsCallback&
         worker_pool_index_for_traits_callback,
     TaskTracker* task_tracker,
     DelayedTaskManager* delayed_task_manager)
     : worker_pool_params_vector_(worker_pool_params_vector),
       worker_pool_index_for_traits_callback_(
           worker_pool_index_for_traits_callback),
       task_tracker_(task_tracker),
       delayed_task_manager_(delayed_task_manager) {
   DCHECK_GT(worker_pool_params_vector_.size(), 0U);
   DCHECK(worker_pool_index_for_traits_callback_);
   DCHECK(task_tracker_);
   DCHECK(delayed_task_manager_);
 }
 
 SchedulerSingleThreadTaskRunnerManager::
     ~SchedulerSingleThreadTaskRunnerManager() {
-  DCHECK(workers_.empty()) << "SchedulerSingleThreadTaskRunners must outlive "
-                              "SchedulerSingleThreadTaskRunnerManager";
+  size_t workers_unregistered_during_join =
+      subtle::NoBarrier_Load(&workers_unregistered_during_join_);
+  DCHECK_EQ(workers_unregistered_during_join, workers_.size())
+      << "SchedulerSingleThreadTaskRunners must outlive "

[fdoray, 2017/03/02 14:37:48]

I think this would be clearer from a user point of view:
"All references to SingleThreadTaskRunners should be dropped before deleting
TaskScheduler."

[robliao, 2017/03/03 04:24:17]

While that's true, it doesn't make sense in the context of the unit tests
strangely enough since we don't have a TaskScheduler.

I've added the note in to cover both cases.

+         "SchedulerSingleThreadTaskRunnerManager";
 }
 
 scoped_refptr<SingleThreadTaskRunner>
 SchedulerSingleThreadTaskRunnerManager::CreateSingleThreadTaskRunnerWithTraits(
     const TaskTraits& traits) {
   size_t index = worker_pool_index_for_traits_callback_.Run(traits);
   DCHECK_LT(index, worker_pool_params_vector_.size());
   return new SchedulerSingleThreadTaskRunner(
       this, traits,
       CreateAndRegisterSchedulerWorker(worker_pool_params_vector_[index]));
@@ skipping 31 matching lines @@
 }
 
 void SchedulerSingleThreadTaskRunnerManager::UnregisterSchedulerWorker(
     SchedulerWorker* worker) {
   // Cleanup uses a SchedulerLock, so call Cleanup() after releasing
   // |workers_lock_|.
   scoped_refptr<SchedulerWorker> worker_to_destroy;
   {
     AutoSchedulerLock auto_lock(workers_lock_);
 
-    // We might be joining, so no-op this if |workers_| is empty.
-    if (workers_.empty())
+    // We might be joining, so record that a worker was unregistered for
+    // verification at destruction.
+    if (workers_.empty()) {
+      subtle::NoBarrier_AtomicIncrement(&workers_unregistered_during_join_, 1);

[fdoray, 2017/03/02 14:37:48]

Use base::AtomicRefCountInc and base::AtomicRefCountIsZero.

[robliao, 2017/03/03 04:24:17]

Those functions are intended for reference counting.
The author(s) don't seem to want people to directly call these functions:
https://cs.chromium.org/chromium/src/base/atomic_ref_count.h?rcl=27c78942d0fb...

[fdoray, 2017/03/03 05:31:13]

They want people to use base/memory/ref_counted.h instead of
base/atomic_ref_count.h, but they don't say anything about using
base/atomic_ref_counter.h instead of base/atomicops.h. You wouldn't be the first
user of this header. I'll let you decide what's best.

[fdoray, 2017/03/03 07:22:39]

ping

[robliao, 2017/03/03 07:50:05]

The intent of atomic_ref_count.h as written is to handle reference counting. We
aren't quite doing that here. We just need a counter.
I would be more okay using AtomicRefCountInc if it were instead AtomicIncrement
(no refcount indication).

Having said that, it might be an interesting refactor of atomic_ref_count ->
atomic_count

       return;
+    }
 
     auto worker_iter =
         std::find_if(workers_.begin(), workers_.end(),
                      [worker](const scoped_refptr<SchedulerWorker>& candidate) {
                        return candidate.get() == worker;
                      });
     DCHECK(worker_iter != workers_.end());
     worker_to_destroy = std::move(*worker_iter);
     workers_.erase(worker_iter);
   }
   worker_to_destroy->Cleanup();
 }
 
 }  // namespace internal
 }  // namespace base