sync: Refactor job ownership in SyncScheduler

sync/engine/sync_scheduler_impl.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 "sync/engine/sync_scheduler_impl.h"
 
 #include <algorithm>
 #include <cstring>
 
 #include "base/auto_reset.h"
@@ skipping 65 matching lines @@
     const base::Closure& ready_task)
     : source(source),
       types_to_download(types_to_download),
       routing_info(routing_info),
       ready_task(ready_task) {
   DCHECK(!ready_task.is_null());
 }
 ConfigurationParams::~ConfigurationParams() {}
 
 SyncSchedulerImpl::WaitInterval::WaitInterval()
-    : mode(UNKNOWN),
-      had_nudge(false) {}
+    : mode(UNKNOWN) {}
 
 SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
-    : mode(mode), had_nudge(false), length(length) {}
+    : mode(mode), length(length) {}
 
 SyncSchedulerImpl::WaitInterval::~WaitInterval() {}
 
 #define ENUM_CASE(x) case x: return #x; break;
 
 const char* SyncSchedulerImpl::WaitInterval::GetModeString(Mode mode) {
   switch (mode) {
     ENUM_CASE(UNKNOWN);
     ENUM_CASE(EXPONENTIAL_BACKOFF);
     ENUM_CASE(THROTTLED);
@@ skipping 209 matching lines @@
     SDVLOG(2) << "No change in routing info, calling ready task directly.";
     params.ready_task.Run();
   }
 
   return true;
 }
 
 SyncSchedulerImpl::JobProcessDecision
 SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob& job,
                                              JobPriority priority) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);

[tim (not reviewing), 2013/04/05 16:56:44]

It looks like all non-DCHECK use of sync_loop_ was removed with this CL. In that
case, we should remove the member and replace these DCHECKs with
NonThreadSafe::CalledOnValidThread.

[rlarocque, 2013/04/05 19:13:29]

Done.

   DCHECK(wait_interval_.get());
   DCHECK_NE(job.purpose(), SyncSessionJob::POLL);
 
   SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
             << WaitInterval::GetModeString(wait_interval_->mode)
-            << (wait_interval_->had_nudge ? " (had nudge)" : "")
             << ((priority == CANARY_PRIORITY) ? " (canary)" : "");
 
   // If we save a job while in a WaitInterval, there is a well-defined moment
   // in time in the future when it makes sense for that SAVE-worthy job to try
   // running again -- the end of the WaitInterval.
   DCHECK(job.purpose() == SyncSessionJob::NUDGE ||
          job.purpose() == SyncSessionJob::CONFIGURATION);
 
   // If throttled, there's a clock ticking to unthrottle. We want to get
   // on the same train.
   if (wait_interval_->mode == WaitInterval::THROTTLED)
     return SAVE;
 
   DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
   if (job.purpose() == SyncSessionJob::NUDGE) {
     if (mode_ == CONFIGURATION_MODE)
       return SAVE;
 
     // If we already had one nudge then just drop this nudge. We will retry
     // later when the timer runs out.
     if (priority == NORMAL_PRIORITY)
-      return wait_interval_->had_nudge ? DROP : CONTINUE;
+      return DROP;
     else // We are here because timer ran out. So retry.

[tim (not reviewing), 2013/04/05 16:56:44]

In fact this comment isn't true, so we should fix it.  We can wind up here if we
get a network change notification during EXPONENTIAL_BACKOFF, for example.  In
that case, we should attempt to sync before the WaitInterval expires.  

[rlarocque, 2013/04/05 19:13:29]

Done.

       return CONTINUE;
   }
   return (priority == CANARY_PRIORITY) ? CONTINUE : SAVE;
 }
 
 SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
     const SyncSessionJob& job,
     JobPriority priority) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
@@ skipping 174 matching lines @@
   }
 
   TimeDelta run_delay =
       pending_nudge_job_->scheduled_start() - TimeTicks::Now();
   if (run_delay < TimeDelta::FromMilliseconds(0))
     run_delay = TimeDelta::FromMilliseconds(0);
   SDVLOG_LOC(nudge_location, 2)
       << "Scheduling a nudge with "
       << run_delay.InMilliseconds() << " ms delay";
 
