sync: make scheduling logic and job ownership more obvious.

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"
 #include "base/bind.h"
+#include "base/bind_helpers.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "sync/engine/backoff_delay_provider.h"
 #include "sync/engine/syncer.h"
 #include "sync/engine/throttled_data_type_tracker.h"
 #include "sync/protocol/proto_enum_conversions.h"
 #include "sync/protocol/sync.pb.h"
 #include "sync/util/data_type_histogram.h"
@@ skipping 54 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) {
+      had_nudge(false),
+      pending_configure_job(NULL) {

[akalin, 2012/09/25 22:37:24]

three different styles for empty function bodies here!

[tim (not reviewing), 2012/10/08 00:20:03]

Done.

 }
 
+SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
+    : mode(mode), had_nudge(false), length(length),
+      pending_configure_job(NULL) { }
+
 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);
   }
   NOTREACHED();
   return "";
 }
 
-SyncSchedulerImpl::SyncSessionJob::SyncSessionJob()
-    : purpose(UNKNOWN),
-      is_canary_job(false) {
-}
-
-SyncSchedulerImpl::SyncSessionJob::~SyncSessionJob() {}
-
-SyncSchedulerImpl::SyncSessionJob::SyncSessionJob(SyncSessionJobPurpose purpose,
-    base::TimeTicks start,
-    linked_ptr<sessions::SyncSession> session,
-    bool is_canary_job,
-    const ConfigurationParams& config_params,
-    const tracked_objects::Location& from_here)
-    : purpose(purpose),
-      scheduled_start(start),
-      session(session),
-      is_canary_job(is_canary_job),
-      config_params(config_params),
-      from_here(from_here) {
-}
-
-const char* SyncSchedulerImpl::SyncSessionJob::GetPurposeString(
-    SyncSchedulerImpl::SyncSessionJob::SyncSessionJobPurpose purpose) {
-  switch (purpose) {
-    ENUM_CASE(UNKNOWN);
-    ENUM_CASE(POLL);
-    ENUM_CASE(NUDGE);
-    ENUM_CASE(CONFIGURATION);
-  }
-  NOTREACHED();
-  return "";
-}
-
 GetUpdatesCallerInfo::GetUpdatesSource GetUpdatesFromNudgeSource(
     NudgeSource source) {
   switch (source) {
     case NUDGE_SOURCE_NOTIFICATION:
       return GetUpdatesCallerInfo::NOTIFICATION;
     case NUDGE_SOURCE_LOCAL:
       return GetUpdatesCallerInfo::LOCAL;
     case NUDGE_SOURCE_CONTINUATION:
       return GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION;
     case NUDGE_SOURCE_LOCAL_REFRESH:
       return GetUpdatesCallerInfo::DATATYPE_REFRESH;
     case NUDGE_SOURCE_UNKNOWN:
       return GetUpdatesCallerInfo::UNKNOWN;
     default:
       NOTREACHED();
       return GetUpdatesCallerInfo::UNKNOWN;
   }
 }
 
-SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
-    : mode(mode), had_nudge(false), length(length) { }
-
 // Helper macros to log with the syncer thread name; useful when there
 // are multiple syncer threads involved.
 
 #define SLOG(severity) LOG(severity) << name_ << ": "
 
 #define SDVLOG(verbose_level) DVLOG(verbose_level) << name_ << ": "
 
 #define SDVLOG_LOC(from_here, verbose_level)             \
   DVLOG_LOC(from_here, verbose_level) << name_ << ": "
 
@@ skipping 30 matching lines @@
       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.
       connection_code_(HttpResponse::SERVER_CONNECTION_OK),
+      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 16 matching lines @@
 void SyncSchedulerImpl::OnConnectionStatusChange() {
   if (HttpResponse::CONNECTION_UNAVAILABLE  == connection_code_) {
     // Optimistically assume that the connection is fixed and try
     // connecting.
     OnServerConnectionErrorFixed();
   }
 }
 
 void SyncSchedulerImpl::OnServerConnectionErrorFixed() {
   connection_code_ = HttpResponse::SERVER_CONNECTION_OK;
+  // There could be a pending nudge or configuration job in several cases:
+  //
+  // 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;
+
   PostTask(FROM_HERE, "DoCanaryJob",
            base::Bind(&SyncSchedulerImpl::DoCanaryJob,
-                      weak_ptr_factory_.GetWeakPtr()));
-
+                      weak_ptr_factory_.GetWeakPtr(),
+                      base::Passed(&pending)));
 }
 
 void SyncSchedulerImpl::UpdateServerConnectionManagerStatus(
     HttpResponse::ServerConnectionCode code) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "New server connection code: "
             << HttpResponse::GetServerConnectionCodeString(code);
 
   connection_code_ = code;
 }
@@ skipping 15 matching lines @@
   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.get());
+      DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
     }
 
-    DoPendingJobIfPossible(false);
+    scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
+    if (pending.get()) {
+      // TODO(tim): We should be able to remove this...
+      scoped_ptr<SyncSession> session(CreateSyncSession(
+          pending->session()->source()));
+      // Also the routing info might have been changed since we cached the
+      // pending nudge. Update it by coalescing to the latest.
+      pending->mutable_session()->Coalesce(*(session));
+      SDVLOG(2) << "Executing pending job. Good luck!";
+      DoSyncSessionJob(pending.Pass());
+    }
   }
 }
 
 void SyncSchedulerImpl::SendInitialSnapshot() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   scoped_ptr<SyncSession> dummy(new SyncSession(session_context_, this,
       SyncSourceInfo(), ModelSafeRoutingInfo(),
       std::vector<ModelSafeWorker*>()));
   SyncEngineEvent event(SyncEngineEvent::STATUS_CHANGED);
   event.snapshot = dummy->TakeSnapshot();
@@ skipping 23 matching lines @@
 bool SyncSchedulerImpl::ScheduleConfiguration(
     const ConfigurationParams& params) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
   DCHECK_EQ(CONFIGURATION_MODE, mode_);
   DCHECK(!params.ready_task.is_null());
   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.get());
