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 66 matching lines @@
     : 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),
-      pending_configure_job(NULL) {}
+      had_nudge(false) {}
 
 SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
-    : mode(mode), had_nudge(false), length(length),
-      pending_configure_job(NULL) {}
+    : mode(mode), had_nudge(false), 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 59 matching lines @@
       name_(name),
       sync_loop_(MessageLoop::current()),
       started_(false),
       syncer_short_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultShortPollIntervalSeconds)),
       syncer_long_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultLongPollIntervalSeconds)),
       sessions_commit_delay_(
           TimeDelta::FromSeconds(kDefaultSessionsCommitDelaySeconds)),
       mode_(NORMAL_MODE),
-      // Start with assuming everything is fine with the connection.
-      // At the end of the sync cycle we would have the correct status.
-      pending_nudge_(NULL),
       delay_provider_(delay_provider),
       syncer_(syncer),
       session_context_(context),
       no_scheduling_allowed_(false) {
   DCHECK(sync_loop_);
 }
 
 SyncSchedulerImpl::~SyncSchedulerImpl() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   StopImpl(base::Closure());
@@ skipping 22 matching lines @@
   //
   // 1. We're in exponential backoff.
   // 2. We're silenced / throttled.
   // 3. A nudge was saved previously due to not having a valid auth token.
   // 4. A nudge was scheduled + saved while in configuration mode.
   //
   // In all cases except (2), we want to retry contacting the server. We
   // call DoCanaryJob to achieve this, and note that nothing -- not even a
   // canary job -- can bypass a THROTTLED WaitInterval. The only thing that
   // has the authority to do that is the Unthrottle timer.
-  scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
-  if (!pending.get())
-    return;
-  DoCanaryJob(pending.Pass());
+  TryCanaryJob();
 }
 
 void SyncSchedulerImpl::Start(Mode mode) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   std::string thread_name = MessageLoop::current()->thread_name();
   if (thread_name.empty())
     thread_name = "<Main thread>";
   SDVLOG(2) << "Start called from thread "
             << thread_name << " with mode " << GetModeString(mode);
   if (!started_) {
     started_ = true;
     SendInitialSnapshot();
   }
 
   DCHECK(!session_context_->account_name().empty());
   DCHECK(syncer_.get());
   Mode old_mode = mode_;
   mode_ = mode;
   AdjustPolling(NULL);  // Will kick start poll timer if needed.
 
-  if (old_mode != mode_) {
-    // We just changed our mode. See if there are any pending jobs that we could
-    // execute in the new mode.
-    if (mode_ == NORMAL_MODE) {
-      // It is illegal to switch to NORMAL_MODE if a previous CONFIGURATION job
-      // has not yet completed.
-      DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
-    }
-
-    scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
-    if (pending.get()) {
-      SDVLOG(2) << "Executing pending job. Good luck!";
-      DoSyncSessionJob(pending.Pass(), NORMAL_PRIORITY);
-    }
+  if (old_mode != mode_ && mode_ == NORMAL_MODE && pending_nudge_job_) {
+    // We just got back to normal mode.  Let's try to run the work that was
+    // queued up while we were configuring.
+    DoNudgeSyncSessionJob(NORMAL_PRIORITY);
   }
 }
 
 void SyncSchedulerImpl::SendInitialSnapshot() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   scoped_ptr<SyncSession> dummy(new SyncSession(
           session_context_, this, SyncSourceInfo()));
   SyncEngineEvent event(SyncEngineEvent::STATUS_CHANGED);
   event.snapshot = dummy->TakeSnapshot();
   session_context_->NotifyListeners(event);
@@ skipping 23 matching lines @@
     const ConfigurationParams& params) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
   DCHECK_EQ(CONFIGURATION_MODE, mode_);
   DCHECK(!params.ready_task.is_null());
   CHECK(started_) << "Scheduler must be running to configure.";
   SDVLOG(2) << "Reconfiguring syncer.";
 
   // Only one configuration is allowed at a time. Verify we're not waiting
   // for a pending configure job.