-  PostDelayedTask(
-      nudge_location,
-      "DoSyncSessionJob",
-      base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoNudgeSyncSessionJob),
-                 weak_ptr_factory_.GetWeakPtr(),
-                 NORMAL_PRIORITY),
-      run_delay);
+  if (started_) {
+    pending_wakeup_timer_.Start(
+        nudge_location,
+        run_delay,
+        base::Bind(&SyncSchedulerImpl::DoNudgeSyncSessionJob,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   NORMAL_PRIORITY));
+  }
 }
 
 const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
   switch (mode) {
     ENUM_CASE(CONFIGURATION_MODE);
     ENUM_CASE(NORMAL_MODE);
   }
   return "";
 }
 
 const char* SyncSchedulerImpl::GetDecisionString(
     SyncSchedulerImpl::JobProcessDecision mode) {
   switch (mode) {
     ENUM_CASE(CONTINUE);
     ENUM_CASE(SAVE);
     ENUM_CASE(DROP);
   }
   return "";
 }
 
-void SyncSchedulerImpl::PostDelayedTask(
-    const tracked_objects::Location& from_here,
-    const char* name, const base::Closure& task, base::TimeDelta delay) {
-  SDVLOG_LOC(from_here, 3) << "Posting " << name << " task with "
-                           << delay.InMilliseconds() << " ms delay";
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  if (!started_) {
-    SDVLOG(1) << "Not posting task as scheduler is stopped.";
-    return;
-  }
-  // This cancels the previous task, if one existed.
-  pending_wakeup_event_.Reset(task);
-  sync_loop_->PostDelayedTask(from_here,
-                              pending_wakeup_event_.callback(),
-                              delay);
-}
-
 bool SyncSchedulerImpl::DoSyncSessionJobImpl(scoped_ptr<SyncSessionJob> job,
                                              JobPriority priority) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
   JobProcessDecision decision = DecideOnJob(*job, priority);
   SDVLOG(2) << "Should run "
             << SyncSessionJob::GetPurposeString(job->purpose())
             << " in mode " << GetModeString(mode_)
             << " with source " << job->source_info().updates_source
             << ": " << GetDecisionString(decision);
   if (decision != CONTINUE) {
     if (decision == SAVE) {
       if (job->purpose() == SyncSessionJob::CONFIGURATION) {
         pending_configure_job_ = job.Pass();
-
-        // It's very unlikely, but possible, that the WaitInterval's wakeup task
-        // isn't actually active at this point.  Sometimes the WaitInterval gets
-        // preempted by a nudge-while-in-backoff.  We can't just assume that
-        // there's a task waiting to wake us up once the WaitInterval expires;
-        // we need to ensure it by rescheduling the WaitInterval (while taking
-        // into account any time already spent waiting, of course).
-        ResumeWaiting();
       } else {
         pending_nudge_job_ = job.Pass();
       }
     } else {
       DCHECK_EQ(decision, DROP);
     }
     return false;
   }
 
   DVLOG(2) << "Creating sync session with routes "
@@ skipping 24 matching lines @@
     RestartWaiting();
     return success;
   }
 
   if (!success)
     ScheduleNextSync(job.Pass(), &session);
 
   return success;
 }
 