+  DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
 
   ModelSafeRoutingInfo restricted_routes;
   BuildModelSafeParams(params.types_to_download,
                        params.routing_info,
                        &restricted_routes);
   session_context_->set_routing_info(params.routing_info);
 
   // Only reconfigure if we have types to download.
   if (!params.types_to_download.Empty()) {
     DCHECK(!restricted_routes.empty());
-    linked_ptr<SyncSession> session(new SyncSession(
+    scoped_ptr<SyncSession> session(new SyncSession(
         session_context_,
         this,
         SyncSourceInfo(params.source,
                        ModelSafeRoutingInfoToStateMap(
                            restricted_routes,
                            std::string())),
         restricted_routes,
         session_context_->workers()));
-    SyncSessionJob job(SyncSessionJob::CONFIGURATION,
-                       TimeTicks::Now(),
-                       session,
-                       false,
-                       params,
-                       FROM_HERE);
-    DoSyncSessionJob(job);
+    scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
+        SyncSessionJob::CONFIGURATION,
+        TimeTicks::Now(),
+        session.Pass(),
+        params,
+        FROM_HERE));
+    bool succeeded = DoSyncSessionJob(job.Pass());
 
     // If we failed, the job would have been saved as the pending configure
     // job and a wait interval would have been set.
-    if (!session->Succeeded()) {
-      DCHECK(wait_interval_.get() &&
-             wait_interval_->pending_configure_job.get());
+    if (!succeeded) {
+      DCHECK(wait_interval_.get() && wait_interval_->pending_configure_job);
       return false;
     }
   } else {
     SDVLOG(2) << "No change in routing info, calling ready task directly.";
     params.ready_task.Run();
   }
 
   return true;
 }
 
 SyncSchedulerImpl::JobProcessDecision
-SyncSchedulerImpl::DecideWhileInWaitInterval(
-    const SyncSessionJob& job) {
+SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob* job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(wait_interval_.get());
 
   SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
             << WaitInterval::GetModeString(wait_interval_->mode)
             << (wait_interval_->had_nudge ? " (had nudge)" : "")
-            << (job.is_canary_job ? " (canary)" : "");
+            << (job->is_canary() ? " (canary)" : "");
 
-  if (job.purpose == SyncSessionJob::POLL)
+  if (job->purpose() == SyncSessionJob::POLL)
     return DROP;
 
-  DCHECK(job.purpose == SyncSessionJob::NUDGE ||
-         job.purpose == SyncSessionJob::CONFIGURATION);
+  // 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 (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 (!job.is_canary_job)
+    if (!job->is_canary())
       return wait_interval_->had_nudge ? DROP : CONTINUE;
     else // We are here because timer ran out. So retry.
       return CONTINUE;
   }
-  return job.is_canary_job ? CONTINUE : SAVE;
+  return job->is_canary() ? CONTINUE : SAVE;
 }
 
 SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