-  DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
+  DCHECK(!pending_configure_job_);
 
   ModelSafeRoutingInfo restricted_routes;
   BuildModelSafeParams(params.types_to_download,
                        params.routing_info,
                        &restricted_routes);
   session_context_->set_routing_info(restricted_routes);
 
   // Only reconfigure if we have types to download.
   if (!params.types_to_download.Empty()) {
     DCHECK(!restricted_routes.empty());
-    scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
+    pending_configure_job_.reset(new SyncSessionJob(
         SyncSessionJob::CONFIGURATION,
         TimeTicks::Now(),
         SyncSourceInfo(params.source,
                        ModelSafeRoutingInfoToInvalidationMap(
                            restricted_routes,
                            std::string())),
         params));
-    bool succeeded = DoSyncSessionJob(job.Pass(), NORMAL_PRIORITY);
+    bool succeeded = DoConfigurationSyncSessionJob(NORMAL_PRIORITY);
 
     // If we failed, the job would have been saved as the pending configure
     // job and a wait interval would have been set.
     if (!succeeded) {
-      DCHECK(wait_interval_.get() && wait_interval_->pending_configure_job);
+      DCHECK(pending_configure_job_);
       return false;
+    } else {
+      DCHECK(!pending_configure_job_);
     }
   } else {
     SDVLOG(2) << "No change in routing info, calling ready task directly.";
     params.ready_task.Run();
   }
 
   return true;
 }
 
 SyncSchedulerImpl::JobProcessDecision
@@ skipping 21 matching lines @@
 
   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;
     else // We are here because timer ran out. So retry.

[tim (not reviewing), 2013/04/04 18:07:30]

Just noticed this clause is redundant and somewhat confusing -- the return on
line 352 handles it and I think it is clearer than this comment.  Maybe we
should remove it?

[rlarocque, 2013/04/05 00:48:51]

I was planning to leave it alone for now.  There's a lot about DecideOnJob()
that is redundant and confusing.  Fixing it is a job for another CL.

       return CONTINUE;
   }
   return (priority == CANARY_PRIORITY) ? CONTINUE : SAVE;
 }
 
 SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
     const SyncSessionJob& job,
     JobPriority priority) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
@@ skipping 66 matching lines @@
   // Decision now rests on state of auth tokens.
   if (!session_context_->connection_manager()->HasInvalidAuthToken())
     return CONTINUE;
 
   SDVLOG(2) << "No valid auth token. Using that to decide on job.";
   // Running the job would require updated auth, so we can't honour
   // job.scheduled_start().
   return job.purpose() == SyncSessionJob::NUDGE ? SAVE : DROP;
 }
 
