sync: Make nudge + config jobs reliable in SyncerThread2

chrome/browser/sync/engine/syncer_thread2.cc

Index: chrome/browser/sync/engine/syncer_thread2.cc
diff --git a/chrome/browser/sync/engine/syncer_thread2.cc b/chrome/browser/sync/engine/syncer_thread2.cc
index e94215e1b5f809ac682d54865f0e2085d812261a..7fcaaff268a5fd6f1994fbeaea70e98114aea0ae 100644
--- a/chrome/browser/sync/engine/syncer_thread2.cc
+++ b/chrome/browser/sync/engine/syncer_thread2.cc
@@ -23,6 +23,18 @@ using sync_pb::GetUpdatesCallerInfo;
 
 namespace s3 {
 
+struct SyncerThread::SyncSessionJob {
+  SyncSessionJobPurpose purpose;
+  base::TimeTicks scheduled_start;
+  linked_ptr<sessions::SyncSession> session;
+  bool is_canary_job;
+
+  // This is the location the nudge came from. used for debugging purpose.
+  // In case of multiple nudges getting coalesced this stores the first nudge
+  // that came in.
+  tracked_objects::Location nudge_location;
+};
+
 struct SyncerThread::WaitInterval {
   enum Mode {
     // A wait interval whose duration has been affected by exponential
@@ -40,20 +52,10 @@ struct SyncerThread::WaitInterval {
   bool had_nudge;
   base::TimeDelta length;
   base::OneShotTimer<SyncerThread> timer;
+  scoped_ptr<SyncSessionJob> pending_job;
   WaitInterval(Mode mode, base::TimeDelta length);
 };
 
-struct SyncerThread::SyncSessionJob {
-  SyncSessionJobPurpose purpose;
-  base::TimeTicks scheduled_start;
-  linked_ptr<sessions::SyncSession> session;
-
-  // This is the location the nudge came from. used for debugging purpose.
-  // In case of multiple nudges getting coalesced this stores the first nudge
-  // that came in.
-  tracked_objects::Location nudge_location;
-};
-
 SyncerThread::DelayProvider::DelayProvider() {}
 SyncerThread::DelayProvider::~DelayProvider() {}
 
@@ -62,6 +64,23 @@ TimeDelta SyncerThread::DelayProvider::GetDelay(
   return SyncerThread::GetRecommendedDelay(last_delay);
 }
 
+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_UNKNOWN:
+      return GetUpdatesCallerInfo::UNKNOWN;
+    default:
+      NOTREACHED();
+      return GetUpdatesCallerInfo::UNKNOWN;
+  }
+}
+
 SyncerThread::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
     : mode(mode), had_nudge(false), length(length) { }
 
@@ -76,7 +95,9 @@ SyncerThread::SyncerThread(sessions::SyncSessionContext* context,
       server_connection_ok_(false),
       delay_provider_(new DelayProvider()),
       syncer_(syncer),
-      session_context_(context) {
+      session_context_(context),
+      saved_nudge_(false),
+      saved_source_(GetUpdatesCallerInfo::UNKNOWN) {
 }
 
 SyncerThread::~SyncerThread() {
@@ -140,75 +161,119 @@ void SyncerThread::StartImpl(Mode mode,
   AdjustPolling(NULL);  // Will kick start poll timer if needed.
   if (callback.get())
     callback->Run();
+
+  if (mode_ == NORMAL_MODE && saved_nudge_ == true) {
+    syncable::ModelTypePayloadMap map;
+    ScheduleNudgeImpl(TimeDelta::FromSeconds(0),
+                      saved_source_, map, false,
+                      FROM_HERE);
+  }
 }
 
-bool SyncerThread::ShouldRunJob(SyncSessionJobPurpose purpose,
-    const TimeTicks& scheduled_start) {
-  DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
+SyncerThread::JobProcessDecision SyncerThread::DecideOnJobWhileBackingOff(
+    const SyncSessionJob& job) {
+
+  DCHECK(wait_interval_.get());
+  DCHECK(job.purpose != CLEAR_USER_DATA);
+
+  if (job.purpose == POLL) {
+    return DROP;
+  }
+
+  DCHECK(job.purpose == NUDGE || job.purpose == CONFIGURATION);
+  if (wait_interval_->mode == WaitInterval::THROTTLED) {
+    return SAVE;
+  }
+
+  DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
+  if (job.purpose == NUDGE) {
+    if (mode_ == CONFIGURATION) {
+      return SAVE;
+    }
+
+    // If we already had one nudge then just drop this nudge. We will retry
+    // later when the timer runs out.
+    if (wait_interval_->had_nudge) {
+      return DROP;
+    } else {
+      // If this is our first nudge during backoff or if our timer ran out
+      // let us retry once more.
+      return CONTINUE;
+    }
+  } else {
+    // This is a config job.
+    // If our timer ran out then continue.

[tim (not reviewing), 2011/04/08 03:37:46]

this can go on above line.

[lipalani1, 2011/04/08 18:40:23]

Done.

+    if (job.is_canary_job) {
+      return CONTINUE;
+    } else {
+      return SAVE;
+    }
+  }
+}
+
+SyncerThread::JobProcessDecision SyncerThread::ProcessJob(

[tim (not reviewing), 2011/04/08 03:37:46]

Perhaps MakeJobDecision or DecideOnJob
Process is misleading.

[lipalani1, 2011/04/08 18:40:23]

Done.

+    const SyncSessionJob& job) {
+  if (job.purpose == CLEAR_USER_DATA) {

[tim (not reviewing), 2011/04/08 03:37:46]

this file in general follows the "no { } on single line if statement", and the
style position is to follow convention as no style is required.  So you should
audit your ifs for this.  (Or change the rest of the file technically but please
don't do that :) )

[lipalani1, 2011/04/08 18:40:23]

Left as such if the if had an else clause. Otherwise changed it.
On 2011/04/08 03:37:46, timsteele wrote:
Done.

+    return CONTINUE;
+  }
 
-  // Check wait interval.
   if (wait_interval_.get()) {
-    // TODO(tim): Consider different handling for CLEAR_USER_DATA (i.e. permit
-    // when throttled).
-    if (wait_interval_->mode == WaitInterval::THROTTLED)
-      return false;
-
-    DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
-    if ((purpose != NUDGE) || wait_interval_->had_nudge)
-      return false;
-  }
-
-  // Mode / purpose contract (See 'Mode' enum in header). Don't run jobs that
-  // were intended for a normal sync if we are in configuration mode, and vice
-  // versa.
-  switch (mode_) {
-    case CONFIGURATION_MODE:
-      if (purpose != CONFIGURATION)
-        return false;
-      break;
-    case NORMAL_MODE:
-      if (purpose == CONFIGURATION)
-        return false;
-      break;
-    default:
-      NOTREACHED() << "Unknown SyncerThread Mode: " << mode_;
-      return false;
+    return DecideOnJobWhileBackingOff(job);

[tim (not reviewing), 2011/04/08 03:37:46]

this method could be named better e.g. not 'BackingfOff' but InWaitInterval 

[lipalani1, 2011/04/08 18:40:23]

Done.

   }
 
-  // Continuation NUDGE tasks have priority over POLLs because they are the
-  // only tasks that trigger exponential backoff, so this prevents them from
-  // being starved from running (e.g. due to a very, very low poll interval,
-  // such as 0ms). It's rare that this would ever matter in practice.
-  if (purpose == POLL && (pending_nudge_.get() &&
-      pending_nudge_->session->source().updates_source ==
-          GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION)) {
-    return false;
+  if (mode_ == CONFIGURATION_MODE) {
+    if (job.purpose == NUDGE) {
+      return SAVE;
+    } else if (job.purpose == CONFIGURATION) {
+      return CONTINUE;
+    } else {
+      return DROP;
+    }
   }
 
-  // Freshness condition.
-  if (purpose == NUDGE &&
-      (scheduled_start < last_sync_session_end_time_)) {
-    return false;
+  // We are in normal mode.
+  DCHECK(mode_ == NORMAL_MODE);
+  DCHECK(job.purpose != CONFIGURATION);
+
+  // Freshness condition
+  if (job.scheduled_start < last_sync_session_end_time_) {
+    return DROP;
+  }
+
+  if (server_connection_ok_) {
+    return CONTINUE;
   }
 
-  return server_connection_ok_;
+  job.purpose == NUDGE ? SAVE : DROP;
 }
 
-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_UNKNOWN:
-      return GetUpdatesCallerInfo::UNKNOWN;
-    default:
-      NOTREACHED();
-      return GetUpdatesCallerInfo::UNKNOWN;
+bool SyncerThread::ShouldRunJob(const SyncSessionJob& job) {
+  JobProcessDecision decision = ProcessJob(job);
+  if (decision == DROP) {

[tim (not reviewing), 2011/04/08 03:37:46]

I'd collapse these 2 if statements into

if (decision != SAVE)
  return decision == CONTINUE;

then the DCHECK_EQ below isn't even needed as it's implicit.

[lipalani1, 2011/04/08 18:40:23]

I am OK with changing it. But readability wise this looks more readable. And
also lesser regressions. 
case in point: if we introduce a 4th state then the above line has to change.
However let me know.
On 2011/04/08 03:37:46, timsteele wrote:

+    return false;
+  }
+
+  if (decision == CONTINUE) {
+    return true;
   }
+
+  DCHECK(job.purpose == NUDGE || job.purpose == CONFIGURATION);
+  DCHECK_EQ(decision, SAVE);
+  if (job.purpose == NUDGE) {
+    saved_nudge_ = true;
+    saved_source_ = job.session->source().updates_source;
+  } else {
+    DCHECK(wait_interval_.get());
+    DCHECK(mode_ == CONFIGURATION_MODE);

[tim (not reviewing), 2011/04/08 03:37:46]

how do we know this??

[lipalani1, 2011/04/08 18:40:23]

If the purpose is config then we should be in config mode. otherwise it is a
bug.
On 2011/04/08 03:37:46, timsteele wrote:

+
+    SyncSession* old = job.session.get();
+    SyncSession* s(new SyncSession(session_context_.get(), this,
+        old->source(), old->routing_info(), old->workers()));
+    SyncSessionJob new_job = {job.purpose, TimeTicks::Now(),
+                          make_linked_ptr(s), false, job.nudge_location};
+    wait_interval_->pending_job.reset(new SyncSessionJob(new_job));
+  }
+  return false;
 }
 
 // Functor for std::find_if to search by ModelSafeGroup.
@@ -240,8 +305,9 @@ void SyncerThread::ScheduleNudge(const TimeDelta& delay,
   ModelTypePayloadMap types_with_payloads =
       syncable::ModelTypePayloadMapFromBitSet(types, std::string());
   thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-      this, &SyncerThread::ScheduleNudgeImpl, delay, source,
-      types_with_payloads, nudge_location));
+      this, &SyncerThread::ScheduleNudgeImpl, delay,
+      GetUpdatesFromNudgeSource(source), types_with_payloads, false,
+      nudge_location));
 }
 
 void SyncerThread::ScheduleNudgeWithPayloads(const TimeDelta& delay,
@@ -253,8 +319,9 @@ void SyncerThread::ScheduleNudgeWithPayloads(const TimeDelta& delay,
   }
 
   thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-      this, &SyncerThread::ScheduleNudgeImpl, delay, source,
-      types_with_payloads, nudge_location));
+      this, &SyncerThread::ScheduleNudgeImpl, delay,
+      GetUpdatesFromNudgeSource(source), types_with_payloads, false,
+      nudge_location));
 }
 
 void SyncerThread::ScheduleClearUserDataImpl() {
@@ -267,15 +334,10 @@ void SyncerThread::ScheduleClearUserDataImpl() {
 }
 
 void SyncerThread::ScheduleNudgeImpl(const TimeDelta& delay,
-    NudgeSource source, const ModelTypePayloadMap& types_with_payloads,
-    const tracked_objects::Location& nudge_location) {
+    GetUpdatesCallerInfo::GetUpdatesSource source,
+    const ModelTypePayloadMap& types_with_payloads,
+    bool is_canary_job, const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
-  TimeTicks rough_start = TimeTicks::Now() + delay;
-  if (!ShouldRunJob(NUDGE, rough_start)) {
-    LOG(WARNING) << "Dropping nudge at scheduling time, source = "
-                 << source;
-    return;
-  }
 
   // Note we currently nudge for all types regardless of the ones incurring
   // the nudge.  Doing different would throw off some syncer commands like
@@ -284,29 +346,39 @@ void SyncerThread::ScheduleNudgeImpl(const TimeDelta& delay,
   std::vector<ModelSafeWorker*> workers;
   session_context_->registrar()->GetModelSafeRoutingInfo(&routes);
   session_context_->registrar()->GetWorkers(&workers);
-  SyncSourceInfo info(GetUpdatesFromNudgeSource(source),
-                      types_with_payloads);
+  SyncSourceInfo info(source, types_with_payloads);
 
-  scoped_ptr<SyncSession> session(new SyncSession(
+  SyncSession* session(new SyncSession(
       session_context_.get(), this, info, routes, workers));
 
+  SyncSessionJob job = {NUDGE, TimeTicks::Now() + delay,
+                        make_linked_ptr(session), is_canary_job,
+                        nudge_location};
+
+  session = NULL;

[tim (not reviewing), 2011/04/08 03:37:46]

what's this for?

[lipalani1, 2011/04/08 18:40:23]

So that nobody reuses session. I can put it in a scope if that makes it easier.
On 2011/04/08 03:37:46, timsteele wrote:

+  if (!ShouldRunJob(job)) {
+    return;
+  }
+
   if (pending_nudge_.get()) {
     if (IsBackingOff() && delay > TimeDelta::FromSeconds(1))
       return;
 
-    pending_nudge_->session->Coalesce(*session.get());
+    pending_nudge_->session->Coalesce(*(job.session.get()));
 
     if (!IsBackingOff()) {
       return;
     } else {
       // Re-schedule the current pending nudge.
       SyncSession* s = pending_nudge_->session.get();
-      session.reset(new SyncSession(s->context(), s->delegate(), s->source(),
-          s->routing_info(), s->workers()));
+      job.session.reset(new SyncSession(s->context(), s->delegate(),
+          s->source(), s->routing_info(), s->workers()));
       pending_nudge_.reset();
     }
   }
-  ScheduleSyncSessionJob(delay, NUDGE, session.release(), nudge_location);
+
+  //  TODO(lipalani) - pass the job itself to ScheduleSyncSessionJob.
+  ScheduleSyncSessionJob(delay, NUDGE, job.session.release(), nudge_location);
 }
 
 // Helper to extract the routing info and workers corresponding to types in
@@ -354,21 +426,23 @@ void SyncerThread::ScheduleConfig(const ModelTypeBitSet& types) {
                              &routes, &workers);
 
   thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-      this, &SyncerThread::ScheduleConfigImpl, routes, workers));
+      this, &SyncerThread::ScheduleConfigImpl, routes, workers,
+      GetUpdatesCallerInfo::FIRST_UPDATE));
 }
 
 void SyncerThread::ScheduleConfigImpl(const ModelSafeRoutingInfo& routing_info,
-    const std::vector<ModelSafeWorker*>& workers) {
+    const std::vector<ModelSafeWorker*>& workers,
+    const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
 
   // TODO(tim): config-specific GetUpdatesCallerInfo value?
   SyncSession* session = new SyncSession(session_context_.get(), this,
-      SyncSourceInfo(GetUpdatesCallerInfo::FIRST_UPDATE,
+      SyncSourceInfo(source,
           syncable::ModelTypePayloadMapFromRoutingInfo(
               routing_info, std::string())),
       routing_info, workers);
   ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), CONFIGURATION, session,
-                         FROM_HERE);
+    FROM_HERE);
 }
 
 void SyncerThread::ScheduleSyncSessionJob(const base::TimeDelta& delay,
@@ -377,7 +451,7 @@ void SyncerThread::ScheduleSyncSessionJob(const base::TimeDelta& delay,
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
 
   SyncSessionJob job = {purpose, TimeTicks::Now() + delay,
-                        make_linked_ptr(session), nudge_location};
+                        make_linked_ptr(session), false, nudge_location};
   if (purpose == NUDGE) {
     DCHECK(!pending_nudge_.get() || pending_nudge_->session.get() == session);
     pending_nudge_.reset(new SyncSessionJob(job));
@@ -409,7 +483,7 @@ void SyncerThread::SetSyncerStepsForPurpose(SyncSessionJobPurpose purpose,
 
 void SyncerThread::DoSyncSessionJob(const SyncSessionJob& job) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
-  if (!ShouldRunJob(job.purpose, job.scheduled_start)) {
+  if (!ShouldRunJob(job)) {
     LOG(WARNING) << "Dropping nudge at DoSyncSessionJob, source = "
         << job.session->source().updates_source;
     return;
@@ -420,6 +494,7 @@ void SyncerThread::DoSyncSessionJob(const SyncSessionJob& job) {
     if (pending_nudge_->session != job.session)
       return;  // Another nudge must have been scheduled in in the meantime.
     pending_nudge_.reset();
+    saved_nudge_ = false;
   }
 
   SyncerStep begin(SYNCER_BEGIN);
@@ -427,7 +502,7 @@ void SyncerThread::DoSyncSessionJob(const SyncSessionJob& job) {
   SetSyncerStepsForPurpose(job.purpose, &begin, &end);
 
   bool has_more_to_sync = true;
-  while (ShouldRunJob(job.purpose, job.scheduled_start) && has_more_to_sync) {
+  while (ShouldRunJob(job) && has_more_to_sync) {
     VLOG(1) << "SyncerThread: Calling SyncShare.";
     // Synchronously perform the sync session from this thread.
     syncer_->SyncShare(job.session.get(), begin, end);
@@ -519,8 +594,15 @@ void SyncerThread::ScheduleNextSync(const SyncSessionJob& old_job) {
   } else {
     // We weren't continuing and we aren't in backoff.  Schedule a normal
     // continuation.
-    ScheduleNudgeImpl(TimeDelta::FromSeconds(0), NUDGE_SOURCE_CONTINUATION,
-                      old_job.session->source().types, FROM_HERE);
+    if (old_job.purpose == NUDGE) {
+      ScheduleNudgeImpl(TimeDelta::FromSeconds(0),
+                        GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION),
+                        old_job.session->source().types, false, FROM_HERE);
+    } else if (old_job.purpose == CONFIGURATION) {
+      ScheduleConfigImpl(old_job.session->routing_info(),
+          old_job.session->workers(),
+          GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION));
+    } // Drop the rest.
   }
 }
 
@@ -556,9 +638,17 @@ void SyncerThread::HandleConsecutiveContinuationError(
       IsBackingOff() ? wait_interval_->length : TimeDelta::FromSeconds(1));
   wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
                                         length));
-  SyncSessionJob job = {NUDGE, TimeTicks::Now() + length,
-                        make_linked_ptr(s), FROM_HERE};
-  pending_nudge_.reset(new SyncSessionJob(job));
+  if (old_job.purpose == CONFIGURATION) {
+    SyncSessionJob job = {old_job.purpose, TimeTicks::Now() + length,
+                        make_linked_ptr(s), false, FROM_HERE};
+    wait_interval_->pending_job.reset(new SyncSessionJob(job));
+  } else {
+    // No matter what type of job it is.(nudge or poll), we are going to treat
+    // it as nudge when doing exponential back off retries.
+    saved_nudge_ = true;
+    saved_source_ = old_job.session->source().updates_source;
+    wait_interval_->pending_job.reset();
+  }
   wait_interval_->timer.Start(length, this, &SyncerThread::DoCanaryJob);
 }
 
@@ -593,10 +683,23 @@ void SyncerThread::Stop() {
 }
 
 void SyncerThread::DoCanaryJob() {
-  DCHECK(pending_nudge_.get());
   wait_interval_->had_nudge = false;
-  SyncSessionJob copy = *pending_nudge_;
-  DoSyncSessionJob(copy);
+
+  // We should have one of 2 things. Otherwise we shouldnt be running the timer.
+  DCHECK(wait_interval_->pending_job.get() || saved_nudge_);
+  if (mode_ == CONFIGURATION_MODE) {
+    if (wait_interval_->pending_job.get()) {
+      SyncSessionJob copy = *(wait_interval_->pending_job);
+      copy.is_canary_job = true;
+      DoSyncSessionJob(copy);
+    }
+  } else {
+    if (saved_nudge_) {
+      syncable::ModelTypePayloadMap map;
+      ScheduleNudgeImpl(TimeDelta::FromSeconds(0), saved_source_, map, true,
+                        FROM_HERE);
+    }
+  }
 }
 
 void SyncerThread::PollTimerCallback() {
@@ -614,6 +717,7 @@ void SyncerThread::PollTimerCallback() {
 
 void SyncerThread::Unthrottle() {
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
+  DoCanaryJob();
   wait_interval_.reset();
 }