[Sync] Refactoring polling to be reliable.

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"

[pavely, 2015/05/08 23:04:13]

I think auto_reset.h is not needed anymore.

 #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/message_loop.h"
 #include "sync/engine/backoff_delay_provider.h"
 #include "sync/engine/syncer.h"
 #include "sync/protocol/proto_enum_conversions.h"
 #include "sync/protocol/sync.pb.h"
@@ skipping 128 matching lines @@
 SyncSchedulerImpl::SyncSchedulerImpl(const std::string& name,
                                      BackoffDelayProvider* delay_provider,
                                      sessions::SyncSessionContext* context,
                                      Syncer* syncer)
     : name_(name),
       started_(false),
       syncer_short_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultShortPollIntervalSeconds)),
       syncer_long_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultLongPollIntervalSeconds)),
-      mode_(NORMAL_MODE),
+      mode_(CONFIGURATION_MODE),
       delay_provider_(delay_provider),
       syncer_(syncer),
       session_context_(context),
-      no_scheduling_allowed_(false),
-      do_poll_after_credentials_updated_(false),
+      last_poll_reset_(base::TimeTicks::Now()),
       next_sync_session_job_priority_(NORMAL_PRIORITY),
       weak_ptr_factory_(this),
       weak_ptr_factory_for_weak_handle_(this) {
   weak_handle_this_ = MakeWeakHandle(
       weak_ptr_factory_for_weak_handle_.GetWeakPtr());
 }
 
 SyncSchedulerImpl::~SyncSchedulerImpl() {
   DCHECK(CalledOnValidThread());
   Stop();
@@ skipping 25 matching lines @@
   // 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 TryCanaryJob 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.
   TryCanaryJob();
 }
 