-void SyncSchedulerImpl::HandleSaveJobDecision(scoped_ptr<SyncSessionJob> job) {
-  const bool is_nudge = job->purpose() == SyncSessionJob::NUDGE;
-  if (is_nudge && pending_nudge_) {
-    SDVLOG(2) << "Coalescing a pending nudge";
-    // TODO(tim): This basically means we never use the more-careful coalescing
-    // logic in ScheduleNudgeImpl that takes the min of the two nudge start
-    // times, because we're calling this function first.  Pull this out
-    // into a function to coalesce + set start times and reuse.
-    pending_nudge_->CoalesceSources(job->source_info());
-    return;
-  }
-
-  scoped_ptr<SyncSessionJob> job_to_save = job->Clone();
-  if (wait_interval_.get() && !wait_interval_->pending_configure_job) {
-    // This job should be made the new canary.
-    if (is_nudge) {
-      pending_nudge_ = job_to_save.get();
-    } else {
-      SDVLOG(2) << "Saving a configuration job";
-      DCHECK_EQ(job->purpose(), SyncSessionJob::CONFIGURATION);
-      DCHECK(!wait_interval_->pending_configure_job);
-      DCHECK_EQ(mode_, CONFIGURATION_MODE);
-      DCHECK(!job->config_params().ready_task.is_null());
-      // The only nudge that could exist is a scheduled canary nudge.
-      DCHECK(!unscheduled_nudge_storage_.get());
-      if (pending_nudge_) {
-        // Pre-empt the nudge canary and abandon the old nudge (owned by task).
-        unscheduled_nudge_storage_ = pending_nudge_->Clone();
-        pending_nudge_ = unscheduled_nudge_storage_.get();
-      }
-      wait_interval_->pending_configure_job = job_to_save.get();
-    }
-    TimeDelta length =
-        wait_interval_->timer.desired_run_time() - TimeTicks::Now();
-    wait_interval_->length = length < TimeDelta::FromSeconds(0) ?
-        TimeDelta::FromSeconds(0) : length;
-    RestartWaiting(job_to_save.Pass());
-    return;
-  }
-
-  // Note that today there are no cases where we SAVE a CONFIGURATION job
-  // when we're not in a WaitInterval. See bug 147736.
-  DCHECK(is_nudge);
-  // There may or may not be a pending_configure_job. Either way this nudge
-  // is unschedulable.
-  pending_nudge_ = job_to_save.get();
-  unscheduled_nudge_storage_ = job_to_save.Pass();
-}
-
-// Functor for std::find_if to search by ModelSafeGroup.
-struct ModelSafeWorkerGroupIs {
-  explicit ModelSafeWorkerGroupIs(ModelSafeGroup group) : group(group) {}
-  bool operator()(ModelSafeWorker* w) {
-    return group == w->GetModelSafeGroup();
-  }
-  ModelSafeGroup group;
-};
-
 void SyncSchedulerImpl::ScheduleNudgeAsync(
     const TimeDelta& desired_delay,
     NudgeSource source, ModelTypeSet types,
     const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG_LOC(nudge_location, 2)
       << "Nudge scheduled with delay "
       << desired_delay.InMilliseconds() << " ms, "
       << "source " << GetNudgeSourceString(source) << ", "
       << "types " << ModelTypeSetToString(types);
@@ skipping 59 matching lines @@
   scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
       SyncSessionJob::NUDGE,
       TimeTicks::Now() + delay,
       info,
       ConfigurationParams()));
   JobProcessDecision decision = DecideOnJob(*job, NORMAL_PRIORITY);
   SDVLOG(2) << "Should run "
             << SyncSessionJob::GetPurposeString(job->purpose())
             << " in mode " << GetModeString(mode_)
             << ": " << GetDecisionString(decision);
-  if (decision != CONTINUE) {
-    // End of the line, though we may save the job for later.
-    if (decision == SAVE) {
-      HandleSaveJobDecision(job.Pass());
-    } else {
-      DCHECK_EQ(decision, DROP);
-    }
+  if (decision == DROP) {
     return;
   }
 
-  if (pending_nudge_) {
-    SDVLOG(2) << "Rescheduling pending nudge";
-    pending_nudge_->CoalesceSources(job->source_info());
-    // Choose the start time as the earliest of the 2. Note that this means
-    // if a nudge arrives with delay (e.g. kDefaultSessionsCommitDelaySeconds)
-    // but a nudge is already scheduled to go out, we'll send the (tab) commit
-    // without waiting.
-    pending_nudge_->set_scheduled_start(
-        std::min(job->scheduled_start(), pending_nudge_->scheduled_start()));
-    // Abandon the old task by cloning and replacing the session.
-    // It's possible that by "rescheduling" we're actually taking a job that
-    // was previously unscheduled and giving it wings, so take care to reset
-    // unscheduled nudge storage.
-    job = pending_nudge_->Clone();
-    pending_nudge_ = NULL;
-    unscheduled_nudge_storage_.reset();
-    // It's also possible we took a canary job, since we allow one nudge
-    // per backoff interval.
-    DCHECK(!wait_interval_ || !wait_interval_->had_nudge);
+  // Try to coalesce in both SAVE and CONTINUE cases.
+  if (pending_nudge_job_) {
+    pending_nudge_job_->CoalesceSources(job->source_info());
+    if (decision == CONTINUE) {
+      // Only update the scheduled_start if we're going to reschedule.
+      pending_nudge_job_->set_scheduled_start(
+          std::min(job->scheduled_start(),
+                   pending_nudge_job_->scheduled_start()));
+    }
+  } else {
+    pending_nudge_job_ = job.Pass();
   }
 
-  TimeDelta run_delay = job->scheduled_start() - TimeTicks::Now();
+  if (decision == SAVE) {
+    return;
+  }
+
+  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";
 
-  pending_nudge_ = job.get();
-  PostDelayedTask(nudge_location, "DoSyncSessionJob",
-      base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoSyncSessionJob),
-                                    weak_ptr_factory_.GetWeakPtr(),
-                                    base::Passed(&job),
-                                    NORMAL_PRIORITY),
+  PostDelayedTask(
+      nudge_location,
+      "DoSyncSessionJob",
+      base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoNudgeSyncSessionJob),
+                 weak_ptr_factory_.GetWeakPtr(),
+                 NORMAL_PRIORITY),
       run_delay);
 }
 
 const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
   switch (mode) {
     ENUM_CASE(CONFIGURATION_MODE);
     ENUM_CASE(NORMAL_MODE);
   }
   return "";
 }