-    const SyncSessionJob& job) {
+    const SyncSessionJob* job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   // See if our type is throttled.
   ModelTypeSet throttled_types =
       session_context_->throttled_data_type_tracker()->GetThrottledTypes();
-  if (job.purpose == SyncSessionJob::NUDGE &&
-      job.session->source().updates_source == GetUpdatesCallerInfo::LOCAL) {
+  if (job->purpose() == SyncSessionJob::NUDGE &&
+      job->session()->source().updates_source == GetUpdatesCallerInfo::LOCAL) {
     ModelTypeSet requested_types;
     for (ModelTypeStateMap::const_iterator i =
-         job.session->source().types.begin();
-         i != job.session->source().types.end();
+         job->session()->source().types.begin();
+         i != job->session()->source().types.end();
          ++i) {
       requested_types.Put(i->first);
     }
 
+    // If all types are throttled, do not CONTINUE.  Today, we don't treat
+    // a per-datatype "unthrottle" event as something that should force a
+    // canary job. For this reason, there's no good time to reschedule this job
+    // to run -- we'll lazily wait for an independent event to trigger a sync.
+    // Note that there may already be such an event if we're in a WaitInterval,
+    // so we can retry it then.
     if (!requested_types.Empty() && throttled_types.HasAll(requested_types))

[rlarocque, 2012/09/24 21:15:14]

This isn't related to your change, but your comment made me think of this
scenario:

Imagine bookmarks are throttled and we receive a bookmarks nudge.  We will save
the nudge here.

Then an autofill nudge arrives.  We merged it with the saved bookmarks nudge.

When the job completes successfully, we forget that the bookmarks nudge ever
happened.  

[tim (not reviewing), 2012/10/08 00:20:03]

Can you define 'forget'? The job would have accounted for both bookmarks and
autofill, and successful job completion implies end of the wait interval.   

[rlarocque, 2012/10/08 19:18:04]

When we calculate the items to commit in GetCommitIds, we'll filter out the
throttled types.  The job would not have accounted for bookmarks.

[tim (not reviewing), 2012/10/11 17:35:14]

Hm, I see. Good point.  I filed crbug.com/155296 to track this.

       return SAVE;
   }
 
   if (wait_interval_.get())
     return DecideWhileInWaitInterval(job);
 
   if (mode_ == CONFIGURATION_MODE) {
-    if (job.purpose == SyncSessionJob::NUDGE)
-      return SAVE;
-    else if (job.purpose == SyncSessionJob::CONFIGURATION)
+    if (job->purpose() == SyncSessionJob::NUDGE)
+      return SAVE; // Running requires a mode switch.
+    else if (job->purpose() == SyncSessionJob::CONFIGURATION)
       return CONTINUE;
     else
       return DROP;
   }
 
   // We are in normal mode.
   DCHECK_EQ(mode_, NORMAL_MODE);
-  DCHECK_NE(job.purpose, SyncSessionJob::CONFIGURATION);
+  DCHECK_NE(job->purpose(), SyncSessionJob::CONFIGURATION);
 
   // Note about some subtle scheduling semantics.
   //
   // It's possible at this point that |job| is known to be unnecessary, and
   // dropping it would be perfectly safe and correct. Consider
   //
   // 1) |job| is a POLL with a |scheduled_start| time that is less than
   //    the time that the last successful all-datatype NUDGE completed.
   //
   // 2) |job| is a NUDGE (for any combination of types) with a
@@ skipping 22 matching lines @@
   //  * It's not strictly "impossible", but it would be reentrant and hence
   //  illegal. e.g. scheduling a job and re-entering the SyncScheduler is NOT a
   //  legal side effect of any of the work being done as part of a sync cycle.
   //  See |no_scheduling_allowed_| for details.
 
   // 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.";
-  return job.purpose == SyncSessionJob::NUDGE ? SAVE : DROP;
+  // Running the job would require updated auth, so we can't honour
+  // job.scheduled_start().
+  return job->purpose() == SyncSessionJob::NUDGE ? SAVE : DROP;
 }
 
-void SyncSchedulerImpl::InitOrCoalescePendingJob(const SyncSessionJob& job) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  DCHECK(job.purpose != SyncSessionJob::CONFIGURATION);
-  if (pending_nudge_.get() == NULL) {
-    SDVLOG(2) << "Creating a pending nudge job";
-    SyncSession* s = job.session.get();
-
-    // Get a fresh session with similar configuration as before (resets
-    // StatusController).
-    scoped_ptr<SyncSession> session(new SyncSession(s->context(),
-        s->delegate(), s->source(), s->routing_info(), s->workers()));
-
-    SyncSessionJob new_job(SyncSessionJob::NUDGE, job.scheduled_start,
-        make_linked_ptr(session.release()), false,
-        ConfigurationParams(), job.from_here);
-    pending_nudge_.reset(new SyncSessionJob(new_job));
-    return;
-  }
-
-  SDVLOG(2) << "Coalescing a pending nudge";
-  pending_nudge_->session->Coalesce(*(job.session.get()));
-  pending_nudge_->scheduled_start = job.scheduled_start;
-
-  // Unfortunately the nudge location cannot be modified. So it stores the
-  // location of the first caller.
-}
-
-bool SyncSchedulerImpl::ShouldRunJob(const SyncSessionJob& job) {
+bool SyncSchedulerImpl::ShouldRunJobSaveIfNecessary(SyncSessionJob* job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(started_);
 
   JobProcessDecision decision = DecideOnJob(job);
   SDVLOG(2) << "Should run "
-            << SyncSessionJob::GetPurposeString(job.purpose)
-            << " job in mode " << GetModeString(mode_)
+            << SyncSessionJob::GetPurposeString(job->purpose())
+            << " job " << job->session()
+            << " in mode " << GetModeString(mode_)
             << ": " << GetDecisionString(decision);
   if (decision != SAVE)

[rlarocque, 2012/09/24 21:15:14]

These lines have always bugged me.  It's not clear to the casual reader that the
decision is necessarily one of {DROP,SAVE,CONTINUE}.  Since you're refactoring
in this area anyway, could you change it so that all three cases are explicitly
mentioned?  ie.

if (decision == DROP)
  return false;

if (decision == CONTINUE)
  return true;

DCHECK_EQ(decision, SAVE)

[tim (not reviewing), 2012/10/08 00:20:03]

Done.

     return decision == CONTINUE;
 
-  DCHECK(job.purpose == SyncSessionJob::NUDGE || job.purpose ==
-      SyncSessionJob::CONFIGURATION);
+  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

[rlarocque, 2012/09/24 21:15:14]

This is a regression, right?  How/Why did this happen?

[tim (not reviewing), 2012/10/08 00:20:03]

It is, but I suspect it has been this way for quite some time. This is one of
the reasons I renamed ShouldRunJob to point out that it's actually doing more
than just returning true / false.  I'd like to keep to no functional change with
my CL here if that sounds okay, but I plan on addressing this soon after.

[rlarocque, 2012/10/08 19:18:04]

Sounds good.  I was confused for a bit there; I didn't realize that this code is
only for saved jobs.

+    // 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_->mutable_session()->Coalesce(*(job->session()));
+    pending_nudge_->set_scheduled_start(job->scheduled_start());

[rlarocque, 2012/09/24 21:15:14]

When a user is actively using their browser, I expect there will be at least one
sessions nudge queued up most of the time.  If that's the case, then this line
would have the effect of making the syncer only commit in batches every ~10
seconds.

[tim (not reviewing), 2012/10/08 00:20:03]

Hmm. Afaik, this set_scheduled_start call doesn't even have a real effect now
that we removed the freshness condition, so I actually removed it altogether
now... lets chat about this in person?

+    return false;
+  }
 
-  SaveJob(job);
+  scoped_ptr<SyncSessionJob> job_to_save = job->CloneAndAbandon();

[rlarocque, 2012/09/24 21:15:14]

This makes me think ShouldRunJobSaveIfNecessary() should return a
SyncSessionJob*.  That way you could claim that the function always takes
ownership of the job passed in and returns an updated job.  The returned job may
or may not be based on the one that was passed in.  

That makes more sense to me than the current semantics where the job may be
corrupted, and the function uses a boolean flag that implies the job *may* have
been corrupted if the return value was false, but is still good if the return
value was true.

[tim (not reviewing), 2012/10/08 00:20:03]

Well, note the intended contract here is simpler than you describe - a false
return means the caller should assume the job is not fit for scheduling (session
== null).

If we do what you propose, I think we'd have a similar (I'd argue, worse,
problem): The returned pointer would *sometimes* require external ownership (in
the case that we Coalesced), but sometimes not (in the case that it was
rescheduled as wait interval). 

Maybe making it more clear that a false return means the job should be assumed
to no longer have a valid session and is now owned elsewhere would address the
confusion you point out?

The only alternative I can see is that in the Coalesce case, we'd
CloneAndAbandon() the pending_nudge_, Coalesce with the passed-in job, and
invoke ScheduleSyncSessionJob on the product. Then we'd return NULL (?) in all
cases we return false, and otherwise return a pointer to the job that was passed
in and already owned by the caller? That seems a bit more complicated to me, and
introduces unnecessary posts... wdyt?
 

+  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_->CloneAndAbandon();

[rlarocque, 2012/09/24 21:15:14]

What effect do these two lines have?

[tim (not reviewing), 2012/10/08 00:20:03]

By "pre-empt" I mean cancel the pending nudge job by calling CloneAndAbandon, so
that we can give priority to the config job.  The first line abandons the old
nudge and stores the clone in unscheduled_nudge_storage_ (no longer a canary). 
The second line upholds the pending_nudge_ invariant, that it always points to
the one and only pending nudge (whether it is stored on the ML or u_n_s). 

+        pending_nudge_ = unscheduled_nudge_storage_.get();
+      }
+      wait_interval_->pending_configure_job = job_to_save.get();
+    }
+    wait_interval_->length =

[rlarocque, 2012/09/24 21:15:14]

Could this end up being zero or negative?  What effect would that have?

[tim (not reviewing), 2012/10/08 00:20:03]

Hmph. Good catch - I had fixed this since I hit a failure in a test, but somehow
undid the fix.  Fixed + added dcheck in RestartWaiting.

+        wait_interval_->timer.desired_run_time() - TimeTicks::Now();
+    RestartWaiting(job_to_save.Pass());
+    return false;
+  }
+
+  // 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();
   return false;
 }
 
