sync: Handle POLL jobs in separate a code path

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 325 matching lines @@
   }
 
   return true;
 }
 
 SyncSchedulerImpl::JobProcessDecision
 SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob& job,
                                              JobPriority priority) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(wait_interval_.get());
+  DCHECK_NE(job.purpose(), SyncSessionJob::POLL);
 
   SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
             << WaitInterval::GetModeString(wait_interval_->mode)
             << (wait_interval_->had_nudge ? " (had nudge)" : "")
             << ((priority == CANARY_PRIORITY) ? " (canary)" : "");
 
-  if (job.purpose() == SyncSessionJob::POLL)
-    return DROP;
-
   // 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;
@@ skipping 11 matching lines @@
       return CONTINUE;
   }
   return (priority == CANARY_PRIORITY) ? CONTINUE : SAVE;
 }
 
 SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
     const SyncSessionJob& job,
     JobPriority priority) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
+  // POLL jobs do not call this function.
+  DCHECK(job.purpose() == SyncSessionJob::NUDGE ||
+         job.purpose() == SyncSessionJob::CONFIGURATION);
+
   // 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) {
     ModelTypeSet requested_types;
     for (ModelTypeInvalidationMap::const_iterator i =
          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))
       return DROP;  // TODO(tim): Don't drop. http://crbug.com/177659
   }
 
   if (wait_interval_.get())
     return DecideWhileInWaitInterval(job, priority);
 
   if (mode_ == CONFIGURATION_MODE) {
     if (job.purpose() == SyncSessionJob::NUDGE)
       return SAVE;  // Running requires a mode switch.
-    else if (job.purpose() == SyncSessionJob::CONFIGURATION)
+    else // job.purpose() == SyncSessionJob::CONFIGURATION

[tim (not reviewing), 2013/03/18 22:57:40]

I think you mean to delete this.

       return CONTINUE;
-    else
-      return DROP;
   }
 
   // We are in normal mode.
   DCHECK_EQ(mode_, NORMAL_MODE);
   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
@@ skipping 243 matching lines @@
   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(

[tim (not reviewing), 2013/03/18 22:57:40]

How about we call this PostNudgeJob, or something, to distinguish it from the
"ScheduleNudge" APIs that may not proceed with scheduling for whatever reason,
and to make the fact that it is only used for NUDGE jobs self-explanatory. 

[rlarocque, 2013/03/19 00:15:03]

I'd like to merge this into ScheduleNudgeImpl.  If you prefer, I could do that
as part of this patch.  I think that would be a better solution that just
renaming it.

[tim (not reviewing), 2013/03/19 21:50:02]

OK, lets merge it there. Make sure you update comments that speak of
ScheduleSyncSessionJob.

[rlarocque, 2013/03/19 22:46:42]

Done.  The merge wasn't entirely straightforward.  I decided to remove redundant
DCHECKs and to change a few comments and variable names.  It should be a
refactoring-only change, though.

     const tracked_objects::Location& loc,
     scoped_ptr<SyncSessionJob> job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK_EQ(job->purpose(), SyncSessionJob::NUDGE);
   if (no_scheduling_allowed_) {
     NOTREACHED() << "Illegal to schedule job while session in progress.";
     return;
   }
 
   TimeDelta delay = job->scheduled_start() - TimeTicks::Now();
   if (delay < TimeDelta::FromMilliseconds(0))
     delay = TimeDelta::FromMilliseconds(0);
   SDVLOG_LOC(loc, 2)
       << "In ScheduleSyncSessionJob with "
       << 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(loc, 2) << "Resetting pending_nudge to ";
-    DCHECK(!pending_nudge_ || pending_nudge_->session() ==
-           job->session());
-    pending_nudge_ = job.get();
-  }
+  SDVLOG_LOC(loc, 2) << "Resetting pending_nudge to ";
+  DCHECK(!pending_nudge_ || pending_nudge_->session() ==
+         job->session());
+  pending_nudge_ = job.get();
 
   PostDelayedTask(loc, "DoSyncSessionJob",
       base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoSyncSessionJob),
                                     weak_ptr_factory_.GetWeakPtr(),
                                     base::Passed(&job),
                                     NORMAL_PRIORITY),
       delay);
 }
 
 bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job,
@@ skipping 32 matching lines @@
     return false;
   }
 
   SDVLOG(2) << "Calling SyncShare with "
             << SyncSessionJob::GetPurposeString(job->purpose()) << " job";
   bool premature_exit = !syncer_->SyncShare(job->mutable_session(),
                                             job->start_step(),
                                             job->end_step());
   SDVLOG(2) << "Done SyncShare, returned: " << premature_exit;
 
-  return FinishSyncSessionJob(job.Pass(), premature_exit);
+  bool success = FinishSyncSessionJob(job.get(), premature_exit);

[rlarocque, 2013/03/18 22:02:18]

One of the goals of this patch is to take all the poll-specific decision-making
out of FinishSyncSessionJob() and ScheduleNextSync().  I admit this change seems
a bit out of place in this patch, but I couldn't find any better place to put
it.  