@@ skipping 16 matching lines @@
   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_.Reset(task);
   sync_loop_->PostDelayedTask(from_here, pending_wakeup_.callback(), delay);
 }
 
-bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job,
-                                         JobPriority priority) {
+bool SyncSchedulerImpl::DoSyncSessionJobImpl(scoped_ptr<SyncSessionJob> job,
+                                             JobPriority priority) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  if (job->purpose() == SyncSessionJob::NUDGE) {
-    pending_nudge_ = NULL;
-  }
 
   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) {
-      HandleSaveJobDecision(job.Pass());
+      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();

[rlarocque, 2013/04/02 23:01:12]

Here's a paragraph that I had originally intended to write in the commit
message:
It turns out I was wrong.  We can decide to SAVE a nudge-while-in-backoff.  That
happens when we switch into configure mode between ScheduleNudgeImpl() and
DoSyncSessionJob().  If we received a CONFIGURE in the interim, and we decided
to SAVE it (because we are in an EXPONENTIAL_BACKOFF), we might be tempted to
assume that it's save to just set the pending_configure_job_ and wait for the
canary at the end of the wait interval.  That would be wrong, because the
nudge-while-in-backoff has preempted the job at the end of the WaitInterval. 
The result is that neither the nudge nor the configure will be run.

ResumeWaiting() solves this problem by making sure that the current pending job
is the canary at the end of the wait interval.

I have some ideas of safer/better ways to fix this problem.  My preference would
be to remove the nudge-while-in-backoff altogether, but there are other options.
 We should talk in person about this.  

[tim (not reviewing), 2013/04/04 18:07:30]

It looks like we actually keep the WaitInterval timer set even though we're
resetting the pending_wakeup_, so it's not as clean of a "preemptive" cut as one
may think.  

I didn't fully realize PostDelayedTask is now a destructive operation. e.g we
need to be really careful not to ever call it (except the one careful case it is
called today) because it replaces anything that is posted. I think
'PostDelayedTask' is a dangerous name because it is *not* the same as posting to
MessageLoop::PostDelayedTask at all. 

We should fix that, at least by renaming. As for WaitIntervals, should
PostDelayedTask really let this half-state happen? Can we avoid having the
responsibility of a wait interval callback split between WaitInterval and
sync_loop_?  We can talk in person this afternoon.

+      } else {
+        pending_nudge_job_ = job.Pass();
+      }
     } else {
       DCHECK_EQ(decision, DROP);
     }
     return false;
   }
 
   DVLOG(2) << "Creating sync session with routes "
            << ModelSafeRoutingInfoToString(session_context_->routing_info())
            << "and purpose " << job->purpose();
   SyncSession session(session_context_, this, job->source_info());
   bool premature_exit = !syncer_->SyncShare(&session,
                                             job->start_step(),
                                             job->end_step());
   SDVLOG(2) << "Done SyncShare, returned: " << premature_exit;
 
   bool success = FinishSyncSessionJob(job.get(),
                                       premature_exit,
                                       &session);
 
   if (IsSyncingCurrentlySilenced()) {
     SDVLOG(2) << "We are currently throttled; scheduling Unthrottle.";
     // If we're here, it's because |job| was silenced until a server specified
     // time. (Note, it had to be |job|, because DecideOnJob would not permit
     // any job through while in WaitInterval::THROTTLED).
-    scoped_ptr<SyncSessionJob> clone = job->Clone();
-    if (clone->purpose() == SyncSessionJob::NUDGE)
-      pending_nudge_ = clone.get();
-    else if (clone->purpose() == SyncSessionJob::CONFIGURATION)
-      wait_interval_->pending_configure_job = clone.get();
+    if (job->purpose() == SyncSessionJob::NUDGE)
+      pending_nudge_job_ = job.Pass();
+    else if (job->purpose() == SyncSessionJob::CONFIGURATION)
+      pending_configure_job_ = job.Pass();
     else
       NOTREACHED();
 
-    RestartWaiting(clone.Pass());
+    RestartWaiting();
     return success;
   }
 
   if (!success)
     ScheduleNextSync(job.Pass(), &session);
 
   return success;
 }
 