-void SyncSchedulerImpl::SaveJob(const SyncSessionJob& job) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  if (job.purpose == SyncSessionJob::NUDGE) {
-    SDVLOG(2) << "Saving a nudge job";
-    InitOrCoalescePendingJob(job);
-  } else if (job.purpose == SyncSessionJob::CONFIGURATION){
-    SDVLOG(2) << "Saving a configuration job";
-    DCHECK(wait_interval_.get());
-    DCHECK(mode_ == CONFIGURATION_MODE);
-
-    // Config params should always get set.
-    DCHECK(!job.config_params.ready_task.is_null());
-    SyncSession* old = job.session.get();
-    SyncSession* s(new SyncSession(session_context_, this, old->source(),
-                                   old->routing_info(), old->workers()));
-    SyncSessionJob new_job(job.purpose,
-                           TimeTicks::Now(),
-                           make_linked_ptr(s),
-                           false,
-                           job.config_params,
-                           job.from_here);
-    wait_interval_->pending_configure_job.reset(new SyncSessionJob(new_job));
-  } // drop the rest.
-  // TODO(sync): Is it okay to drop the rest?  It's weird that
-  // SaveJob() only does what it says sometimes.  (See
-  // http://crbug.com/90868.)
-}
-
 // 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& 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 " << delay.InMilliseconds() << " ms, "
       << "source " << GetNudgeSourceString(source) << ", "
       << "types " << ModelTypeSetToString(types);
 
   ModelTypeStateMap type_state_map =
       ModelTypeSetToStateMap(types, std::string());
   SyncSchedulerImpl::ScheduleNudgeImpl(delay,
                                        GetUpdatesFromNudgeSource(source),
                                        type_state_map,
-                                       false,
                                        nudge_location);
 }
 
 void SyncSchedulerImpl::ScheduleNudgeWithStatesAsync(
     const TimeDelta& delay,
     NudgeSource source, const ModelTypeStateMap& type_state_map,
     const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG_LOC(nudge_location, 2)
       << "Nudge scheduled with delay " << delay.InMilliseconds() << " ms, "
       << "source " << GetNudgeSourceString(source) << ", "
       << "payloads "
       << ModelTypeStateMapToString(type_state_map);
 
   SyncSchedulerImpl::ScheduleNudgeImpl(delay,
                                        GetUpdatesFromNudgeSource(source),
                                        type_state_map,
-                                       false,
                                        nudge_location);
 }
 
 void SyncSchedulerImpl::ScheduleNudgeImpl(
     const TimeDelta& delay,
     GetUpdatesCallerInfo::GetUpdatesSource source,
     const ModelTypeStateMap& type_state_map,
-    bool is_canary_job, const tracked_objects::Location& nudge_location) {
+    const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(!type_state_map.empty()) << "Nudge scheduled for no types!";
 
   SDVLOG_LOC(nudge_location, 2)
       << "In ScheduleNudgeImpl with delay "
       << delay.InMilliseconds() << " ms, "
       << "source " << GetUpdatesSourceString(source) << ", "
       << "payloads "
-      << ModelTypeStateMapToString(type_state_map)
-      << (is_canary_job ? " (canary)" : "");
+      << ModelTypeStateMapToString(type_state_map);
 
   SyncSourceInfo info(source, type_state_map);
   UpdateNudgeTimeRecords(info);
 
-  SyncSession* session(CreateSyncSession(info));
-  SyncSessionJob job(SyncSessionJob::NUDGE, TimeTicks::Now() + delay,
-                     make_linked_ptr(session), is_canary_job,
-                     ConfigurationParams(), nudge_location);
+  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
+      SyncSessionJob::NUDGE,
+      TimeTicks::Now() + delay,
+      CreateSyncSession(info).Pass(),
+      ConfigurationParams(),
+      nudge_location));
 
-  session = NULL;
-  if (!ShouldRunJob(job))
+  if (!ShouldRunJobSaveIfNecessary(job.get()))
     return;
 
-  if (pending_nudge_.get()) {
-    if (IsBackingOff() && delay > TimeDelta::FromSeconds(1)) {
-      SDVLOG(2) << "Dropping the nudge because we are in backoff";
-      return;
-    }
-
-    SDVLOG(2) << "Coalescing pending nudge";
-    pending_nudge_->session->Coalesce(*(job.session.get()));
-
+  if (pending_nudge_) {
     SDVLOG(2) << "Rescheduling pending nudge";
-    SyncSession* s = pending_nudge_->session.get();
-    job.session.reset(new SyncSession(s->context(), s->delegate(),
-        s->source(), s->routing_info(), s->workers()));
-
-    // Choose the start time as the earliest of the 2.
-    job.scheduled_start = std::min(job.scheduled_start,
-                                   pending_nudge_->scheduled_start);
-    pending_nudge_.reset();
+    pending_nudge_->mutable_session()->Coalesce(*(job->session()));
+    // 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_->CloneAndAbandon();
+    unscheduled_nudge_storage_.reset();
+    pending_nudge_ = NULL;
   }
 
   // TODO(zea): Consider adding separate throttling/backoff for datatype
   // refresh requests.
-  ScheduleSyncSessionJob(job);
+  ScheduleSyncSessionJob(job.Pass());
 }
 
 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 "";
 }
 
-// static
-void SyncSchedulerImpl::SetSyncerStepsForPurpose(
-    SyncSessionJob::SyncSessionJobPurpose purpose,
-    SyncerStep* start,
-    SyncerStep* end) {
-  switch (purpose) {
-    case SyncSessionJob::CONFIGURATION:
-      *start = DOWNLOAD_UPDATES;
-      *end = APPLY_UPDATES;
-      return;
-    case SyncSessionJob::NUDGE:
-    case SyncSessionJob::POLL:
-      *start = SYNCER_BEGIN;
-      *end = SYNCER_END;
-      return;
-    default:
-      NOTREACHED();
-      *start = SYNCER_END;
-      *end = SYNCER_END;
-      return;
-  }
-}
-
 void SyncSchedulerImpl::PostTask(
     const tracked_objects::Location& from_here,
     const char* name, const base::Closure& task) {
   SDVLOG_LOC(from_here, 3) << "Posting " << name << " task";
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   if (!started_) {
     SDVLOG(1) << "Not posting task as scheduler is stopped.";
     return;
   }
   sync_loop_->PostTask(from_here, task);
 }
 
 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;
   }
   sync_loop_->PostDelayedTask(from_here, task, delay);
 }
 