-bool SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {
-  return DoSyncSessionJobImpl(pending_nudge_job_.Pass(), priority);
+void SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {
+  DoSyncSessionJobImpl(pending_nudge_job_.Pass(), priority);
 }
 
 bool SyncSchedulerImpl::DoConfigurationSyncSessionJob(JobPriority priority) {
   return DoSyncSessionJobImpl(pending_configure_job_.Pass(), priority);
 }
 
 bool SyncSchedulerImpl::ShouldPoll() {
   if (wait_interval_.get()) {
     SDVLOG(2) << "Not running poll in wait interval.";
     return false;
@@ skipping 108 matching lines @@
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(finished_job->purpose() == SyncSessionJob::CONFIGURATION
          || finished_job->purpose() == SyncSessionJob::NUDGE);
 
   // TODO(rlarocque): There's no reason why we should blindly backoff and retry
   // if we don't succeed.  Some types of errors are not likely to disappear on
   // their own.  With the return values now available in the old_job.session,
   // we should be able to detect such errors and only retry when we detect
   // transient errors.
 
-  if (IsBackingOff() && wait_interval_->timer.IsRunning() &&
-      mode_ == NORMAL_MODE) {
-    // When in normal mode, we allow up to one nudge per backoff interval.  It
-    // appears that this was our nudge for this interval, and it failed.
-    //
-    // Note: This does not prevent us from running canary jobs.  For example,
-    // an IP address change might still result in another nudge being executed
-    // during this backoff interval.
-    SDVLOG(2) << "A nudge during backoff failed, creating new pending nudge.";
-    DCHECK_EQ(SyncSessionJob::NUDGE, finished_job->purpose());
-    DCHECK(!wait_interval_->had_nudge);
-
-    wait_interval_->had_nudge = true;
-    DCHECK(!pending_nudge_job_);
-
-    pending_nudge_job_ = finished_job.Pass();
-    RestartWaiting();
-  } else {
-    // Either this is the first failure or a consecutive failure after our
-    // backoff timer expired.  We handle it the same way in either case.
-    SDVLOG(2) << "Non-'backoff nudge' SyncShare job failed";
-    HandleContinuationError(finished_job.Pass(), session);
-  }
+  SDVLOG(2) << "SyncShare job failed; will start or update backoff";
+  HandleContinuationError(finished_job.Pass(), session);
 }
 
 void SyncSchedulerImpl::AdjustPolling(const SyncSessionJob* old_job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   TimeDelta poll  = (!session_context_->notifications_enabled()) ?
       syncer_short_poll_interval_seconds_ :
       syncer_long_poll_interval_seconds_;
   bool rate_changed = !poll_timer_.IsRunning() ||
                        poll != poll_timer_.GetCurrentDelay();
 
   if (old_job && old_job->purpose() != SyncSessionJob::POLL && !rate_changed)
     poll_timer_.Reset();
 
   if (!rate_changed)
     return;
 
   // Adjust poll rate.
   poll_timer_.Stop();
   poll_timer_.Start(FROM_HERE, poll, this,
                     &SyncSchedulerImpl::PollTimerCallback);
 }
 
-void SyncSchedulerImpl::ResumeWaiting() {
-  TimeDelta length =
-      wait_interval_->timer.desired_run_time() - TimeTicks::Now();
-  wait_interval_->length = length < TimeDelta::FromSeconds(0) ?
-      TimeDelta::FromSeconds(0) : length;
-  RestartWaiting();
-}
-
 void SyncSchedulerImpl::RestartWaiting() {
   CHECK(wait_interval_.get());
-  wait_interval_->timer.Stop();
   DCHECK(wait_interval_->length >= TimeDelta::FromSeconds(0));
   if (wait_interval_->mode == WaitInterval::THROTTLED) {
-    pending_wakeup_event_.Reset(base::Bind(&SyncSchedulerImpl::Unthrottle,
-                                           weak_ptr_factory_.GetWeakPtr()));
-
+    pending_wakeup_timer_.Start(
+        FROM_HERE,
+        wait_interval_->length,
+        base::Bind(&SyncSchedulerImpl::Unthrottle,
+                   weak_ptr_factory_.GetWeakPtr()));
   } else {
-    pending_wakeup_event_.Reset(base::Bind(&SyncSchedulerImpl::TryCanaryJob,
-                                           weak_ptr_factory_.GetWeakPtr()));
+    pending_wakeup_timer_.Start(
+        FROM_HERE,
+        wait_interval_->length,
+        base::Bind(&SyncSchedulerImpl::TryCanaryJob,
+                   weak_ptr_factory_.GetWeakPtr()));
   }
-  wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
-                              pending_wakeup_event_.callback());
 }
 
 void SyncSchedulerImpl::HandleContinuationError(
     scoped_ptr<SyncSessionJob> old_job,
     SyncSession* session) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   TimeDelta length = delay_provider_->GetDelay(
       IsBackingOff() ? wait_interval_->length :
           delay_provider_->GetInitialDelay(
               session->status_controller().model_neutral_state()));
 
   SDVLOG(2) << "In handle continuation error with "
             << SyncSessionJob::GetPurposeString(old_job->purpose())
             << " job. The time delta(ms) is "
             << length.InMilliseconds();
 
-  // This will reset the had_nudge variable as well.
   wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
                                         length));
   NotifyRetryTime(base::Time::Now() + length);
   old_job->set_scheduled_start(TimeTicks::Now() + length);
   if (old_job->purpose() == SyncSessionJob::CONFIGURATION) {
     SDVLOG(2) << "Configuration did not succeed, scheduling retry.";
     // Config params should always get set.
     DCHECK(!old_job->config_params().ready_task.is_null());
     DCHECK(!pending_configure_job_);
     pending_configure_job_ = old_job.Pass();
@@ skipping 18 matching lines @@
 
 void SyncSchedulerImpl::StopImpl(const base::Closure& callback) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "StopImpl called";
 
   // Kill any in-flight method calls.
   weak_ptr_factory_.InvalidateWeakPtrs();
   wait_interval_.reset();
   NotifyRetryTime(base::Time());
   poll_timer_.Stop();
-  pending_wakeup_event_.Cancel();
+  pending_wakeup_timer_.Stop();
   pending_nudge_job_.reset();
   pending_configure_job_.reset();
   if (started_) {
     started_ = false;
   }
   if (!callback.is_null())
     callback.Run();
 }
 
 // This is the only place where we invoke DoSyncSessionJob with canary
@@ skipping 133 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // namespace syncer