+bool SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {

[tim (not reviewing), 2013/04/04 00:22:11]

Digging around, it looks like these Do*SSJ methods exist solely because you
don't want to hand ownership over to the posted task (which makes sense). Is
that right?  

Couldn't we have a single function taking a Purpose be responsible for Pass()ing
the right job to DoSyncSessionJobImpl? Or DoSyncSessionJobImpl itself? The code
as written made me think there were significant differences between the two and
the fact that it was all about ownership wasn't very clear.

[rlarocque, 2013/04/04 00:59:48]

Your suggestion would work for now, but it wouldn't work once we remove
SyncSessionJob.  These functions will diverge significantly in the near future.

+  return 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;
   }
 
   if (mode_ == CONFIGURATION_MODE) {
     SDVLOG(2) << "Not running poll in configuration mode.";
     return false;
   }
 
   // TODO(rlarocque): Refactor decision-making logic common to all types
   // of jobs into a shared function.
 
   if (session_context_->connection_manager()->HasInvalidAuthToken()) {
     SDVLOG(2) << "Not running poll because auth token is invalid.";
     return false;
   }
 
   return true;
 }
 
-void SyncSchedulerImpl::DoPollSyncSessionJob(scoped_ptr<SyncSessionJob> job) {
-  DCHECK_EQ(job->purpose(), SyncSessionJob::POLL);
+void SyncSchedulerImpl::DoPollSyncSessionJob() {
+  ModelSafeRoutingInfo r;
+  ModelTypeInvalidationMap invalidation_map =
+      ModelSafeRoutingInfoToInvalidationMap(r, std::string());
+  SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, invalidation_map);
+  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(SyncSessionJob::POLL,
+                                                    TimeTicks::Now(),
+                                                    info,
+                                                    ConfigurationParams()));
 
   base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
 
   if (!ShouldPoll())
     return;
 
   DVLOG(2) << "Polling with routes "
            << ModelSafeRoutingInfoToString(session_context_->routing_info());
   SyncSession session(session_context_, this, job->source_info());
   bool premature_exit = !syncer_->SyncShare(&session,
                                             job->start_step(),
                                             job->end_step());
   SDVLOG(2) << "Done SyncShare, returned: " << premature_exit;
 
   FinishSyncSessionJob(job.get(), premature_exit, &session);
 
   if (IsSyncingCurrentlySilenced()) {
-    // This will start the countdown to unthrottle.  Other kinds of jobs would
-    // schedule themselves as the post-unthrottle canary.  A poll job is not
-    // that urgent, so it does not get to be the canary.  We still need to start
-    // the timer regardless.  Otherwise there could be no one to clear the
-    // WaitInterval when the throttling expires.
-    RestartWaiting(scoped_ptr<SyncSessionJob>());
+    // Normally we would only call RestartWaiting() if we had a
+    // pending_nudge_job_ or pending_configure_job_ set.  In this case, it's
+    // possible that neither is set.  We create the wait interval anyway because
+    // we need it to make sure we get unthrottled on time.
+    RestartWaiting();
   }
 }
 
 void SyncSchedulerImpl::UpdateNudgeTimeRecords(const SyncSourceInfo& info) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   // We are interested in recording time between local nudges for datatypes.
   // TODO(tim): Consider tracking LOCAL_NOTIFICATION as well.
   if (info.updates_source != GetUpdatesCallerInfo::LOCAL)
     return;