-void SyncSchedulerImpl::ScheduleSyncSessionJob(const SyncSessionJob& job) {
+void SyncSchedulerImpl::ScheduleSyncSessionJob(
+    scoped_ptr<SyncSessionJob> job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   if (no_scheduling_allowed_) {
     NOTREACHED() << "Illegal to schedule job while session in progress.";
     return;
   }
 
-  TimeDelta delay = job.scheduled_start - TimeTicks::Now();
+  TimeDelta delay = job->scheduled_start() - TimeTicks::Now();
   if (delay < TimeDelta::FromMilliseconds(0))
     delay = TimeDelta::FromMilliseconds(0);
-  SDVLOG_LOC(job.from_here, 2)
+  SDVLOG_LOC(job->from_location(), 2)
       << "In ScheduleSyncSessionJob with "
-      << SyncSessionJob::GetPurposeString(job.purpose)
+      << SyncSessionJob::GetPurposeString(job->purpose())
       << " job and " << delay.InMilliseconds() << " ms delay";
 
-  DCHECK(job.purpose == SyncSessionJob::NUDGE ||
-         job.purpose == SyncSessionJob::POLL);
-  if (job.purpose == SyncSessionJob::NUDGE) {
-    SDVLOG_LOC(job.from_here, 2) << "Resetting pending_nudge";
-    DCHECK(!pending_nudge_.get() || pending_nudge_->session.get() ==
-           job.session);
-    pending_nudge_.reset(new SyncSessionJob(job));
+  DCHECK(job->purpose() == SyncSessionJob::NUDGE ||
+         job->purpose() == SyncSessionJob::POLL);
+  if (job->purpose() == SyncSessionJob::NUDGE) {
+    SDVLOG_LOC(job->from_location(), 2) << "Resetting pending_nudge to ";
+    DCHECK(!pending_nudge_ || pending_nudge_->session() ==
+           job->session());
+    pending_nudge_ = job.get();
   }
-  PostDelayedTask(job.from_here, "DoSyncSessionJob",
-                  base::Bind(&SyncSchedulerImpl::DoSyncSessionJob,
-                             weak_ptr_factory_.GetWeakPtr(),
-                             job),
-                  delay);
+
+  tracked_objects::Location loc(job->from_location());
+  PostDelayedTask(loc, "DoSyncSessionJob",

[akalin, 2012/09/25 22:37:24]

why not just inline job->from_location() here?  Or define loc higher, since you
use it in multiple places

[tim (not reviewing), 2012/10/08 00:20:03]

:( Because base::Passed takes ownership of the job, and the statement is
evaluated right-to-left, so job is null.  Am I misusing base::Passed?  It seemed
weird to me too. Moved declaration up higher.

+      base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoSyncSessionJob),
+                                    weak_ptr_factory_.GetWeakPtr(),
+                                    base::Passed(&job)),
+      delay);
 }
 
-void SyncSchedulerImpl::DoSyncSessionJob(const SyncSessionJob& job) {
+bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  if (job->purpose() == SyncSessionJob::NUDGE) {

[rlarocque, 2012/09/24 21:15:14]

I get the feeling that there's not much point in sharing this function between
CONFIGURE and non-CONFIGURE jobs.  Most of the code in here is irrelevant to
either one job or the other.

[tim (not reviewing), 2012/10/08 00:20:03]

There's still a chunk of common code, but also, for now I'd like to leave it so
that the callsites of DoSyncSessionJob don't need to check the job type, and be
able to state "DoSyncSessionJob" as the thing that takes ownership of running
and ending a job.  We could potentially break it up into one-per type but still
leave DoSyncSessionJob as the top level call that forks internally to address my
concerns, (and make it that much simpler if we ever had to add a new job type,
though this is a lower priority concern).

[rlarocque, 2012/10/08 19:18:04]

I see your point, but I think the call sites should be easy to fix.  They
already know which kind of job is being scheduled.

-> ScheduleConfiguration() only call DoSyncSessionJob on CONFIGURE jobs.
-> ScheduleSyncSessionJob() only deals with POLL and NUDGE jobs.
-> Except for the part where it calls DoSyncSessionJob, DoPendingJobIfPossible()
is one big if statement with branches for CONFIGURATION_MODE and NORMAL_MODE.

[tim (not reviewing), 2012/10/11 17:35:14]

I'm not convinced we should do this yet.  I still see common code in here, I
think it is reasonable to assume we'll want to add common code to all jobs in
the future (random thought - the minute we want more UMA), and for now while I
wrap up this patch it's easiest to say "DoSyncSessionJob" takes ownership if the
job is scheduled, else unscheduled_nudge_storage.

I can add a TODO mentioning we should consider splitting per-job type code into
tailored methods. I'm still of the opinion the fork should happen in here
though, so we have a single choke point, but we can revisit later.

[rlarocque, 2012/10/11 18:15:08]

I think it would be easier to define a function "DoCommonJobTasks()" and call it
from both paths.  

I agree that we should revisit later.

+    if (pending_nudge_ == NULL ||
+        pending_nudge_->session() != job->session()) {
+      // |job| is abandoned.
+      SDVLOG(2) << "Dropping a nudge in "
+                << "DoSyncSessionJob because another nudge was scheduled";
+      return false;
+    }
+    pending_nudge_ = NULL;
+
+    // Rebase the session with the latest model safe table and use it to purge
+    // and update any disabled or modified entries in the job.
+    job->mutable_session()->RebaseRoutingInfoWithLatest(
+        session_context_->routing_info(), session_context_->workers());
+  }
 
   AutoReset<bool> protector(&no_scheduling_allowed_, true);
-  if (!ShouldRunJob(job)) {
+  GetUpdatesCallerInfo::GetUpdatesSource source(
+      job->session()->source().updates_source);
+  if (!ShouldRunJobSaveIfNecessary(job.get())) {
     SLOG(WARNING)
         << "Not executing "
-        << SyncSessionJob::GetPurposeString(job.purpose) << " job from "
-        << GetUpdatesSourceString(job.session->source().updates_source);
-    return;
+        << SyncSessionJob::GetPurposeString(job->purpose()) << " job from "
+        << GetUpdatesSourceString(source);
+    return false;
   }
 
-  if (job.purpose == SyncSessionJob::NUDGE) {
-    if (pending_nudge_.get() == NULL ||
-        pending_nudge_->session != job.session) {
-      SDVLOG(2) << "Dropping a nudge in "
-                << "DoSyncSessionJob because another nudge was scheduled";
-      return;  // Another nudge must have been scheduled in in the meantime.
-    }
-    pending_nudge_.reset();
-
-    // Create the session with the latest model safe table and use it to purge
-    // and update any disabled or modified entries in the job.
-    scoped_ptr<SyncSession> session(CreateSyncSession(job.session->source()));
-
-    job.session->RebaseRoutingInfoWithLatest(*session);
-  }
   SDVLOG(2) << "DoSyncSessionJob with "
-            << SyncSessionJob::GetPurposeString(job.purpose) << " job";
-
-  SyncerStep begin(SYNCER_END);
-  SyncerStep end(SYNCER_END);
-  SetSyncerStepsForPurpose(job.purpose, &begin, &end);
+            << SyncSessionJob::GetPurposeString(job->purpose()) << " job";
 
   bool has_more_to_sync = true;
-  while (ShouldRunJob(job) && has_more_to_sync) {
+  bool premature_exit = false;
+  while (DecideOnJob(job.get()) == CONTINUE && has_more_to_sync) {
     SDVLOG(2) << "Calling SyncShare.";
     // Synchronously perform the sync session from this thread.
-    syncer_->SyncShare(job.session.get(), begin, end);
-    has_more_to_sync = job.session->HasMoreToSync();
+    premature_exit = !syncer_->SyncShare(job->mutable_session(),
+                                         job->start_step(),
+                                         job->end_step());
+
+    has_more_to_sync = job->session()->HasMoreToSync();
     if (has_more_to_sync)
-      job.session->PrepareForAnotherSyncCycle();
+      job->mutable_session()->PrepareForAnotherSyncCycle();
   }
   SDVLOG(2) << "Done SyncShare looping.";
 
-  FinishSyncSessionJob(job);
+  FinishSyncSessionJob(job.get(), premature_exit);

[akalin, 2012/10/03 00:11:34]

can you make SyncSessionJob::Finish() return the value of job->Succeeded() and
then pass job (+ ownership) directly to FinishSyncSessionJob?  I think
simplifies the ownership of job.  You may be able to eliminate
SyncSessionJob::Succeeded() completely, too.

[tim (not reviewing), 2012/10/08 00:20:03]

Yeah, that works! Done.

+  return job->Succeeded();
 }
 
 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;
 
   base::TimeTicks now = TimeTicks::Now();
   // Update timing information for how often datatypes are triggering nudges.
   for (ModelTypeStateMap::const_iterator iter = info.types.begin();
        iter != info.types.end();
        ++iter) {
     base::TimeTicks previous = last_local_nudges_by_model_type_[iter->first];
     last_local_nudges_by_model_type_[iter->first] = now;
     if (previous.is_null())
       continue;
 
 #define PER_DATA_TYPE_MACRO(type_str) \
     SYNC_FREQ_HISTOGRAM("Sync.Freq" type_str, now - previous);
-      SYNC_DATA_TYPE_HISTOGRAM(iter->first);
+    SYNC_DATA_TYPE_HISTOGRAM(iter->first);
 #undef PER_DATA_TYPE_MACRO
   }
 }
 
-void SyncSchedulerImpl::FinishSyncSessionJob(const SyncSessionJob& job) {
+void SyncSchedulerImpl::FinishSyncSessionJob(SyncSessionJob* job,
+                                             bool exited_prematurely) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   // Now update the status of the connection from SCM. We need this to decide
-  // whether we need to save/run future jobs. The notifications from SCM are not
-  // reliable.
+  // whether we need to save/run future jobs. The notifications from SCM are
+  // not reliable.
   //
   // TODO(rlarocque): crbug.com/110954
   // We should get rid of the notifications and it is probably not needed to
-  // maintain this status variable in 2 places. We should query it directly from
-  // SCM when needed.
+  // maintain this status variable in 2 places. We should query it directly
+  // from SCM when needed.
   ServerConnectionManager* scm = session_context_->connection_manager();
   UpdateServerConnectionManagerStatus(scm->server_status());
 
   if (IsSyncingCurrentlySilenced()) {
-    SDVLOG(2) << "We are currently throttled; not scheduling the next sync.";
-    // TODO(sync): Investigate whether we need to check job.purpose
-    // here; see DCHECKs in SaveJob().  (See http://crbug.com/90868.)
-    SaveJob(job);
-    return;  // Nothing to do.
-  } else if (job.session->Succeeded() &&
-             !job.config_params.ready_task.is_null()) {
-    // If this was a configuration job with a ready task, invoke it now that
-    // we finished successfully.
+    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->CloneAndAbandon();

[akalin, 2012/10/03 00:11:34]

Assume you implement my comment above re. passing ownership of job into this
function, do you still need this CloneAndAbandon() call?

[tim (not reviewing), 2012/10/08 00:20:03]

Interesting point.  In general the logic is if you're cloning a job before
having Finish()ed it, you need to CloneAndAbandon, because, well, it's not
Finished.  But agreed that's strange here, since we have full control over what
happens with the job.  This suggested to me that the call to Finish() below
should be moved up before this block, which let me convert CloneAndAbandon to
Clone. Then I kind of realized it's weird to do this scheduling (in the
THROTTLED) case here, and then call in to ScheduleNextSync, so I moved the
rescheduling code into that function as well.

I think it's clearer now, ptal.

+    if (job->purpose() == SyncSessionJob::NUDGE)
+      pending_nudge_ = clone.get();
+    else if (job->purpose() == SyncSessionJob::CONFIGURATION)
+      wait_interval_->pending_configure_job = clone.get();
+    else
+      clone.reset();  // Unthrottling is enough, no need to force a canary.
+
+    RestartWaiting(clone.Pass());
+    return;
+  }
+
+  // Let job know that we're through syncing (calling SyncShare) at this point.
+  {
     AutoReset<bool> protector(&no_scheduling_allowed_, true);
-    job.config_params.ready_task.Run();
+    job->Finish(exited_prematurely);
   }
 
   SDVLOG(2) << "Updating the next polling time after SyncMain";
   ScheduleNextSync(job);
 }
 
-void SyncSchedulerImpl::ScheduleNextSync(const SyncSessionJob& old_job) {
+void SyncSchedulerImpl::ScheduleNextSync(const SyncSessionJob* old_job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  DCHECK(!old_job.session->HasMoreToSync());
+  DCHECK(!old_job->session()->HasMoreToSync());
 
-  AdjustPolling(&old_job);
+  AdjustPolling(old_job);
 
-  if (old_job.session->Succeeded()) {
+  if (old_job->Succeeded()) {
     // Only reset backoff if we actually reached the server.
-    if (old_job.session->SuccessfullyReachedServer())
+    // It's possible that we reached the server on one attempt, then had an
+    // error on the next (or didn't perform some of the server-communicating
+    // commands). We want to verify that, for all commands attempted, we
+    // successfully spoke with the server. Therefore, we verify no errors
+    // and at least one SYNCER_OK.
+    if (old_job->session()->DidReachServer())
       wait_interval_.reset();
     SDVLOG(2) << "Job succeeded so not scheduling more jobs";
     return;
   }
 
-  if (old_job.purpose == SyncSessionJob::POLL) {
+  if (old_job->purpose() == SyncSessionJob::POLL) {
     return; // We don't retry POLL jobs.
   }
 
   // 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
+  // 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
+    // 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";
-
-    DCHECK_EQ(SyncSessionJob::NUDGE, old_job.purpose);
+    SDVLOG(2) << "A nudge during backoff failed, creating new pending nudge.";
+    DCHECK_EQ(SyncSessionJob::NUDGE, old_job->purpose());
     DCHECK(!wait_interval_->had_nudge);
 
     wait_interval_->had_nudge = true;
-    InitOrCoalescePendingJob(old_job);
-    RestartWaiting();
+    DCHECK(!pending_nudge_);
+
+    scoped_ptr<SyncSessionJob> new_job = old_job->Clone();

[akalin, 2012/10/03 00:11:34]

same question here -- can you pass ownership to this function and avoid the
clone?

[tim (not reviewing), 2012/10/08 00:20:03]

I did change this around a bit, see my response above. But note finished job is
finished -- it can't be run again. I commented about this elsewhere but I think
it's easier to follow this pattern than define a bunch of "transform for retry"
style methods on SyncSessionJob.

+    pending_nudge_ = new_job.get();
+    RestartWaiting(new_job.Pass());
   } 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(old_job);
   }
 }
 
 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)
+  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() {
+void SyncSchedulerImpl::RestartWaiting(scoped_ptr<SyncSessionJob> job) {
   CHECK(wait_interval_.get());
   wait_interval_->timer.Stop();
-  wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
-                              this, &SyncSchedulerImpl::DoCanaryJob);
+  if (wait_interval_->mode == WaitInterval::THROTTLED) {
+    wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
+                                base::Bind(&SyncSchedulerImpl::Unthrottle,
+                                    weak_ptr_factory_.GetWeakPtr(),
+                                    base::Passed(&job)));
+  } else {
+    wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
+                                base::Bind(&SyncSchedulerImpl::DoCanaryJob,
+                                    weak_ptr_factory_.GetWeakPtr(),
+                                    base::Passed(&job)));
+  }
 }
 
 void SyncSchedulerImpl::HandleContinuationError(
-    const SyncSessionJob& old_job) {
+    const SyncSessionJob* old_job) {

[akalin, 2012/10/03 00:11:34]

similar to the above, can you pass ownership to this function and avoid the
clone? (or change it to just reset the location)

[tim (not reviewing), 2012/10/08 00:20:03]

I'm going with the logic that a job has a well defined life  (until
DoSyncSessionJob is run through) and supports very limited modifications, versus
having one job that lives forever that we poke at in varying ways and add an
army of fancy setters and methods (to e.g.
reset-this-but-not-that-and-tweak-this for each different case). I actually
started by going down that route, but it was getting messy / fragmented. I think
overall this makes for more consistent code. It makes SyncSessionJob itself
simpler and easier to test and reason about.

