Make base::Timer sequence-friendly.

base/timer/timer.cc

Index: base/timer/timer.cc
diff --git a/base/timer/timer.cc b/base/timer/timer.cc
index 6ec18f181488c3a5b533abe93cf9b793914e6451..0efe8d984fe29ed61bca0502b08c1402e79cb8ba 100644
--- a/base/timer/timer.cc
+++ b/base/timer/timer.cc
@@ -11,68 +11,208 @@
 #include "base/logging.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/ref_counted.h"
-#include "base/single_thread_task_runner.h"
-#include "base/threading/platform_thread.h"
-#include "base/threading/thread_task_runner_handle.h"
+#include "base/sequenced_task_runner.h"
+#include "base/threading/sequenced_task_runner_handle.h"
 #include "base/time/tick_clock.h"
 
 namespace base {
 
-// BaseTimerTaskInternal is a simple delegate for scheduling a callback to
-// Timer in the thread's default task runner. It also handles the following
-// edge cases:
-// - deleted by the task runner.
-// - abandoned (orphaned) by Timer.
-class BaseTimerTaskInternal {
+// BaseTimerTaskInternal is a helper for scheduling Timer's callback on the
+// destination task runner (it is in charge of all methods to be executed on
+// destination task runner while Timer is in charge of all methods to be
+// executed on origin task runner). It owns itself on the destination task
+// runner and supports a few use cases:
+//  - Non-repeating timers:
+//      Invokes |stop_callback| on |origin_task_runner| when Run() fires,
+//      *before* running |task|. This ensures that anything posted to origin
+//      task runner from |task| will run in the scope of a stopped task runner
+//      as expected.
+//      Self-destructs when Run() or Abandon() fires, whichever happens first
+//      (any pending Run()/Abandon()/Reset() being cancelled through the
+//      WeakPtr).
+//  - Repeating timers:
+//      Re-enqueues |task| after each invocation of Run() until Abandon() is
+//      invoked (which will invalidate the WeakPtr of the pending Run()).
+//
+// An invariant assumed below is that there always is exactly one pending Run()
+// task associated with this BaseTimerTaskInternal. That invariant is maintained
+// by having Run() always either triggering self-destruction or reposting self.
+//
+// Note: Timer and BaseTimerTaskInternal try to communicate synchronously when
+// possible (i.e. when SetTaskRunner() wasn't used, which they can verify
+// through RunsTasksOnCurrentThread()). Timer must also always use asynchronous
+// calls when invoking Abandon() to avoid calling "delete this" from a reentrant
+// call while handling Run().
+class BaseTimerTaskInternal : public SupportsWeakPtr<BaseTimerTaskInternal> {
  public:
-  explicit BaseTimerTaskInternal(Timer* timer)
-      : timer_(timer) {
+  // A helper which invokes a Closure when destroyed. Required to notify
+  // BaseTimerTaskInternal when its Run() task is suppressed before it's
+  // scheduled (e.g. because the sequence is shutdown). Note: such cleanups
+  // can't be bound directly to ~BaseTimerTaskInternal() as it owns itself (its
+  // tasks use WeakPtrs and deleting them won't trigger
+  // ~BaseTimerTaskInternal()).
+  class CleanupTrigger {
+   public:
+    ~CleanupTrigger() { std::move(on_destruction_).Run(); }
+
+   private:
+    friend class BaseTimerTaskInternal;
+    CleanupTrigger(OnceClosure on_destruction)
+        : on_destruction_(std::move(on_destruction)) {}
+
+    OnceClosure on_destruction_;
+
+    DISALLOW_COPY_AND_ASSIGN(CleanupTrigger);
+  };
+
+  BaseTimerTaskInternal(const tracked_objects::Location& posted_from,
+                        const Closure& task,
+                        TimeDelta delay,
+                        bool is_repeating,
+                        TickClock* tick_clock,
+                        scoped_refptr<SequencedTaskRunner> origin_task_runner,
+                        const Closure& stop_callback)
+      : posted_from_(posted_from),
+        task_(task),
+        delay_(delay),
+        is_repeating_(is_repeating),
+        tick_clock_(tick_clock),
+        desired_run_time_(Now() + delay_),
+        origin_task_runner_(std::move(origin_task_runner)),
+        stop_callback_(stop_callback) {
+    destination_sequence_checker_.DetachFromSequence();
   }
 
-  ~BaseTimerTaskInternal() {
-    // This task may be getting cleared because the task runner has been
-    // destructed.  If so, don't leave Timer with a dangling pointer
-    // to this.
-    if (timer_)
-      timer_->StopAndAbandon();
-  }
+  ~BaseTimerTaskInternal() = default;
+
+  void Run(std::unique_ptr<CleanupTrigger> cleanup_trigger) {
+    DCHECK(destination_sequence_checker_.CalledOnValidSequence());
 
-  void Run() {
-    // timer_ is NULL if we were abandoned.
-    if (!timer_)
+    const TimeDelta delay_remaining = desired_run_time_ - Now();
+    if (delay_remaining > TimeDelta::FromMicroseconds(0)) {
+#if DCHECK_IS_ON()
+      DCHECK(was_reset_) << delay_remaining;
+#endif
+      SequencedTaskRunnerHandle::Get()->PostDelayedTask(
+          posted_from_, base::Bind(&BaseTimerTaskInternal::Run, AsWeakPtr(),
+                                   Passed(std::move(cleanup_trigger))),
+          delay_remaining);
       return;
+    }
 
-    // *this will be deleted by the task runner, so Timer needs to
-    // forget us:
-    timer_->scheduled_task_ = NULL;
+    // Trigger cleanup before running the |task_|. e.g. to Stop() OneShotTimers
+    // so that anything posted from the task runs in the context of a stopped
+    // timer as expected.
+    if (!is_repeating_)
+      cleanup_trigger.reset();
+
+    task_.Run();
+
+    if (is_repeating_) {
+      // TODO(gab): Currently rebasing on Now() to match prior status quo but
+      // think it would be better to merely |+= delay_| here to avoid the timer
+      // drifting with the delayed task overhead: http://crbug.com/676836.
+      desired_run_time_ = Now() + delay_;
+      SequencedTaskRunnerHandle::Get()->PostDelayedTask(
+          posted_from_, base::Bind(&BaseTimerTaskInternal::Run, AsWeakPtr(),
+                                   Passed(std::move(cleanup_trigger))),
+          delay_);
+#if DCHECK_IS_ON()
+      was_reset_ = false;
+#endif
+    } else {
+      delete this;
+    }
+  }
+
+  // Postpones the |desired_run_time_| by |delay_|.
+  void Reset() {
+    DCHECK(destination_sequence_checker_.CalledOnValidSequence());
+    DCHECK(!delay_.is_zero());
+
+    // Since Reset() is sequenced with Run() (and cancelled via the WeakPtr if
+    // Run() is scheduled before it -- i.e. delay expires as Reset() is posted):
+    // it is sufficient to merely update |desired_run_time_|.
+    desired_run_time_ = Now() + delay_;
 
-    // Although Timer should not call back into *this, let's clear
-    // the timer_ member first to be pedantic.
-    Timer* timer = timer_;
-    timer_ = NULL;
-    timer->RunScheduledTask();
+#if DCHECK_IS_ON()
+    was_reset_ = true;
+#endif
   }
 
-  // The task remains in the MessageLoop queue, but nothing will happen when it
-  // runs.
   void Abandon() {
-    timer_ = NULL;
+    DCHECK(destination_sequence_checker_.CalledOnValidSequence());
+
+    // Delete self, invalidating WeakPtrs (and as such cancelling the pending
+    // Run() task).
+    delete this;
+  }
+
+  std::unique_ptr<CleanupTrigger> CreateCleanupTrigger() {
+    // Note: it's important that CleanupTrigger's Closure be bound to a WeakPtr
+    // so that it only triggers if the Run() task is suppressed from its
+    // sequence while it was live (i.e. it shouldn't trigger when the task is
+    // destroyed because the WeakPtr was invalid when an abandoned Run() task
+    // was scheduled).
+    return WrapUnique(
+        new CleanupTrigger(Bind(&BaseTimerTaskInternal::Cleanup, AsWeakPtr())));
   }
 
  private:
-  Timer* timer_;
+  TimeTicks Now() const {
+    return tick_clock_ ? tick_clock_->NowTicks() : TimeTicks::Now();
+  }
+
+  // Cleans up state on the Timer side when this BaseTimerTaskInternal is done
+  // with its role.
+  void Cleanup() {
+    DCHECK(destination_sequence_checker_.CalledOnValidSequence());
+
+    if (origin_task_runner_->RunsTasksOnCurrentThread()) {
+      stop_callback_.Run();
+    } else {
+      origin_task_runner_->PostTask(FROM_HERE, stop_callback_);
+    }
+  }
+
+  const tracked_objects::Location posted_from_;
+  const Closure task_;
+  const TimeDelta delay_;
+  const bool is_repeating_;
+
+  TickClock* const tick_clock_;
+  TimeTicks desired_run_time_;
+
+  const scoped_refptr<SequencedTaskRunner> origin_task_runner_;
+  const Closure stop_callback_;
+
+  // Verifies that every operation (but construction) happens on the sequence
+  // where the task is intended to run.
+  SequenceChecker destination_sequence_checker_;
+
+#if DCHECK_IS_ON()
+  // Used to verify that reposts only occur when a Reset() was involved.
+  bool was_reset_ = false;
+#endif
+
+  DISALLOW_COPY_AND_ASSIGN(BaseTimerTaskInternal);
 };
 
 Timer::Timer(bool retain_user_task, bool is_repeating)
     : Timer(retain_user_task, is_repeating, nullptr) {}
 
 Timer::Timer(bool retain_user_task, bool is_repeating, TickClock* tick_clock)
-    : scheduled_task_(nullptr),
-      thread_id_(0),
-      is_repeating_(is_repeating),
+    : is_repeating_(is_repeating),
       retain_user_task_(retain_user_task),
       tick_clock_(tick_clock),
-      is_running_(false) {}
+      is_running_(false),
+      weak_ptr_factory_(this) {
+  // It is safe for the timer to be created on a different thread/sequence than
+  // the one from which the timer APIs are called. The first call to the
+  // checker's CalledOnValidSequence() method will re-bind the checker, and
+  // later calls will verify that the same task runner is used.
+  origin_sequence_checker_.DetachFromSequence();
+}
 
 Timer::Timer(const tracked_objects::Location& posted_from,
              TimeDelta delay,
@@ -85,150 +225,150 @@ Timer::Timer(const tracked_objects::Location& posted_from,
              const base::Closure& user_task,
              bool is_repeating,
              TickClock* tick_clock)
-    : scheduled_task_(nullptr),
-      posted_from_(posted_from),
+    : posted_from_(posted_from),
       delay_(delay),
       user_task_(user_task),
-      thread_id_(0),
       is_repeating_(is_repeating),
       retain_user_task_(true),
       tick_clock_(tick_clock),
-      is_running_(false) {}
+      is_running_(false),
+      weak_ptr_factory_(this) {
+  // See comment in other constructor.
+  origin_sequence_checker_.DetachFromSequence();
+}
 
 Timer::~Timer() {
-  StopAndAbandon();
+  // Note: |origin_sequence_checker_| is a SequenceCheckerImpl so this call
+  // is valid in non-DCHECK builds.
+  if (origin_sequence_checker_.CalledOnValidSequence()) {
+    Stop();
+  } else {
+    // As highlighted in the constructor. It's okay to start the Timer on a
+    // different sequence but it must then be sequentially stopped on that
+    // sequence as well before it can be deleted on its original sequence.
+    DCHECK(!is_running_);
+  }
 }
 
 bool Timer::IsRunning() const {
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
   return is_running_;
 }
 
 TimeDelta Timer::GetCurrentDelay() const {
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
   return delay_;
 }
 
-void Timer::SetTaskRunner(scoped_refptr<SingleThreadTaskRunner> task_runner) {
-  // Do not allow changing the task runner once something has been scheduled.
-  DCHECK_EQ(thread_id_, 0);
-  task_runner_.swap(task_runner);
+void Timer::SetTaskRunner(scoped_refptr<SequencedTaskRunner> task_runner) {
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
+
+  // Do not allow changing the task runner while something is scheduled.
+  DCHECK(!is_running_);
+  destination_task_runner_.swap(task_runner);
 }
 
 void Timer::Start(const tracked_objects::Location& posted_from,
                   TimeDelta delay,
                   const base::Closure& user_task) {
-  SetTaskInfo(posted_from, delay, user_task);
-  Reset();
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
+
+  // Do a best-effort to cancel the previous task, it may still racily fire
+  // though.
+  AbandonScheduledTask();
+
+  DCHECK(!is_running_);
+  posted_from_ = posted_from;
+  delay_ = delay;
+  user_task_ = user_task;
+
+  PostNewScheduledTask(delay_);
 }
 
 void Timer::Stop() {
-  is_running_ = false;
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
+
+  AbandonScheduledTask();
   if (!retain_user_task_)
     user_task_.Reset();
 }
 
 void Timer::Reset() {
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
   DCHECK(!user_task_.is_null());
 
-  // If there's no pending task, start one up and return.
-  if (!scheduled_task_) {
+  if (is_running_) {

[vmpstr, 2017/01/12 20:04:22]

I kinda prefer 

if (!is_running_) {
  PostNewScheduledTask(delay_);
  return;
}

...

I find that it makes it a bit more readable

[gab, 2017/01/25 20:14:56]

Agreed, done.

+    if (destination_task_runner_) {
+      destination_task_runner_->PostTask(
+          FROM_HERE,
+          base::Bind(&BaseTimerTaskInternal::Reset, scheduled_task_weak_ref_));
+    } else {
+      // |is_running_| guarantees a valid pointer when not using an external
+      // |destination_task_runner_|.
+      DCHECK(scheduled_task_weak_ref_);
+      scheduled_task_weak_ref_->Reset();
+    }
+  } else {
+    // This is required to handle the following use case (when
+    // |retain_user_task_ ==true|): Start() => Stop() => Reset().
     PostNewScheduledTask(delay_);
-    return;
-  }
-
-  // Set the new desired_run_time_.
-  if (delay_ > TimeDelta::FromMicroseconds(0))
-    desired_run_time_ = Now() + delay_;
-  else
-    desired_run_time_ = TimeTicks();
-
-  // We can use the existing scheduled task if it arrives before the new
-  // desired_run_time_.
-  if (desired_run_time_ >= scheduled_run_time_) {
-    is_running_ = true;
-    return;
   }
-
-  // We can't reuse the scheduled_task_, so abandon it and post a new one.
-  AbandonScheduledTask();
-  PostNewScheduledTask(delay_);
 }
 
-TimeTicks Timer::Now() const {
-  return tick_clock_ ? tick_clock_->NowTicks() : TimeTicks::Now();
-}
+void Timer::PostNewScheduledTask(TimeDelta delay) {
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
+  DCHECK(SequencedTaskRunnerHandle::IsSet());
+  DCHECK(!is_running_);
 
-void Timer::SetTaskInfo(const tracked_objects::Location& posted_from,
-                        TimeDelta delay,
-                        const base::Closure& user_task) {
-  posted_from_ = posted_from;
-  delay_ = delay;
-  user_task_ = user_task;
-}
+  scoped_refptr<SequencedTaskRunner> destination_task_runner =

[vmpstr, 2017/01/12 20:04:22]

Do you need a ref here? Can it be SequencedTaskRunner* with .get() on the rhs

[gab, 2017/01/25 20:14:56]

I had the same thought and had originally written it as a raw pointer but doing
so crashes because SequencedTaskRunnerHandle::Get() returns a ref as an r-value
(and in the case of SequencedWorkerPool that's actually a temp
SequencedTaskRunner object under the hood).

+      destination_task_runner_ ? destination_task_runner_
+                               : SequencedTaskRunnerHandle::Get();
 
-void Timer::PostNewScheduledTask(TimeDelta delay) {
-  DCHECK(scheduled_task_ == NULL);
   is_running_ = true;
-  scheduled_task_ = new BaseTimerTaskInternal(this);
-  if (delay > TimeDelta::FromMicroseconds(0)) {
-    GetTaskRunner()->PostDelayedTask(posted_from_,
-        base::Bind(&BaseTimerTaskInternal::Run, base::Owned(scheduled_task_)),
-        delay);
-    scheduled_run_time_ = desired_run_time_ = Now() + delay;
-  } else {
-    GetTaskRunner()->PostTask(posted_from_,
-        base::Bind(&BaseTimerTaskInternal::Run, base::Owned(scheduled_task_)));
-    scheduled_run_time_ = desired_run_time_ = TimeTicks();
-  }
-  // Remember the thread ID that posts the first task -- this will be verified
-  // later when the task is abandoned to detect misuse from multiple threads.
-  if (!thread_id_)
-    thread_id_ = static_cast<int>(PlatformThread::CurrentId());
-}
-
-scoped_refptr<SingleThreadTaskRunner> Timer::GetTaskRunner() {
-  return task_runner_.get() ? task_runner_ : ThreadTaskRunnerHandle::Get();
+  BaseTimerTaskInternal* scheduled_task = new BaseTimerTaskInternal(

[vmpstr, 2017/01/12 20:04:22]

Can you add a comment saying that the task will delete itself?

[gab, 2017/01/25 20:14:56]

Done.

+      posted_from_, user_task_, delay, is_repeating_, tick_clock_,
+      SequencedTaskRunnerHandle::Get(),
+      base::Bind(&Timer::Stop, weak_ptr_factory_.GetWeakPtr()));
+  scheduled_task_weak_ref_ = scheduled_task->AsWeakPtr();
+  destination_task_runner->PostDelayedTask(
+      posted_from_,
+      base::Bind(&BaseTimerTaskInternal::Run, scheduled_task_weak_ref_,
+                 Passed(scheduled_task->CreateCleanupTrigger())),
+      delay);
 }
 
 void Timer::AbandonScheduledTask() {
-  DCHECK(thread_id_ == 0 ||
-         thread_id_ == static_cast<int>(PlatformThread::CurrentId()));
-  if (scheduled_task_) {
-    scheduled_task_->Abandon();
-    scheduled_task_ = NULL;
-  }
-}
-
-void Timer::RunScheduledTask() {
-  // Task may have been disabled.
-  if (!is_running_)
-    return;
-
-  // First check if we need to delay the task because of a new target time.
-  if (desired_run_time_ > scheduled_run_time_) {
-    // Now() can be expensive, so only call it if we know the user has changed
-    // the desired_run_time_.
-    TimeTicks now = Now();
-    // Task runner may have called us late anyway, so only post a continuation
-    // task if the desired_run_time_ is in the future.
-    if (desired_run_time_ > now) {
-      // Post a new task to span the remaining time.
-      PostNewScheduledTask(desired_run_time_ - now);
-      return;
+  DCHECK(origin_sequence_checker_.CalledOnValidSequence());
+
+  if (is_running_) {

[vmpstr, 2017/01/12 20:04:22]

if (!is_running_)
  return;

...

[gab, 2017/01/25 20:14:56]

Done.

+    // Toggle |is_running| first to be reentrancy safe, just in case.
+    is_running_ = false;
+
+    const Closure abandon =
+        base::Bind(&BaseTimerTaskInternal::Abandon, scheduled_task_weak_ref_);
+    if (destination_task_runner_) {
+      destination_task_runner_->PostTask(FROM_HERE, abandon);
+    } else if (SequencedTaskRunnerHandle::IsSet()) {
+      // Post Abandon() as an asynchronous call to avoid deleting
+      // BaseTimerTaskInternal while it handles Run() (reentrancy is possible,
+      // e.g. when the task itself deletes the Timer or when a OneShotTimer's
+      // BaseTimerTaskInternal::Run() invoked its |stop_callback_|
+      // synchronously).
+      SequencedTaskRunnerHandle::Get()->PostTask(FROM_HERE, abandon);
+    } else {
+      // |is_running_| guarantees a valid pointer when not using an external
+      // |destination_task_runner_|.
+      DCHECK(scheduled_task_weak_ref_);
+      // The reentrancy situations described above happen while Run() is active
+      // which implies SequencedTaskRunnerHandle::IsSet().
+      // SequencedTaskRunnerHandle not being set typically means this
+      // AbandonScheduledTask() was triggered by ~Timer() on shutdown per the
+      // sequence being destroyed and releasing unscheduled tasks in which case
+      // it's safe to abandon synchronously.
+      scheduled_task_weak_ref_->Abandon();
     }
+    scheduled_task_weak_ref_ = nullptr;
   }
-
-  // Make a local copy of the task to run. The Stop method will reset the
-  // user_task_ member if retain_user_task_ is false.
-  base::Closure task = user_task_;
-
-  if (is_repeating_)
-    PostNewScheduledTask(delay_);
-  else
-    Stop();
-
-  task.Run();
-
-  // No more member accesses here: *this could be deleted at this point.
 }
 
 }  // namespace base