@@ skipping 62 matching lines @@
     // 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_);
+    DCHECK(!pending_nudge_job_);
 
-    scoped_ptr<SyncSessionJob> new_job = finished_job->Clone();
-    pending_nudge_ = new_job.get();
-    RestartWaiting(new_job.Pass());
+    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);
   }
 }
 
 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::RestartWaiting(scoped_ptr<SyncSessionJob> job) {
+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_.Reset(base::Bind(&SyncSchedulerImpl::Unthrottle,
-                                     weak_ptr_factory_.GetWeakPtr(),
-                                     base::Passed(&job)));
+                                     weak_ptr_factory_.GetWeakPtr()));
 
   } else {
-    pending_wakeup_.Reset(base::Bind(&SyncSchedulerImpl::DoCanaryJob,
-                                     weak_ptr_factory_.GetWeakPtr(),
-                                     base::Passed(&job)));
+    pending_wakeup_.Reset(base::Bind(&SyncSchedulerImpl::TryCanaryJob,
+                                     weak_ptr_factory_.GetWeakPtr()));
   }
   wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
                               pending_wakeup_.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);
-  scoped_ptr<SyncSessionJob> new_job(old_job->Clone());
-  new_job->set_scheduled_start(TimeTicks::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());
-    wait_interval_->pending_configure_job = new_job.get();
+    DCHECK(!pending_configure_job_);
+    pending_configure_job_ = old_job.Pass();
   } else {
-    // We are not in configuration mode. So wait_interval's pending job
-    // should be null.
-    DCHECK(wait_interval_->pending_configure_job == NULL);
-    DCHECK(!pending_nudge_);
-    pending_nudge_ = new_job.get();
+    // We're not in configure mode so we should not have a configure job.
+    DCHECK(!pending_configure_job_);
+    DCHECK(!pending_nudge_job_);
+    pending_nudge_job_ = old_job.Pass();
   }
 
-  RestartWaiting(new_job.Pass());
+  RestartWaiting();
 }
 
 void SyncSchedulerImpl::RequestStop(const base::Closure& callback) {
   syncer_->RequestEarlyExit();  // Safe to call from any thread.
   DCHECK(weak_handle_this_.IsInitialized());
   SDVLOG(3) << "Posting StopImpl";
   weak_handle_this_.Call(FROM_HERE,
                          &SyncSchedulerImpl::StopImpl,
                          callback);
 }
 
 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_nudge_ = NULL;
-  unscheduled_nudge_storage_.reset();
   pending_wakeup_.Cancel();

[tim (not reviewing), 2013/04/04 00:22:11]

Can we rename pending_wakeup_ to pending_wakeup_event_? I was confused for a
moment that we owned the job twice, once with pending_[nudge,configure]_job_ and
once with pending_wakeup_.  'event' would make it clearer that it is not a job.

[rlarocque, 2013/04/04 00:59:48]

Done.

+  pending_nudge_job_.reset();
+  pending_configure_job_.reset();
   if (started_) {
     started_ = false;
   }
   if (!callback.is_null())
     callback.Run();
 }
 