As for resetting location, tracked_objects::Location doesn't allow assignment.
We'd have to make it a pointer and reset :/

   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   if (DCHECK_IS_ON()) {
     if (IsBackingOff()) {
-      DCHECK(wait_interval_->timer.IsRunning() || old_job.is_canary_job);
+      DCHECK(wait_interval_->timer.IsRunning() || old_job->is_canary());
     }
   }
 
   TimeDelta length = delay_provider_->GetDelay(
       IsBackingOff() ? wait_interval_->length :
           delay_provider_->GetInitialDelay(
-              old_job.session->status_controller().model_neutral_state()));
+              old_job->session()->status_controller().model_neutral_state()));
 
   SDVLOG(2) << "In handle continuation error with "
-            << SyncSessionJob::GetPurposeString(old_job.purpose)
+            << 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));
-  if (old_job.purpose == SyncSessionJob::CONFIGURATION) {
+  scoped_ptr<SyncSessionJob> new_job(old_job->CloneFromLocation(FROM_HERE));
+  new_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());
-    SyncSession* old = old_job.session.get();
-    SyncSession* s(new SyncSession(session_context_, this,
-        old->source(), old->routing_info(), old->workers()));
-    SyncSessionJob job(old_job.purpose, TimeTicks::Now() + length,
-                       make_linked_ptr(s), false, old_job.config_params,
-                       FROM_HERE);
-    wait_interval_->pending_configure_job.reset(new SyncSessionJob(job));
+    DCHECK(!old_job->config_params().ready_task.is_null());
+    wait_interval_->pending_configure_job = new_job.get();
   } else {
     // We are not in configuration mode. So wait_interval's pending job
     // should be null.
-    DCHECK(wait_interval_->pending_configure_job.get() == NULL);
+    DCHECK(wait_interval_->pending_configure_job == NULL);
+    DCHECK(!pending_nudge_);
+    pending_nudge_ = new_job.get();
+  }
 