To implement this, it makes ScheduleNextSync() a function that always resets the
backoff interval, and inlines most of the rest of its functionality into
DoSyncSessionJob().  The few parts of ShceduleNextSync() that apply to poll jobs
were inlined into DoPollSyncSessionJob().

FinishSyncSessionJob no longer calls ScheduleNextSync() directly, and it is
called by both DoSyncSessionJob and DoPollSyncSessionJob().

[tim (not reviewing), 2013/03/18 22:57:40]

Please audit all the comments that reference DoSyncSessionJob and update them to
(e.g) Do[Poll]SyncSessionJob as necessary.

+
+  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();
+    else
+      NOTREACHED();
+
+    RestartWaiting(clone.Pass());
+    return success;
+  }
+
+  if (!success) {

[tim (not reviewing), 2013/03/18 22:57:40]

nit - please make { } usage consistent for single line ifs in this file.

[rlarocque, 2013/03/19 00:15:03]

I think it's been inconsistent for a while.  For example, see lines 825 and and
836 in the old file.  

I can remove the brackets here.  That would at least make this function
consistent.

+    ScheduleNextSync(job.Pass());
+  }
+
+  return success;
+}
+
+void SyncSchedulerImpl::DoPollSyncSessionJob(scoped_ptr<SyncSessionJob> job) {
+  DCHECK_EQ(job->purpose(), SyncSessionJob::POLL);
+
+  base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
+
+  if (wait_interval_.get()) {

[rlarocque, 2013/03/18 22:02:18]

The if statements on lines 787-800 are intended to be equivalent to
DecideOnJob's old decision making.  We never saved a POLL job, so the only
decision to be made is whether to drop it (return immediately) or continue
(don't return).

[tim (not reviewing), 2013/03/18 22:57:40]

I know this may seem pedantic, but I'd like to try really hard to keep logic of
when to DROP / SAVE jobs in one place. At one time we had it split-up just like
this. It was harder to grasp all the different reasons we'd decide to continue /
save / drop by looking at the code without reading the whole file; we would
frequently forget about "that other place we let jobs through".

Could we a) at least have a DecideOnPollJob, so that we force ourselves to stick
to a pattern (DecideOn*Job) and don't scatter more and more of this code around
the class, and b) ideally call it from DecideOnJob if a parameter was POLL, so
that we still have one single gauntlet (to examine, to tweak if new requirements
arise for existing jobs (or possibly new job types down the road), to test).  

Maybe you can suggest something better, but I'd like it if the spirit of that
reasoning wasn't lost.

[rlarocque, 2013/03/19 00:15:03]

I see your point.  I agree that we want to keep all the decision-making logic in
one place and all the job running logic in a different place.  

However, I think that the decision-making is handled differently in different
contexts, and trying to merge them all can have its own set of problems.  For
example, we have crbug.com/179515, which may have come about because it made
sense to DROP a job in one context, but not another.  There is also
crbug.com/177659, where a decision to return "SAVE" made sense for some contexts
where DecideOnJob() was being called, but not others.  

In other words, it's a tradeoff.  Both approaches have potential for bugs.

I'm hoping that the other changes coming up will reduce the probability of rogue
jobs.  For one, I plan to limit the number of jobs currently in flight to one. 
I'd also like to decouple the save/drop logic from the run/don't run logic.  

Your suggestion to add a DecideOnPollJob() makes sense.  All logic for a given
purpose (poll, nudge, or config) should be in one place.  For example, that
means that we'll probably need to keep a "DecideOnNudgeJob" function to share
logic between ScheduleNudgeImpl() and DoNudgeSyncSessionJob().  However, I don't
see much benefit in sharing code between purpose, since their decision logic is
so different.  

If you're still not convinced, we should talk in person about this.

+    SDVLOG(2) << "Dropping POLL job in wait interval.";
+    return;
+  }
+
+  if (mode_ == CONFIGURATION_MODE) {
+    SDVLOG(2) << "Dropping POLL job in configuration mode.";
+    return;
+  }
+
+  if (session_context_->connection_manager()->HasInvalidAuthToken()) {
+    SDVLOG(2) << "Dropping POLL job because auth token is invalid.";
+    return;
+  }
+
+  SDVLOG(2) << "Calling SyncShare with "
+            << SyncSessionJob::GetPurposeString(job->purpose()) << " job";
+  bool premature_exit = !syncer_->SyncShare(job->mutable_session(),
+                                            job->start_step(),
+                                            job->end_step());
+  SDVLOG(2) << "Done SyncShare, returned: " << premature_exit;
+
+  FinishSyncSessionJob(job.get(), premature_exit);
+
+  if (IsSyncingCurrentlySilenced()) {
+    RestartWaiting(scoped_ptr<SyncSessionJob>());

[tim (not reviewing), 2013/03/18 22:57:40]

Comment that we send a NULL job here because a POLL job does not warrant a
canary, but we still want to make sure our bookkeeping is up to date so that any
other jobs that come in are unblocked.

[rlarocque, 2013/03/19 00:15:03]

Will do.

+  }
 }
 
 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 (ModelTypeInvalidationMap::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);
 #undef PER_DATA_TYPE_MACRO
   }
 }
 