-void SyncSchedulerImpl::DoCanaryJob(scoped_ptr<SyncSessionJob> to_be_canary) {
+// This is the only place where we invoke DoSyncSessionJob with canary
+// privileges.  Everyone else should use NORMAL_PRIORITY.
+void SyncSchedulerImpl::TryCanaryJob() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SDVLOG(2) << "Do canary job";
 
-  // This is the only place where we invoke DoSyncSessionJob with canary
-  // privileges.  Everyone else should use NORMAL_PRIORITY.
-  DoSyncSessionJob(to_be_canary.Pass(), CANARY_PRIORITY);
-}
-
-scoped_ptr<SyncSessionJob> SyncSchedulerImpl::TakePendingJobForCurrentMode() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  // If we find a scheduled pending_ job, abandon the old one and return a
-  // a clone.  If unscheduled, just hand over ownership.
-  scoped_ptr<SyncSessionJob> candidate;
-  if (mode_ == CONFIGURATION_MODE && wait_interval_.get()
-      && wait_interval_->pending_configure_job) {
-    SDVLOG(2) << "Found pending configure job";
-    candidate =
-        wait_interval_->pending_configure_job->Clone().Pass();
-    wait_interval_->pending_configure_job = candidate.get();
-  } else if (mode_ == NORMAL_MODE && pending_nudge_) {
-    SDVLOG(2) << "Found pending nudge job";
-    candidate = pending_nudge_->Clone();
-    pending_nudge_ = candidate.get();
-    unscheduled_nudge_storage_.reset();
+  if (mode_ == CONFIGURATION_MODE && pending_configure_job_) {

[tim (not reviewing), 2013/04/04 18:07:30]

There's an implicit point here that canary jobs have priorities beyond the
JobPriority. If a nudge sends us to backoff and we then switch to config mode,
we have to run the config job first, even though there was a nudge canary. I
think this is covered by tests, but it's not that obvious looking at the code
unless you consider this function along with the fact that once we go to config
mode we can never leave until the job is run.  A comment would be a good idea.

+    SDVLOG(2) << "Found pending configure job; will run as canary";
+    DoConfigurationSyncSessionJob(CANARY_PRIORITY);
+  } else if (mode_ == NORMAL_MODE && pending_nudge_job_) {
+    SDVLOG(2) << "Found pending nudge job; will run as canary";
+    DoNudgeSyncSessionJob(CANARY_PRIORITY);
+  } else {
+    SDVLOG(2) << "Found no work to do; will not run a canary";
   }
-  // If we took a job and there's a wait interval, we took the pending canary.
-  if (candidate && wait_interval_)
-    wait_interval_->timer.Stop();
-  return candidate.Pass();
 }
 
 void SyncSchedulerImpl::PollTimerCallback() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  ModelSafeRoutingInfo r;
-  ModelTypeInvalidationMap invalidation_map =
-      ModelSafeRoutingInfoToInvalidationMap(r, std::string());
-  SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, invalidation_map);
-  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(SyncSessionJob::POLL,
-                                                    TimeTicks::Now(),
-                                                    info,
-                                                    ConfigurationParams()));
   if (no_scheduling_allowed_) {
     // The no_scheduling_allowed_ flag is set by a function-scoped AutoReset in
     // functions that are called only on the sync thread.  This function is also
     // called only on the sync thread, and only when it is posted by an expiring
     // timer.  If we find that no_scheduling_allowed_ is set here, then
     // something is very wrong.  Maybe someone mistakenly called us directly, or
     // mishandled the book-keeping for no_scheduling_allowed_.
     NOTREACHED() << "Illegal to schedule job while session in progress.";
     return;
   }
 
-  DoPollSyncSessionJob(job.Pass());
+  DoPollSyncSessionJob();
 }
 
-void SyncSchedulerImpl::Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary) {
+void SyncSchedulerImpl::Unthrottle() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
-  DCHECK(!to_be_canary.get() || pending_nudge_ == to_be_canary.get() ||
-         wait_interval_->pending_configure_job == to_be_canary.get());
-  SDVLOG(2) << "Unthrottled " << (to_be_canary.get() ? "with " : "without ")
-           << "canary.";
 
   // We're no longer throttled, so clear the wait interval.
   wait_interval_.reset();
   NotifyRetryTime(base::Time());
 
   // We treat this as a 'canary' in the sense that it was originally scheduled
   // to run some time ago, failed, and we now want to retry, versus a job that
   // was just created (e.g via ScheduleNudgeImpl). The main implication is
   // that we're careful to update routing info (etc) with such potentially
   // stale canary jobs.
-  if (to_be_canary.get()) {
-    DoCanaryJob(to_be_canary.Pass());
-  } else {
-    DCHECK(!unscheduled_nudge_storage_.get());
-  }
+  TryCanaryJob();
 }
 
 void SyncSchedulerImpl::Notify(SyncEngineEvent::EventCause cause) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   session_context_->NotifyListeners(SyncEngineEvent(cause));
 }
 
 void SyncSchedulerImpl::NotifyRetryTime(base::Time retry_time) {
   SyncEngineEvent event(SyncEngineEvent::RETRY_TIME_CHANGED);
   event.retry_time = retry_time;
@@ skipping 78 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // namespace syncer