-    // TODO(lipalani) - handle clear user data.
-    InitOrCoalescePendingJob(old_job);
-  }
-  RestartWaiting();
+  RestartWaiting(new_job.Pass());
 }
 
 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();
   poll_timer_.Stop();
   if (started_) {
     started_ = false;
   }
   if (!callback.is_null())
     callback.Run();
 }
 
-void SyncSchedulerImpl::DoCanaryJob() {
+void SyncSchedulerImpl::DoCanaryJob(scoped_ptr<SyncSessionJob> to_be_canary) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "Do canary job";
-  DoPendingJobIfPossible(true);
+
+  // Only set canary privlieges here, when we are about to run the job. This

[akalin, 2012/09/25 22:37:24]

privileges

[tim (not reviewing), 2012/10/08 00:20:03]

Done.

+  // avoids confusion in managing canary bits during scheduling, when you
+  // consider that mode switches (e.g., to config) can "pre-empt" a NUDGE that
+  // was scheduled as canary, and send it to an "unscheduled" state.
+  to_be_canary->GrantCanaryPrivilege();
+
+  if (to_be_canary->purpose() == SyncSessionJob::NUDGE) {
+    // TODO(tim): We should be able to remove this...
+    scoped_ptr<SyncSession> temp = CreateSyncSession(
+        to_be_canary->session()->source()).Pass();
+    // The routing info might have been changed since we cached the
+    // pending nudge. Update it by coalescing to the latest.
+    to_be_canary->mutable_session()->Coalesce(*(temp));
+  }
+  DoSyncSessionJob(to_be_canary.Pass());
 }
 