-bool SyncSchedulerImpl::FinishSyncSessionJob(scoped_ptr<SyncSessionJob> job,
+bool SyncSchedulerImpl::FinishSyncSessionJob(SyncSessionJob* job,
                                              bool exited_prematurely) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   // Let job know that we're through syncing (calling SyncShare) at this point.
   bool succeeded = false;
   {
     base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
     succeeded = job->Finish(exited_prematurely);
   }
 
   SDVLOG(2) << "Updating the next polling time after SyncMain";
-  ScheduleNextSync(job.Pass(), succeeded);
-  return succeeded;
-}
 
-void SyncSchedulerImpl::ScheduleNextSync(
-    scoped_ptr<SyncSessionJob> finished_job, bool succeeded) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-
-  AdjustPolling(finished_job.get());
+  AdjustPolling(job);
 
   if (succeeded) {
     // No job currently supported by the scheduler could succeed without
     // successfully reaching the server.  Therefore, if we make it here, it is
     // appropriate to reset the backoff interval.
     wait_interval_.reset();
     NotifyRetryTime(base::Time());
     SDVLOG(2) << "Job succeeded so not scheduling more jobs";
-    return;
+    return succeeded;

[tim (not reviewing), 2013/03/18 22:57:40]

Just fall through?

[rlarocque, 2013/03/19 00:15:03]

Will do.

   }
 
-  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 = finished_job->Clone();
-    if (clone->purpose() == SyncSessionJob::NUDGE)
-      pending_nudge_ = clone.get();
-    else if (clone->purpose() == SyncSessionJob::CONFIGURATION)
-      wait_interval_->pending_configure_job = clone.get();
-    else
-      clone.reset();  // Unthrottling is enough, no need to force a canary.
+  return succeeded;
+}
 
-    RestartWaiting(clone.Pass());
-    return;
-  }
-
-  if (finished_job->purpose() == SyncSessionJob::POLL) {
-    return; // We don't retry POLL jobs.
-  }
+void SyncSchedulerImpl::ScheduleNextSync(
+    scoped_ptr<SyncSessionJob> finished_job) {
+  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(finished_job->purpose() == SyncSessionJob::CONFIGURATION
+         || finished_job->purpose() == SyncSessionJob::NUDGE);
 
   // TODO(rlarocque): There's no reason why we should blindly backoff and retry
   // if we don't succeed.  Some types of errors are not likely to disappear on
   // their own.  With the return values now available in the old_job.session,
   // we should be able to detect such errors and only retry when we detect
   // transient errors.
 
   if (IsBackingOff() && wait_interval_->timer.IsRunning() &&
       mode_ == NORMAL_MODE) {
     // When in normal mode, we allow up to one nudge per backoff interval.  It
@@ skipping 179 matching lines @@
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ModelSafeRoutingInfo r;
   ModelTypeInvalidationMap invalidation_map =
       ModelSafeRoutingInfoToInvalidationMap(r, std::string());
   SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, invalidation_map);
   scoped_ptr<SyncSession> s(CreateSyncSession(info));
   scoped_ptr<SyncSessionJob> job(new SyncSessionJob(SyncSessionJob::POLL,
                                                     TimeTicks::Now(),
                                                     s.Pass(),
                                                     ConfigurationParams()));
-  ScheduleSyncSessionJob(FROM_HERE, job.Pass());
+  if (no_scheduling_allowed_) {
+    NOTREACHED() << "Illegal to schedule job while session in progress.";

[tim (not reviewing), 2013/03/18 22:57:40]

Can you explain why we can't be here, given that this is a timer callback and
hence not an explicit synchronous call from any of the logic in the scheduler? 
It's not obvious how we "protect" against this value being true when an
essentially randomly-timed callback fires.

(i.e because if we're processing the callback task and no_scheduling_allowed_ is
true, it would imply we either forgot to unset it or we're in DoSyncSessionJob
and calling this method explicitly, which would be wrong).

[rlarocque, 2013/03/19 00:15:03]

The no_scheduling_allowed_ flag is true only when a function that runs only on
the sync thread is currently running.  Because this callback fires only on the
sync thread, we can't be running such a function when this callback is being
executed.  

I'll add a comment to document this fact.

By the way, the logic to maintain this flag is a bit strange.  We end up
unsetting the flag in FinishSyncSessionJob while the protector declared on the
stack in DoSyncSessionJob is still in scope.  I don't know if that's intentional
or not.

+    return;
+  }
+
+  DoPollSyncSessionJob(job.Pass());
 }
 
 void SyncSchedulerImpl::Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary) {
   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.";
 
@@ skipping 102 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // namespace syncer