-void SyncSchedulerImpl::Start(Mode mode) {
+void SyncSchedulerImpl::Start(Mode mode, base::Time last_poll_time) {
   DCHECK(CalledOnValidThread());
   std::string thread_name = base::MessageLoop::current()->thread_name();
   if (thread_name.empty())
     thread_name = "<Main thread>";
   SDVLOG(2) << "Start called from thread "
             << thread_name << " with mode " << GetModeString(mode);
   if (!started_) {
     started_ = true;
     SendInitialSnapshot();
   }
 
   DCHECK(!session_context_->account_name().empty());
   DCHECK(syncer_.get());
   Mode old_mode = mode_;
   mode_ = mode;
-  AdjustPolling(UPDATE_INTERVAL);  // Will kick start poll timer if needed.
+  // Only adjust the poll reset time if it was valid and in the past.
+  if (!last_poll_time.is_null() && last_poll_time < base::Time::Now()) {
+    // Convert from base::Time to base::TimeTicks. The reason we use Time
+    // for persisting is that TimeTicks can stop making forward progress when
+    // the machine is suspended. This implies that on resume the client might
+    // actually have miss the real poll, unless the client is restarted. Fixing
+    // that would require using an AlarmTimer though, which is only supported
+    // on certain platforms.
+    last_poll_reset_ =
+        base::TimeTicks::Now() - (base::Time::Now() - last_poll_time);
+  }
 
   if (old_mode != mode_ && mode_ == NORMAL_MODE) {
     // We just got back to normal mode.  Let's try to run the work that was
     // queued up while we were configuring.
 
+    AdjustPolling(UPDATE_INTERVAL);  // Will kick start poll timer if needed.
+
     // Update our current time before checking IsRetryRequired().
     nudge_tracker_.SetSyncCycleStartTime(base::TimeTicks::Now());
     if (nudge_tracker_.IsSyncRequired() && CanRunNudgeJobNow(NORMAL_PRIORITY)) {
       TrySyncSessionJob();
     }
   }
 }
 
 ModelTypeSet SyncSchedulerImpl::GetEnabledAndUnthrottledTypes() {
   ModelTypeSet enabled_types = session_context_->GetEnabledTypes();
@@ skipping 57 matching lines @@
     pending_configure_params_.reset(new ConfigurationParams(params));
     TrySyncSessionJob();
   } else {
     SDVLOG(2) << "No change in routing info, calling ready task directly.";
     params.ready_task.Run();
   }
 }
 
 bool SyncSchedulerImpl::CanRunJobNow(JobPriority priority) {
   DCHECK(CalledOnValidThread());
-  if (wait_interval_ && wait_interval_->mode == WaitInterval::THROTTLED) {
+  if (IsCurrentlyThrottled()) {
     SDVLOG(1) << "Unable to run a job because we're throttled.";
     return false;
   }
 
-  if (wait_interval_
-      && wait_interval_->mode == WaitInterval::EXPONENTIAL_BACKOFF
-      && priority != CANARY_PRIORITY) {
+  if (IsBackingOff() && priority != CANARY_PRIORITY) {
     SDVLOG(1) << "Unable to run a job because we're backing off.";
     return false;
   }
 
   if (session_context_->connection_manager()->HasInvalidAuthToken()) {
     SDVLOG(1) << "Unable to run a job because we have no valid auth token.";
     return false;
   }
 
   return true;
@@ skipping 70 matching lines @@
   nudge_tracker_.RecordInitialSyncRequired(model_type);
   ScheduleNudgeImpl(TimeDelta::FromSeconds(0), FROM_HERE);
 }
 
 // TODO(zea): Consider adding separate throttling/backoff for datatype
 // refresh requests.
 void SyncSchedulerImpl::ScheduleNudgeImpl(
     const TimeDelta& delay,
     const tracked_objects::Location& nudge_location) {
   DCHECK(CalledOnValidThread());
-
-  if (no_scheduling_allowed_) {
-    NOTREACHED() << "Illegal to schedule job while session in progress.";
-    return;
-  }
+  CHECK(!syncer_->IsSyncing());
 
   if (!started_) {
     SDVLOG_LOC(nudge_location, 2)
         << "Dropping nudge, scheduler is not running.";
     return;
   }
 
   SDVLOG_LOC(nudge_location, 2)
       << "In ScheduleNudgeImpl with delay "
       << delay.InMilliseconds() << " ms";
@@ skipping 35 matching lines @@
   nudge_tracker_.SetDefaultNudgeDelay(delay_ms);
 }
 
 void SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {
   DCHECK(CalledOnValidThread());
   DCHECK(CanRunNudgeJobNow(priority));
 
   DVLOG(2) << "Will run normal mode sync cycle with types "
            << ModelTypeSetToString(session_context_->GetEnabledTypes());
   scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
-  bool premature_exit = !syncer_->NormalSyncShare(
+  bool success = syncer_->NormalSyncShare(
       GetEnabledAndUnthrottledTypes(), &nudge_tracker_, session.get());
-  AdjustPolling(FORCE_RESET);
-  // Don't run poll job till the next time poll timer fires.
-  do_poll_after_credentials_updated_ = false;
-
-  bool success = !premature_exit
-      && !sessions::HasSyncerError(
-          session->status_controller().model_neutral_state());
 
   if (success) {
     // That cycle took care of any outstanding work we had.
     SDVLOG(2) << "Nudge succeeded.";
     nudge_tracker_.RecordSuccessfulSyncCycle();
     scheduled_nudge_time_ = base::TimeTicks();
+    HandleSuccess();
 
-    // If we're here, then we successfully reached the server.  End all backoff.
-    wait_interval_.reset();
-    NotifyRetryTime(base::Time());
+    // If this was a canary, we may need to restart the poll timer (the poll
+    // timer may have fired while the scheduler was in an error state, ignoring
+    // the poll).
+    if (!poll_timer_.IsRunning()) {
+      SDVLOG(1) << "Canary succeeded, restarting polling.";
+      AdjustPolling(UPDATE_INTERVAL);
+    }
   } else {
     HandleFailure(session->status_controller().model_neutral_state());
   }
 }
 
 void SyncSchedulerImpl::DoConfigurationSyncSessionJob(JobPriority priority) {
   DCHECK(CalledOnValidThread());
   DCHECK_EQ(mode_, CONFIGURATION_MODE);
   DCHECK(pending_configure_params_ != NULL);
 
   if (!CanRunJobNow(priority)) {
     SDVLOG(2) << "Unable to run configure job right now.";
     if (!pending_configure_params_->retry_task.is_null()) {
       pending_configure_params_->retry_task.Run();
       pending_configure_params_->retry_task.Reset();
     }
     return;
   }
 
   SDVLOG(2) << "Will run configure SyncShare with types "
             << ModelTypeSetToString(session_context_->GetEnabledTypes());
   scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
-  bool premature_exit = !syncer_->ConfigureSyncShare(
+  bool success = syncer_->ConfigureSyncShare(
       pending_configure_params_->types_to_download,
       pending_configure_params_->source,
       session.get());
-  AdjustPolling(FORCE_RESET);
-  // Don't run poll job till the next time poll timer fires.
-  do_poll_after_credentials_updated_ = false;
-
-  bool success = !premature_exit
-      && !sessions::HasSyncerError(
-          session->status_controller().model_neutral_state());
 
   if (success) {
     SDVLOG(2) << "Configure succeeded.";
     pending_configure_params_->ready_task.Run();
     pending_configure_params_.reset();
-
-    // If we're here, then we successfully reached the server.  End all backoff.
-    wait_interval_.reset();
-    NotifyRetryTime(base::Time());
+    HandleSuccess();
   } else {
     HandleFailure(session->status_controller().model_neutral_state());
     // Sync cycle might receive response from server that causes scheduler to
     // stop and draws pending_configure_params_ invalid.
     if (started_ && !pending_configure_params_->retry_task.is_null()) {
       pending_configure_params_->retry_task.Run();
       pending_configure_params_->retry_task.Reset();
     }
   }
 }
 
+void SyncSchedulerImpl::HandleSuccess() {
+  // If we're here, then we successfully reached the server.  End all backoff.
+  wait_interval_.reset();
+  NotifyRetryTime(base::Time());
+}
+
 void SyncSchedulerImpl::HandleFailure(
     const sessions::ModelNeutralState& model_neutral_state) {
   if (IsCurrentlyThrottled()) {
     SDVLOG(2) << "Was throttled during previous sync cycle.";
-    RestartWaiting();
   } else if (!IsBackingOff()) {
     // Setup our backoff if this is our first such failure.
     TimeDelta length = delay_provider_->GetDelay(
         delay_provider_->GetInitialDelay(model_neutral_state));
     wait_interval_.reset(
         new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, length));
     SDVLOG(2) << "Sync cycle failed.  Will back off for "
         << wait_interval_->length.InMilliseconds() << "ms.";
-    RestartWaiting();
+  } else {
+    // Increase our backoff interval and schedule another retry.
+    TimeDelta length = delay_provider_->GetDelay(wait_interval_->length);
+    wait_interval_.reset(
+        new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, length));
+    SDVLOG(2) << "Sync cycle failed.  Will back off for "
+        << wait_interval_->length.InMilliseconds() << "ms.";
+  }
+  RestartWaiting();
+}
+
+void SyncSchedulerImpl::DoPollSyncSessionJob() {
+  SDVLOG(2) << "Polling with types "
+            << ModelTypeSetToString(GetEnabledAndUnthrottledTypes());
+  scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
+  bool success = syncer_->PollSyncShare(
+      GetEnabledAndUnthrottledTypes(),
+      session.get());
+
+  // Only restart the timer if the poll succeeded. Otherwise rely on normal
+  // failure handling to retry with backoff.
+  if (success) {
+    AdjustPolling(FORCE_RESET);
+    HandleSuccess();
+  } else {
+    HandleFailure(session->status_controller().model_neutral_state());
   }
 }
 
-void SyncSchedulerImpl::DoPollSyncSessionJob() {
-  base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
-
-  SDVLOG(2) << "Polling with types "
-            << ModelTypeSetToString(GetEnabledAndUnthrottledTypes());
-  scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
-  syncer_->PollSyncShare(
-      GetEnabledAndUnthrottledTypes(),
-      session.get());
-
-  AdjustPolling(FORCE_RESET);
-
-  if (IsCurrentlyThrottled()) {
-    SDVLOG(2) << "Poll request got us throttled.";
-    // The OnSilencedUntil() call set up the WaitInterval for us.  All we need
-    // to do is start the timer.
-    RestartWaiting();
-  }
-}
-
 void SyncSchedulerImpl::UpdateNudgeTimeRecords(ModelTypeSet types) {
   DCHECK(CalledOnValidThread());
   base::TimeTicks now = TimeTicks::Now();
   // Update timing information for how often datatypes are triggering nudges.
   for (ModelTypeSet::Iterator iter = types.First(); iter.Good(); iter.Inc()) {
     base::TimeTicks previous = last_local_nudges_by_model_type_[iter.Get()];
     last_local_nudges_by_model_type_[iter.Get()] = 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.Get());
 #undef PER_DATA_TYPE_MACRO
   }
 }
 
 TimeDelta SyncSchedulerImpl::GetPollInterval() {
   return (!session_context_->notifications_enabled() ||
           !session_context_->ShouldFetchUpdatesBeforeCommit()) ?
       syncer_short_poll_interval_seconds_ :
       syncer_long_poll_interval_seconds_;
 }
 
 void SyncSchedulerImpl::AdjustPolling(PollAdjustType type) {
   DCHECK(CalledOnValidThread());
+  if (!started_)
+    return;
 
-  TimeDelta poll = GetPollInterval();
-  bool rate_changed = !poll_timer_.IsRunning() ||
-                       poll != poll_timer_.GetCurrentDelay();
+  TimeDelta poll_interval = GetPollInterval();
+  TimeDelta poll_delay = poll_interval;
+  const TimeTicks now = TimeTicks::Now();
 
-  if (type == FORCE_RESET) {
-    last_poll_reset_ = base::TimeTicks::Now();
-    if (!rate_changed)
-      poll_timer_.Reset();
+  if (type == UPDATE_INTERVAL) {
+    // Override the delay based on the last successful poll time (if it was
+    // set).
+    TimeTicks new_poll_time;
+    if (!last_poll_reset_.is_null())
+      new_poll_time = poll_interval + last_poll_reset_;
+    else
+      new_poll_time = poll_interval + now;
+    poll_delay = new_poll_time - now;
+    if (poll_delay < TimeDelta()) {
+      // The desired poll time was in the past, so trigger a poll now (the timer
+      /// will post the task asynchronously, so re-entrancy isn't an issue).
+      poll_delay = TimeDelta();
+    }
+  } else {
+    // Otherwise just restart the timer.
+    DCHECK_EQ(FORCE_RESET, type);
+    DCHECK_EQ(GetPollInterval(), poll_delay);
+    last_poll_reset_ = now;
   }
 
-  if (!rate_changed)
-    return;
+  SDVLOG(1) << "Updating polling delay to " << poll_delay.InMinutes()
+            << " minutes.";
 
-  // Adjust poll rate.
-  poll_timer_.Stop();
-  poll_timer_.Start(FROM_HERE, poll, this,
+  // Adjust poll rate. Start will reset the timer if it was already running.
+  poll_timer_.Start(FROM_HERE, poll_delay, this,
                     &SyncSchedulerImpl::PollTimerCallback);
 }
 
 void SyncSchedulerImpl::RestartWaiting() {
   CHECK(wait_interval_.get());
   DCHECK(wait_interval_->length >= TimeDelta::FromSeconds(0));
   NotifyRetryTime(base::Time::Now() + wait_interval_->length);
   SDVLOG(2) << "Starting WaitInterval timer of length "
       << wait_interval_->length.InMilliseconds() << "ms.";
   if (wait_interval_->mode == WaitInterval::THROTTLED) {
@@ skipping 23 matching lines @@
   pending_wakeup_timer_.Stop();
   pending_configure_params_.reset();
   if (started_)
     started_ = false;
 }
 
 // This is the only place where we invoke DoSyncSessionJob with canary
 // privileges.  Everyone else should use NORMAL_PRIORITY.
 void SyncSchedulerImpl::TryCanaryJob() {
   next_sync_session_job_priority_ = CANARY_PRIORITY;
+  SDVLOG(2) << "Attempting canary job";
   TrySyncSessionJob();
 }
 
 void SyncSchedulerImpl::TrySyncSessionJob() {
   // Post call to TrySyncSessionJobImpl on current thread. Later request for
   // access token will be here.
   base::MessageLoop::current()->PostTask(FROM_HERE, base::Bind(
       &SyncSchedulerImpl::TrySyncSessionJobImpl,
       weak_ptr_factory_.GetWeakPtr()));
 }
 
 void SyncSchedulerImpl::TrySyncSessionJobImpl() {
   JobPriority priority = next_sync_session_job_priority_;
   next_sync_session_job_priority_ = NORMAL_PRIORITY;
 
   nudge_tracker_.SetSyncCycleStartTime(base::TimeTicks::Now());
 
   DCHECK(CalledOnValidThread());
   if (mode_ == CONFIGURATION_MODE) {
     if (pending_configure_params_) {
       SDVLOG(2) << "Found pending configure job";
       DoConfigurationSyncSessionJob(priority);
     }
   } else if (CanRunNudgeJobNow(priority)) {
     if (nudge_tracker_.IsSyncRequired()) {
       SDVLOG(2) << "Found pending nudge job";
       DoNudgeSyncSessionJob(priority);
-    } else if (do_poll_after_credentials_updated_ ||
-        ((base::TimeTicks::Now() - last_poll_reset_) >= GetPollInterval())) {
+    } else if (((base::TimeTicks::Now() - last_poll_reset_) >=
+                GetPollInterval())) {
+      SDVLOG(2) << "Found pending poll";
       DoPollSyncSessionJob();
-      // Poll timer fires infrequently. Usually by this time access token is
-      // already expired and poll job will fail with auth error. Set flag to
-      // retry poll once ProfileSyncService gets new access token, TryCanaryJob
-      // will be called after access token is retrieved.
-      if (HttpResponse::SYNC_AUTH_ERROR ==
-          session_context_->connection_manager()->server_status()) {
-        do_poll_after_credentials_updated_ = true;
-      }
     }
-  }
-
-  if (priority == CANARY_PRIORITY) {
-    // If this is canary job then whatever result was don't run poll job till
-    // the next time poll timer fires.
-    do_poll_after_credentials_updated_ = false;
+  } else {
+    // We must be in an error state. Transitioning out of each of these
+    // error states should trigger a canary job.
+    DCHECK(IsCurrentlyThrottled() || IsBackingOff() ||
+           session_context_->connection_manager()->HasInvalidAuthToken());
   }
 
   if (IsBackingOff() && !pending_wakeup_timer_.IsRunning()) {
     // If we succeeded, our wait interval would have been cleared.  If it hasn't
     // been cleared, then we should increase our backoff interval and schedule
     // another retry.
     TimeDelta length = delay_provider_->GetDelay(wait_interval_->length);
     wait_interval_.reset(
-      new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, length));
+        new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, length));
     SDVLOG(2) << "Sync cycle failed.  Will back off for "
         << wait_interval_->length.InMilliseconds() << "ms.";
     RestartWaiting();
   }
 }
 
 void SyncSchedulerImpl::PollTimerCallback() {
   DCHECK(CalledOnValidThread());
-  if (no_scheduling_allowed_) {
-    // The no_scheduling_allowed_ flag is set by a function-scoped AutoReset in
-    // functions that are called only on the sync thread.  This function is also
-    // called only on the sync thread, and only when it is posted by an expiring
-    // timer.  If we find that no_scheduling_allowed_ is set here, then
-    // something is very wrong.  Maybe someone mistakenly called us directly, or
-    // mishandled the book-keeping for no_scheduling_allowed_.
-    NOTREACHED() << "Illegal to schedule job while session in progress.";
-    return;
-  }
+  CHECK(!syncer_->IsSyncing());
 
   TrySyncSessionJob();
 }
 
 void SyncSchedulerImpl::RetryTimerCallback() {
   TrySyncSessionJob();
 }
 
 void SyncSchedulerImpl::Unthrottle() {
   DCHECK(CalledOnValidThread());
@@ skipping 75 matching lines @@
                                         throttle_duration));
   NotifyRetryTime(base::Time::Now() + wait_interval_->length);
   NotifyThrottledTypesChanged(ModelTypeSet::All());
 }
 
 void SyncSchedulerImpl::OnTypesThrottled(
     ModelTypeSet types,
     const base::TimeDelta& throttle_duration) {
   base::TimeTicks now = base::TimeTicks::Now();
 
+  SDVLOG(1) << "Throttling " << ModelTypeSetToString(types) << " for "
+            << throttle_duration.InMinutes() << " minutes.";
+
   nudge_tracker_.SetTypesThrottledUntil(types, throttle_duration, now);
   base::TimeDelta time_until_next_unthrottle =
       nudge_tracker_.GetTimeUntilNextUnthrottle(now);
   type_unthrottle_timer_.Start(
       FROM_HERE,
       time_until_next_unthrottle,
       base::Bind(&SyncSchedulerImpl::TypeUnthrottle,
                  weak_ptr_factory_.GetWeakPtr(),
                  now + time_until_next_unthrottle));
   NotifyThrottledTypesChanged(nudge_tracker_.GetThrottledTypes());
 }
 
 bool SyncSchedulerImpl::IsCurrentlyThrottled() {
   DCHECK(CalledOnValidThread());
   return wait_interval_.get() && wait_interval_->mode ==
       WaitInterval::THROTTLED;
 }
 
 void SyncSchedulerImpl::OnReceivedShortPollIntervalUpdate(
     const base::TimeDelta& new_interval) {
   DCHECK(CalledOnValidThread());
+  if (new_interval == syncer_short_poll_interval_seconds_)
+    return;
+  SDVLOG(1) << "Updating short poll interval to " << new_interval.InMinutes()
+            << " minutes.";
   syncer_short_poll_interval_seconds_ = new_interval;
+  AdjustPolling(UPDATE_INTERVAL);
 }
 
 void SyncSchedulerImpl::OnReceivedLongPollIntervalUpdate(
     const base::TimeDelta& new_interval) {
   DCHECK(CalledOnValidThread());
+  if (new_interval == syncer_long_poll_interval_seconds_)
+    return;
+  SDVLOG(1) << "Updating long poll interval to " << new_interval.InMinutes()
+            << " minutes.";
   syncer_long_poll_interval_seconds_ = new_interval;
+  AdjustPolling(UPDATE_INTERVAL);
 }
 
 void SyncSchedulerImpl::OnReceivedCustomNudgeDelays(
     const std::map<ModelType, base::TimeDelta>& nudge_delays) {
   DCHECK(CalledOnValidThread());
   nudge_tracker_.OnReceivedCustomNudgeDelays(nudge_delays);
 }
 
 void SyncSchedulerImpl::OnReceivedClientInvalidationHintBufferSize(int size) {
   if (size > 0)
@@ skipping 40 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // namespace syncer