-void SyncSchedulerImpl::DoPendingJobIfPossible(bool is_canary_job) {
+scoped_ptr<SyncSessionJob> SyncSchedulerImpl::TakePendingJobForCurrentMode() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SyncSessionJob* job_to_execute = NULL;
+  // 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.get()) {
+      && wait_interval_->pending_configure_job) {
     SDVLOG(2) << "Found pending configure job";
-    job_to_execute = wait_interval_->pending_configure_job.get();
-  } else if (mode_ == NORMAL_MODE && pending_nudge_.get()) {
+    candidate =
+        wait_interval_->pending_configure_job->CloneAndAbandon().Pass();
+    wait_interval_->pending_configure_job = candidate.get();
+  } else if (mode_ == NORMAL_MODE && pending_nudge_) {
     SDVLOG(2) << "Found pending nudge job";
-
-    scoped_ptr<SyncSession> session(CreateSyncSession(
-        pending_nudge_->session->source()));
-
-    // Also the routing info might have been changed since we cached the
-    // pending nudge. Update it by coalescing to the latest.
-    pending_nudge_->session->Coalesce(*(session.get()));
-    // The pending nudge would be cleared in the DoSyncSessionJob function.
-    job_to_execute = pending_nudge_.get();
+    candidate = pending_nudge_->CloneAndAbandon();
+    pending_nudge_ = candidate.get();
+    unscheduled_nudge_storage_.reset();
   }
-
-  if (job_to_execute != NULL) {
-    SDVLOG(2) << "Executing pending job";
-    SyncSessionJob copy = *job_to_execute;
-    copy.is_canary_job = is_canary_job;
-    DoSyncSessionJob(copy);
-  }
+  return candidate.Pass();
 }
 
-SyncSession* SyncSchedulerImpl::CreateSyncSession(
+scoped_ptr<SyncSession> SyncSchedulerImpl::CreateSyncSession(
     const SyncSourceInfo& source) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DVLOG(2) << "Creating sync session with routes "
            << ModelSafeRoutingInfoToString(session_context_->routing_info());
 
   SyncSourceInfo info(source);
-  SyncSession* session(new SyncSession(session_context_, this, info,
+  return scoped_ptr<SyncSession>(new SyncSession(session_context_, this, info,
       session_context_->routing_info(), session_context_->workers()));
-
-  return session;
 }
 
 void SyncSchedulerImpl::PollTimerCallback() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ModelSafeRoutingInfo r;
   ModelTypeStateMap type_state_map =
       ModelSafeRoutingInfoToStateMap(r, std::string());
   SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, type_state_map);
-  SyncSession* s = CreateSyncSession(info);
-
-  SyncSessionJob job(SyncSessionJob::POLL, TimeTicks::Now(),
-                     make_linked_ptr(s),
-                     false,
-                     ConfigurationParams(),
-                     FROM_HERE);
-
-  ScheduleSyncSessionJob(job);
+  scoped_ptr<SyncSession> s(CreateSyncSession(info));
+  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(SyncSessionJob::POLL,
+                                                    TimeTicks::Now(),
+                                                    s.Pass(),
+                                                    ConfigurationParams(),
+                                                    FROM_HERE));
+  ScheduleSyncSessionJob(job.Pass());
 }
 
-void SyncSchedulerImpl::Unthrottle() {
+void SyncSchedulerImpl::Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
-  SDVLOG(2) << "Unthrottled.";
-  DoCanaryJob();
+  SDVLOG(2) << "Unthrottled " << (to_be_canary.get() ? "with " : "without ")
+            << "canary.";
+  if (to_be_canary.get())
+    DoCanaryJob(to_be_canary.Pass());
+
+  // TODO(tim): ?! This must have been broken.  The way DecideOnJob works today

[rlarocque, 2012/09/24 21:15:14]

Agreed, but this comment won't make much sense other than in the context of this
diff.

[tim (not reviewing), 2012/10/08 00:20:03]

Ah, right.  I had left this as a comment for myself while working.  Reworded and
filed bug.

[rlarocque, 2012/10/08 19:18:04]

On second thought, this function resets the wait_interval_.  IIRC, the flag that
indicates we're currently "globally" throttled is the wait_interval_'s mode. 
Wouldn't clearing the wait_interval_ get us out of the throttled state?

(Granted, it may be too late for this to matter for the Canary job.  However, it
should take effect eventually, right?)

[tim (not reviewing), 2012/10/11 17:35:14]

You're right. It should mean we're good to go on the next nudge. I updated the
bug (155296). 

+  // canary privileges aren't enough to bypass a THROTTLED wait interval, which
+  // would suggest we need to reset first (though trusting canary in Decide is
+  // probably the "right" thing to do).
   wait_interval_.reset();
 }
 
 void SyncSchedulerImpl::Notify(SyncEngineEvent::EventCause cause) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   session_context_->NotifyListeners(SyncEngineEvent(cause));
 }
 
 bool SyncSchedulerImpl::IsBackingOff() const {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   return wait_interval_.get() && wait_interval_->mode ==
       WaitInterval::EXPONENTIAL_BACKOFF;
 }
 
 void SyncSchedulerImpl::OnSilencedUntil(
     const base::TimeTicks& silenced_until) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   wait_interval_.reset(new WaitInterval(WaitInterval::THROTTLED,
                                         silenced_until - TimeTicks::Now()));
-  wait_interval_->timer.Start(FROM_HERE, wait_interval_->length, this,
-      &SyncSchedulerImpl::Unthrottle);
 }
 
 bool SyncSchedulerImpl::IsSyncingCurrentlySilenced() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   return wait_interval_.get() && wait_interval_->mode ==
       WaitInterval::THROTTLED;
 }
 
 void SyncSchedulerImpl::OnReceivedShortPollIntervalUpdate(
     const base::TimeDelta& new_interval) {
@@ skipping 53 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // namespace syncer