sync/engine/sync_scheduler_impl.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "sync/engine/sync_scheduler_impl.h"
   6
   7 #include <algorithm>
   8 #include <cstring>
   9
  10 #include "base/bind.h"
  11 #include "base/bind_helpers.h"
  12 #include "base/compiler_specific.h"
  13 #include "base/location.h"
  14 #include "base/logging.h"
  15 #include "base/message_loop/message_loop.h"
  16 #include "sync/engine/backoff_delay_provider.h"
  17 #include "sync/engine/syncer.h"
  18 #include "sync/protocol/proto_enum_conversions.h"
  19 #include "sync/protocol/sync.pb.h"
  20 #include "sync/util/data_type_histogram.h"
  21 #include "sync/util/logging.h"
  22
  23 using base::TimeDelta;
  24 using base::TimeTicks;
  25
  26 namespace syncer {
  27
  28 using sessions::SyncSession;
  29 using sessions::SyncSessionSnapshot;
  30 using sync_pb::GetUpdatesCallerInfo;
  31
  32 namespace {
  33
  34 bool IsConfigRelatedUpdateSourceValue(
  35     GetUpdatesCallerInfo::GetUpdatesSource source) {
  36   switch (source) {
  37     case GetUpdatesCallerInfo::RECONFIGURATION:
  38     case GetUpdatesCallerInfo::MIGRATION:
  39     case GetUpdatesCallerInfo::NEW_CLIENT:
  40     case GetUpdatesCallerInfo::NEWLY_SUPPORTED_DATATYPE:
  41     case GetUpdatesCallerInfo::PROGRAMMATIC:
  42       return true;
  43     default:
  44       return false;
  45   }
  46 }
  47
  48 bool ShouldRequestEarlyExit(const SyncProtocolError& error) {
  49   switch (error.error_type) {
  50     case SYNC_SUCCESS:
  51     case MIGRATION_DONE:
  52     case THROTTLED:
  53     case TRANSIENT_ERROR:
  54       return false;
  55     case NOT_MY_BIRTHDAY:
  56     case CLEAR_PENDING:
  57     case DISABLED_BY_ADMIN:
  58     case USER_ROLLBACK:
  59       // If we send terminate sync early then |sync_cycle_ended| notification
  60       // would not be sent. If there were no actions then |ACTIONABLE_ERROR|
  61       // notification wouldnt be sent either. Then the UI layer would be left
  62       // waiting forever. So assert we would send something.
  63       DCHECK_NE(error.action, UNKNOWN_ACTION);
  64       return true;
  65     case INVALID_CREDENTIAL:
  66       // The notification for this is handled by PostAndProcessHeaders|.
  67       // Server does no have to send any action for this.
  68       return true;
  69     // Make the default a NOTREACHED. So if a new error is introduced we
  70     // think about its expected functionality.
  71     default:
  72       NOTREACHED();
  73       return false;
  74   }
  75 }
  76
  77 bool IsActionableError(
  78     const SyncProtocolError& error) {
  79   return (error.action != UNKNOWN_ACTION);
  80 }
  81
  82 }  // namespace
  83
  84 ConfigurationParams::ConfigurationParams()
  85     : source(GetUpdatesCallerInfo::UNKNOWN) {}
  86 ConfigurationParams::ConfigurationParams(
  87     const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource& source,
  88     ModelTypeSet types_to_download,
  89     const ModelSafeRoutingInfo& routing_info,
  90     const base::Closure& ready_task,
  91     const base::Closure& retry_task)
  92     : source(source),
  93       types_to_download(types_to_download),
  94       routing_info(routing_info),
  95       ready_task(ready_task),
  96       retry_task(retry_task) {
  97   DCHECK(!ready_task.is_null());
  98 }
  99 ConfigurationParams::~ConfigurationParams() {}
 100
 101 SyncSchedulerImpl::WaitInterval::WaitInterval()
 102     : mode(UNKNOWN) {}
 103
 104 SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
 105     : mode(mode), length(length) {}
 106
 107 SyncSchedulerImpl::WaitInterval::~WaitInterval() {}
 108
 109 #define ENUM_CASE(x) case x: return #x; break;
 110
 111 const char* SyncSchedulerImpl::WaitInterval::GetModeString(Mode mode) {
 112   switch (mode) {
 113     ENUM_CASE(UNKNOWN);
 114     ENUM_CASE(EXPONENTIAL_BACKOFF);
 115     ENUM_CASE(THROTTLED);
 116   }
 117   NOTREACHED();
 118   return "";
 119 }
 120
 121 GetUpdatesCallerInfo::GetUpdatesSource GetUpdatesFromNudgeSource(
 122     NudgeSource source) {
 123   switch (source) {
 124     case NUDGE_SOURCE_NOTIFICATION:
 125       return GetUpdatesCallerInfo::NOTIFICATION;
 126     case NUDGE_SOURCE_LOCAL:
 127       return GetUpdatesCallerInfo::LOCAL;
 128     case NUDGE_SOURCE_LOCAL_REFRESH:
 129       return GetUpdatesCallerInfo::DATATYPE_REFRESH;
 130     case NUDGE_SOURCE_UNKNOWN:
 131       return GetUpdatesCallerInfo::UNKNOWN;
 132     default:
 133       NOTREACHED();
 134       return GetUpdatesCallerInfo::UNKNOWN;
 135   }
 136 }
 137
 138 // Helper macros to log with the syncer thread name; useful when there
 139 // are multiple syncer threads involved.
 140
 141 #define SLOG(severity) LOG(severity) << name_ << ": "
 142
 143 #define SDVLOG(verbose_level) DVLOG(verbose_level) << name_ << ": "
 144
 145 #define SDVLOG_LOC(from_here, verbose_level)             \
 146   DVLOG_LOC(from_here, verbose_level) << name_ << ": "
 147
 148 SyncSchedulerImpl::SyncSchedulerImpl(const std::string& name,
 149                                      BackoffDelayProvider* delay_provider,
 150                                      sessions::SyncSessionContext* context,
 151                                      Syncer* syncer)
 152     : name_(name),
 153       started_(false),
 154       syncer_short_poll_interval_seconds_(
 155           TimeDelta::FromSeconds(kDefaultShortPollIntervalSeconds)),
 156       syncer_long_poll_interval_seconds_(
 157           TimeDelta::FromSeconds(kDefaultLongPollIntervalSeconds)),
 158       mode_(CONFIGURATION_MODE),
 159       delay_provider_(delay_provider),
 160       syncer_(syncer),
 161       session_context_(context),
 162       next_sync_session_job_priority_(NORMAL_PRIORITY),
 163       weak_ptr_factory_(this),
 164       weak_ptr_factory_for_weak_handle_(this) {
 165   weak_handle_this_ = MakeWeakHandle(
 166       weak_ptr_factory_for_weak_handle_.GetWeakPtr());
 167 }
 168
 169 SyncSchedulerImpl::~SyncSchedulerImpl() {
 170   DCHECK(CalledOnValidThread());
 171   Stop();
 172 }
 173
 174 void SyncSchedulerImpl::OnCredentialsUpdated() {
 175   DCHECK(CalledOnValidThread());
 176
 177   if (HttpResponse::SYNC_AUTH_ERROR ==
 178       session_context_->connection_manager()->server_status()) {
 179     OnServerConnectionErrorFixed();
 180   }
 181 }
 182
 183 void SyncSchedulerImpl::OnConnectionStatusChange() {
 184   if (HttpResponse::CONNECTION_UNAVAILABLE  ==
 185       session_context_->connection_manager()->server_status()) {
 186     // Optimistically assume that the connection is fixed and try
 187     // connecting.
 188     OnServerConnectionErrorFixed();
 189   }
 190 }
 191
 192 void SyncSchedulerImpl::OnServerConnectionErrorFixed() {
 193   // There could be a pending nudge or configuration job in several cases:
 194   //
 195   // 1. We're in exponential backoff.
 196   // 2. We're silenced / throttled.
 197   // 3. A nudge was saved previously due to not having a valid auth token.
 198   // 4. A nudge was scheduled + saved while in configuration mode.
 199   //
 200   // In all cases except (2), we want to retry contacting the server. We
 201   // call TryCanaryJob to achieve this, and note that nothing -- not even a
 202   // canary job -- can bypass a THROTTLED WaitInterval. The only thing that
 203   // has the authority to do that is the Unthrottle timer.
 204   TryCanaryJob();
 205 }
 206
 207 void SyncSchedulerImpl::Start(Mode mode, base::Time last_poll_time) {
 208   DCHECK(CalledOnValidThread());
 209   std::string thread_name = base::MessageLoop::current()->thread_name();
 210   if (thread_name.empty())
 211     thread_name = "<Main thread>";
 212   SDVLOG(2) << "Start called from thread "
 213             << thread_name << " with mode " << GetModeString(mode);
 214   if (!started_) {
 215     started_ = true;
 216     SendInitialSnapshot();
 217   }
 218
 219   DCHECK(!session_context_->account_name().empty());
 220   DCHECK(syncer_.get());
 221   Mode old_mode = mode_;
 222   mode_ = mode;
 223   // Only adjust the poll reset time if it was valid and in the past.
 224   if (!last_poll_time.is_null() && last_poll_time < base::Time::Now()) {
 225     // Convert from base::Time to base::TimeTicks. The reason we use Time
 226     // for persisting is that TimeTicks can stop making forward progress when
 227     // the machine is suspended. This implies that on resume the client might
 228     // actually have miss the real poll, unless the client is restarted. Fixing
 229     // that would require using an AlarmTimer though, which is only supported
 230     // on certain platforms.
 231     last_poll_reset_ =
 232         base::TimeTicks::Now() - (base::Time::Now() - last_poll_time);
 233   }
 234
 235   if (old_mode != mode_ && mode_ == NORMAL_MODE) {
 236     // We just got back to normal mode.  Let's try to run the work that was
 237     // queued up while we were configuring.
 238
 239     AdjustPolling(UPDATE_INTERVAL);  // Will kick start poll timer if needed.
 240
 241     // Update our current time before checking IsRetryRequired().
 242     nudge_tracker_.SetSyncCycleStartTime(base::TimeTicks::Now());
 243     if (nudge_tracker_.IsSyncRequired() && CanRunNudgeJobNow(NORMAL_PRIORITY)) {
 244       TrySyncSessionJob();
 245     }
 246   }
 247 }
 248
 249 ModelTypeSet SyncSchedulerImpl::GetEnabledAndUnthrottledTypes() {
 250   ModelTypeSet enabled_types = session_context_->GetEnabledTypes();
 251   ModelTypeSet enabled_protocol_types =
 252       Intersection(ProtocolTypes(), enabled_types);
 253   ModelTypeSet throttled_types = nudge_tracker_.GetThrottledTypes();
 254   return Difference(enabled_protocol_types, throttled_types);
 255 }
 256
 257 void SyncSchedulerImpl::SendInitialSnapshot() {
 258   DCHECK(CalledOnValidThread());
 259   scoped_ptr<SyncSession> dummy(SyncSession::Build(session_context_, this));
 260   SyncCycleEvent event(SyncCycleEvent::STATUS_CHANGED);
 261   event.snapshot = dummy->TakeSnapshot();
 262   FOR_EACH_OBSERVER(SyncEngineEventListener,
 263                     *session_context_->listeners(),
 264                     OnSyncCycleEvent(event));
 265 }
 266
 267 namespace {
 268
 269 // Helper to extract the routing info corresponding to types in
 270 // |types_to_download| from |current_routes|.
 271 void BuildModelSafeParams(
 272     ModelTypeSet types_to_download,
 273     const ModelSafeRoutingInfo& current_routes,
 274     ModelSafeRoutingInfo* result_routes) {
 275   for (ModelTypeSet::Iterator iter = types_to_download.First(); iter.Good();
 276        iter.Inc()) {
 277     ModelType type = iter.Get();
 278     ModelSafeRoutingInfo::const_iterator route = current_routes.find(type);
 279     DCHECK(route != current_routes.end());
 280     ModelSafeGroup group = route->second;
 281     (*result_routes)[type] = group;
 282   }
 283 }
 284
 285 }  // namespace.
 286
 287 void SyncSchedulerImpl::ScheduleConfiguration(
 288     const ConfigurationParams& params) {
 289   DCHECK(CalledOnValidThread());
 290   DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
 291   DCHECK_EQ(CONFIGURATION_MODE, mode_);
 292   DCHECK(!params.ready_task.is_null());
 293   CHECK(started_) << "Scheduler must be running to configure.";
 294   SDVLOG(2) << "Reconfiguring syncer.";
 295
 296   // Only one configuration is allowed at a time. Verify we're not waiting
 297   // for a pending configure job.
 298   DCHECK(!pending_configure_params_);
 299
 300   ModelSafeRoutingInfo restricted_routes;
 301   BuildModelSafeParams(params.types_to_download,
 302                        params.routing_info,
 303                        &restricted_routes);
 304   session_context_->SetRoutingInfo(restricted_routes);
 305
 306   // Only reconfigure if we have types to download.
 307   if (!params.types_to_download.Empty()) {
 308     pending_configure_params_.reset(new ConfigurationParams(params));
 309     TrySyncSessionJob();
 310   } else {
 311     SDVLOG(2) << "No change in routing info, calling ready task directly.";
 312     params.ready_task.Run();
 313   }
 314 }
 315
 316 bool SyncSchedulerImpl::CanRunJobNow(JobPriority priority) {
 317   DCHECK(CalledOnValidThread());
 318   if (IsCurrentlyThrottled()) {
 319     SDVLOG(1) << "Unable to run a job because we're throttled.";
 320     return false;
 321   }
 322
 323   if (IsBackingOff() && priority != CANARY_PRIORITY) {
 324     SDVLOG(1) << "Unable to run a job because we're backing off.";
 325     return false;
 326   }
 327
 328   if (session_context_->connection_manager()->HasInvalidAuthToken()) {
 329     SDVLOG(1) << "Unable to run a job because we have no valid auth token.";
 330     return false;
 331   }
 332
 333   return true;
 334 }
 335
 336 bool SyncSchedulerImpl::CanRunNudgeJobNow(JobPriority priority) {
 337   DCHECK(CalledOnValidThread());
 338
 339   if (!CanRunJobNow(priority)) {
 340     SDVLOG(1) << "Unable to run a nudge job right now";
 341     return false;
 342   }
 343
 344   const ModelTypeSet enabled_types = session_context_->GetEnabledTypes();
 345   if (nudge_tracker_.GetThrottledTypes().HasAll(enabled_types)) {
 346     SDVLOG(1) << "Not running a nudge because we're fully type throttled.";
 347     return false;
 348   }
 349
 350   if (mode_ == CONFIGURATION_MODE) {
 351     SDVLOG(1) << "Not running nudge because we're in configuration mode.";
 352     return false;
 353   }
 354
 355   return true;
 356 }
 357
 358 void SyncSchedulerImpl::ScheduleLocalNudge(
 359     ModelTypeSet types,
 360     const tracked_objects::Location& nudge_location) {
 361   DCHECK(CalledOnValidThread());
 362   DCHECK(!types.Empty());
 363
 364   SDVLOG_LOC(nudge_location, 2)
 365       << "Scheduling sync because of local change to "
 366       << ModelTypeSetToString(types);
 367   UpdateNudgeTimeRecords(types);
 368   base::TimeDelta nudge_delay = nudge_tracker_.RecordLocalChange(types);
 369   ScheduleNudgeImpl(nudge_delay, nudge_location);
 370 }
 371
 372 void SyncSchedulerImpl::ScheduleLocalRefreshRequest(
 373     ModelTypeSet types,
 374     const tracked_objects::Location& nudge_location) {
 375   DCHECK(CalledOnValidThread());
 376   DCHECK(!types.Empty());
 377
 378   SDVLOG_LOC(nudge_location, 2)
 379       << "Scheduling sync because of local refresh request for "
 380       << ModelTypeSetToString(types);
 381   base::TimeDelta nudge_delay = nudge_tracker_.RecordLocalRefreshRequest(types);
 382   ScheduleNudgeImpl(nudge_delay, nudge_location);
 383 }
 384
 385 void SyncSchedulerImpl::ScheduleInvalidationNudge(
 386     syncer::ModelType model_type,
 387     scoped_ptr<InvalidationInterface> invalidation,
 388     const tracked_objects::Location& nudge_location) {
 389   DCHECK(CalledOnValidThread());
 390
 391   SDVLOG_LOC(nudge_location, 2)
 392       << "Scheduling sync because we received invalidation for "
 393       << ModelTypeToString(model_type);
 394   base::TimeDelta nudge_delay =
 395       nudge_tracker_.RecordRemoteInvalidation(model_type, invalidation.Pass());
 396   ScheduleNudgeImpl(nudge_delay, nudge_location);
 397 }
 398
 399 void SyncSchedulerImpl::ScheduleInitialSyncNudge(syncer::ModelType model_type) {
 400   DCHECK(CalledOnValidThread());
 401
 402   SDVLOG(2) << "Scheduling non-blocking initial sync for "
 403             << ModelTypeToString(model_type);
 404   nudge_tracker_.RecordInitialSyncRequired(model_type);
 405   ScheduleNudgeImpl(TimeDelta::FromSeconds(0), FROM_HERE);
 406 }
 407
 408 // TODO(zea): Consider adding separate throttling/backoff for datatype
 409 // refresh requests.
 410 void SyncSchedulerImpl::ScheduleNudgeImpl(
 411     const TimeDelta& delay,
 412     const tracked_objects::Location& nudge_location) {
 413   DCHECK(CalledOnValidThread());
 414   CHECK(!syncer_->IsSyncing());
 415
 416   if (!started_) {
 417     SDVLOG_LOC(nudge_location, 2)
 418         << "Dropping nudge, scheduler is not running.";
 419     return;
 420   }
 421
 422   SDVLOG_LOC(nudge_location, 2)
 423       << "In ScheduleNudgeImpl with delay "
 424       << delay.InMilliseconds() << " ms";
 425
 426   if (!CanRunNudgeJobNow(NORMAL_PRIORITY))
 427     return;
 428
 429   TimeTicks incoming_run_time = TimeTicks::Now() + delay;
 430   if (!scheduled_nudge_time_.is_null() &&
 431     (scheduled_nudge_time_ < incoming_run_time)) {
 432     // Old job arrives sooner than this one.  Don't reschedule it.
 433     return;
 434   }
 435
 436   // Either there is no existing nudge in flight or the incoming nudge should be
 437   // made to arrive first (preempt) the existing nudge.  We reschedule in either
 438   // case.
 439   SDVLOG_LOC(nudge_location, 2)
 440       << "Scheduling a nudge with "
 441       << delay.InMilliseconds() << " ms delay";
 442   scheduled_nudge_time_ = incoming_run_time;
 443   pending_wakeup_timer_.Start(
 444       nudge_location,
 445       delay,
 446       base::Bind(&SyncSchedulerImpl::PerformDelayedNudge,
 447                  weak_ptr_factory_.GetWeakPtr()));
 448 }
 449
 450 const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
 451   switch (mode) {
 452     ENUM_CASE(CONFIGURATION_MODE);
 453     ENUM_CASE(NORMAL_MODE);
 454   }
 455   return "";
 456 }
 457
 458 void SyncSchedulerImpl::SetDefaultNudgeDelay(base::TimeDelta delay_ms) {
 459   DCHECK(CalledOnValidThread());
 460   nudge_tracker_.SetDefaultNudgeDelay(delay_ms);
 461 }
 462
 463 void SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {
 464   DCHECK(CalledOnValidThread());
 465   DCHECK(CanRunNudgeJobNow(priority));
 466
 467   DVLOG(2) << "Will run normal mode sync cycle with types "
 468            << ModelTypeSetToString(session_context_->GetEnabledTypes());
 469   scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
 470   bool success = syncer_->NormalSyncShare(
 471       GetEnabledAndUnthrottledTypes(), &nudge_tracker_, session.get());
 472
 473   if (success) {
 474     // That cycle took care of any outstanding work we had.
 475     SDVLOG(2) << "Nudge succeeded.";
 476     nudge_tracker_.RecordSuccessfulSyncCycle();
 477     scheduled_nudge_time_ = base::TimeTicks();
 478     HandleSuccess();
 479
 480     // If this was a canary, we may need to restart the poll timer (the poll
 481     // timer may have fired while the scheduler was in an error state, ignoring
 482     // the poll).
 483     if (!poll_timer_.IsRunning()) {
 484       SDVLOG(1) << "Canary succeeded, restarting polling.";
 485       AdjustPolling(UPDATE_INTERVAL);
 486     }
 487   } else {
 488     HandleFailure(session->status_controller().model_neutral_state());
 489   }
 490 }
 491
 492 void SyncSchedulerImpl::DoConfigurationSyncSessionJob(JobPriority priority) {
 493   DCHECK(CalledOnValidThread());
 494   DCHECK_EQ(mode_, CONFIGURATION_MODE);
 495   DCHECK(pending_configure_params_ != NULL);
 496
 497   if (!CanRunJobNow(priority)) {
 498     SDVLOG(2) << "Unable to run configure job right now.";
 499     if (!pending_configure_params_->retry_task.is_null()) {
 500       pending_configure_params_->retry_task.Run();
 501       pending_configure_params_->retry_task.Reset();
 502     }
 503     return;
 504   }
 505
 506   SDVLOG(2) << "Will run configure SyncShare with types "
 507             << ModelTypeSetToString(session_context_->GetEnabledTypes());
 508   scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
 509   bool success = syncer_->ConfigureSyncShare(
 510       pending_configure_params_->types_to_download,
 511       pending_configure_params_->source,
 512       session.get());
 513
 514   if (success) {
 515     SDVLOG(2) << "Configure succeeded.";
 516     pending_configure_params_->ready_task.Run();
 517     pending_configure_params_.reset();
 518     HandleSuccess();
 519   } else {
 520     HandleFailure(session->status_controller().model_neutral_state());
 521     // Sync cycle might receive response from server that causes scheduler to
 522     // stop and draws pending_configure_params_ invalid.
 523     if (started_ && !pending_configure_params_->retry_task.is_null()) {
 524       pending_configure_params_->retry_task.Run();
 525       pending_configure_params_->retry_task.Reset();
 526     }
 527   }
 528 }
 529
 530 void SyncSchedulerImpl::HandleSuccess() {
 531   // If we're here, then we successfully reached the server.  End all backoff.
 532   wait_interval_.reset();
 533   NotifyRetryTime(base::Time());
 534 }
 535
 536 void SyncSchedulerImpl::HandleFailure(
 537     const sessions::ModelNeutralState& model_neutral_state) {
 538   if (IsCurrentlyThrottled()) {
 539     SDVLOG(2) << "Was throttled during previous sync cycle.";
 540   } else if (!IsBackingOff()) {
 541     // Setup our backoff if this is our first such failure.
 542     TimeDelta length = delay_provider_->GetDelay(
 543         delay_provider_->GetInitialDelay(model_neutral_state));
 544     wait_interval_.reset(
 545         new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, length));
 546     SDVLOG(2) << "Sync cycle failed.  Will back off for "
 547         << wait_interval_->length.InMilliseconds() << "ms.";
 548   } else {
 549     // Increase our backoff interval and schedule another retry.
 550     TimeDelta length = delay_provider_->GetDelay(wait_interval_->length);
 551     wait_interval_.reset(
 552         new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, length));
 553     SDVLOG(2) << "Sync cycle failed.  Will back off for "
 554         << wait_interval_->length.InMilliseconds() << "ms.";
 555   }
 556   RestartWaiting();
 557 }
 558
 559 void SyncSchedulerImpl::DoPollSyncSessionJob() {
 560   SDVLOG(2) << "Polling with types "
 561             << ModelTypeSetToString(GetEnabledAndUnthrottledTypes());
 562   scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
 563   bool success = syncer_->PollSyncShare(
 564       GetEnabledAndUnthrottledTypes(),
 565       session.get());
 566
 567   // Only restart the timer if the poll succeeded. Otherwise rely on normal
 568   // failure handling to retry with backoff.
 569   if (success) {
 570     AdjustPolling(FORCE_RESET);
 571     HandleSuccess();
 572   } else {
 573     HandleFailure(session->status_controller().model_neutral_state());
 574   }
 575 }
 576
 577 void SyncSchedulerImpl::UpdateNudgeTimeRecords(ModelTypeSet types) {
 578   DCHECK(CalledOnValidThread());
 579   base::TimeTicks now = TimeTicks::Now();
 580   // Update timing information for how often datatypes are triggering nudges.
 581   for (ModelTypeSet::Iterator iter = types.First(); iter.Good(); iter.Inc()) {
 582     base::TimeTicks previous = last_local_nudges_by_model_type_[iter.Get()];
 583     last_local_nudges_by_model_type_[iter.Get()] = now;
 584     if (previous.is_null())
 585       continue;
 586
 587 #define PER_DATA_TYPE_MACRO(type_str) \
 588     SYNC_FREQ_HISTOGRAM("Sync.Freq" type_str, now - previous);
 589     SYNC_DATA_TYPE_HISTOGRAM(iter.Get());
 590 #undef PER_DATA_TYPE_MACRO
 591   }
 592 }
 593
 594 TimeDelta SyncSchedulerImpl::GetPollInterval() {
 595   return (!session_context_->notifications_enabled() ||
 596           !session_context_->ShouldFetchUpdatesBeforeCommit()) ?
 597       syncer_short_poll_interval_seconds_ :
 598       syncer_long_poll_interval_seconds_;
 599 }
 600
 601 void SyncSchedulerImpl::AdjustPolling(PollAdjustType type) {
 602   DCHECK(CalledOnValidThread());
 603   if (!started_)
 604     return;
 605
 606   TimeDelta poll_interval = GetPollInterval();
 607   TimeDelta poll_delay = poll_interval;
 608   const TimeTicks now = TimeTicks::Now();
 609
 610   if (type == UPDATE_INTERVAL) {
 611     if (!last_poll_reset_.is_null()) {
 612       // Override the delay based on the last successful poll time (if it was
 613       // set).
 614       TimeTicks new_poll_time = poll_interval + last_poll_reset_;
 615       poll_delay = new_poll_time - TimeTicks::Now();
 616
 617       if (poll_delay < TimeDelta()) {
 618         // The desired poll time was in the past, so trigger a poll now (the
 619         // timer will post the task asynchronously, so re-entrancy isn't an
 620         // issue).
 621         poll_delay = TimeDelta();
 622       }
 623     } else {
 624       // There was no previous poll. Keep the delay set to the normal interval,
 625       // as if we had just completed a poll.
 626       DCHECK_EQ(GetPollInterval(), poll_delay);
 627       last_poll_reset_ = now;
 628     }
 629   } else {
 630     // Otherwise just restart the timer.
 631     DCHECK_EQ(FORCE_RESET, type);
 632     DCHECK_EQ(GetPollInterval(), poll_delay);
 633     last_poll_reset_ = now;
 634   }
 635
 636   SDVLOG(1) << "Updating polling delay to " << poll_delay.InMinutes()
 637             << " minutes.";
 638
 639   // Adjust poll rate. Start will reset the timer if it was already running.
 640   poll_timer_.Start(FROM_HERE, poll_delay, this,
 641                     &SyncSchedulerImpl::PollTimerCallback);
 642 }
 643
 644 void SyncSchedulerImpl::RestartWaiting() {
 645   CHECK(wait_interval_.get());
 646   DCHECK(wait_interval_->length >= TimeDelta::FromSeconds(0));
 647   NotifyRetryTime(base::Time::Now() + wait_interval_->length);
 648   SDVLOG(2) << "Starting WaitInterval timer of length "
 649       << wait_interval_->length.InMilliseconds() << "ms.";
 650   if (wait_interval_->mode == WaitInterval::THROTTLED) {
 651     pending_wakeup_timer_.Start(
 652         FROM_HERE,
 653         wait_interval_->length,
 654         base::Bind(&SyncSchedulerImpl::Unthrottle,
 655                    weak_ptr_factory_.GetWeakPtr()));
 656   } else {
 657     pending_wakeup_timer_.Start(
 658         FROM_HERE,
 659         wait_interval_->length,
 660         base::Bind(&SyncSchedulerImpl::ExponentialBackoffRetry,
 661                    weak_ptr_factory_.GetWeakPtr()));
 662   }
 663 }
 664
 665 void SyncSchedulerImpl::Stop() {
 666   DCHECK(CalledOnValidThread());
 667   SDVLOG(2) << "Stop called";
 668
 669   // Kill any in-flight method calls.
 670   weak_ptr_factory_.InvalidateWeakPtrs();
 671   wait_interval_.reset();
 672   NotifyRetryTime(base::Time());
 673   poll_timer_.Stop();
 674   pending_wakeup_timer_.Stop();
 675   pending_configure_params_.reset();
 676   if (started_)
 677     started_ = false;
 678 }
 679
 680 // This is the only place where we invoke DoSyncSessionJob with canary
 681 // privileges.  Everyone else should use NORMAL_PRIORITY.
 682 void SyncSchedulerImpl::TryCanaryJob() {
 683   next_sync_session_job_priority_ = CANARY_PRIORITY;
 684   SDVLOG(2) << "Attempting canary job";
 685   TrySyncSessionJob();
 686 }
 687
 688 void SyncSchedulerImpl::TrySyncSessionJob() {
 689   // Post call to TrySyncSessionJobImpl on current thread. Later request for
 690   // access token will be here.
 691   base::MessageLoop::current()->PostTask(FROM_HERE, base::Bind(
 692       &SyncSchedulerImpl::TrySyncSessionJobImpl,
 693       weak_ptr_factory_.GetWeakPtr()));
 694 }
 695
 696 void SyncSchedulerImpl::TrySyncSessionJobImpl() {
 697   JobPriority priority = next_sync_session_job_priority_;
 698   next_sync_session_job_priority_ = NORMAL_PRIORITY;
 699
 700   nudge_tracker_.SetSyncCycleStartTime(base::TimeTicks::Now());
 701
 702   DCHECK(CalledOnValidThread());
 703   if (mode_ == CONFIGURATION_MODE) {
 704     if (pending_configure_params_) {
 705       SDVLOG(2) << "Found pending configure job";
 706       DoConfigurationSyncSessionJob(priority);
 707     }
 708   } else if (CanRunNudgeJobNow(priority)) {
 709     if (nudge_tracker_.IsSyncRequired()) {
 710       SDVLOG(2) << "Found pending nudge job";
 711       DoNudgeSyncSessionJob(priority);
 712     } else if (((base::TimeTicks::Now() - last_poll_reset_) >=
 713                 GetPollInterval())) {
 714       SDVLOG(2) << "Found pending poll";
 715       DoPollSyncSessionJob();
 716     }
 717   } else {
 718     // We must be in an error state. Transitioning out of each of these
 719     // error states should trigger a canary job.
 720     DCHECK(IsCurrentlyThrottled() || IsBackingOff() ||
 721            session_context_->connection_manager()->HasInvalidAuthToken());
 722   }
 723
 724   if (IsBackingOff() && !pending_wakeup_timer_.IsRunning()) {
 725     // If we succeeded, our wait interval would have been cleared.  If it hasn't
 726     // been cleared, then we should increase our backoff interval and schedule
 727     // another retry.
 728     TimeDelta length = delay_provider_->GetDelay(wait_interval_->length);
 729     wait_interval_.reset(
 730         new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, length));
 731     SDVLOG(2) << "Sync cycle failed.  Will back off for "
 732         << wait_interval_->length.InMilliseconds() << "ms.";
 733     RestartWaiting();
 734   }
 735 }
 736
 737 void SyncSchedulerImpl::PollTimerCallback() {
 738   DCHECK(CalledOnValidThread());
 739   CHECK(!syncer_->IsSyncing());
 740
 741   TrySyncSessionJob();
 742 }
 743
 744 void SyncSchedulerImpl::RetryTimerCallback() {
 745   TrySyncSessionJob();
 746 }
 747
 748 void SyncSchedulerImpl::Unthrottle() {
 749   DCHECK(CalledOnValidThread());
 750   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
 751
 752   // We're no longer throttled, so clear the wait interval.
 753   wait_interval_.reset();
 754   NotifyRetryTime(base::Time());
 755   NotifyThrottledTypesChanged(nudge_tracker_.GetThrottledTypes());
 756
 757   // We treat this as a 'canary' in the sense that it was originally scheduled
 758   // to run some time ago, failed, and we now want to retry, versus a job that
 759   // was just created (e.g via ScheduleNudgeImpl). The main implication is
 760   // that we're careful to update routing info (etc) with such potentially
 761   // stale canary jobs.
 762   TryCanaryJob();
 763 }
 764
 765 void SyncSchedulerImpl::TypeUnthrottle(base::TimeTicks unthrottle_time) {
 766   DCHECK(CalledOnValidThread());
 767   nudge_tracker_.UpdateTypeThrottlingState(unthrottle_time);
 768   NotifyThrottledTypesChanged(nudge_tracker_.GetThrottledTypes());
 769
 770   if (nudge_tracker_.IsAnyTypeThrottled()) {
 771     const base::TimeTicks now = base::TimeTicks::Now();
 772     base::TimeDelta time_until_next_unthrottle =
 773         nudge_tracker_.GetTimeUntilNextUnthrottle(now);
 774     type_unthrottle_timer_.Start(
 775         FROM_HERE,
 776         time_until_next_unthrottle,
 777         base::Bind(&SyncSchedulerImpl::TypeUnthrottle,
 778                    weak_ptr_factory_.GetWeakPtr(),
 779                    now + time_until_next_unthrottle));
 780   }
 781
 782   // Maybe this is a good time to run a nudge job.  Let's try it.
 783   if (nudge_tracker_.IsSyncRequired() && CanRunNudgeJobNow(NORMAL_PRIORITY))
 784     TrySyncSessionJob();
 785 }
 786
 787 void SyncSchedulerImpl::PerformDelayedNudge() {
 788   // Circumstances may have changed since we scheduled this delayed nudge.
 789   // We must check to see if it's OK to run the job before we do so.
 790   if (CanRunNudgeJobNow(NORMAL_PRIORITY))
 791     TrySyncSessionJob();
 792
 793   // We're not responsible for setting up any retries here.  The functions that
 794   // first put us into a state that prevents successful sync cycles (eg. global
 795   // throttling, type throttling, network errors, transient errors) will also
 796   // setup the appropriate retry logic (eg. retry after timeout, exponential
 797   // backoff, retry when the network changes).
 798 }
 799
 800 void SyncSchedulerImpl::ExponentialBackoffRetry() {
 801   TryCanaryJob();
 802 }
 803
 804 void SyncSchedulerImpl::NotifyRetryTime(base::Time retry_time) {
 805   FOR_EACH_OBSERVER(SyncEngineEventListener,
 806                     *session_context_->listeners(),
 807                     OnRetryTimeChanged(retry_time));
 808 }
 809
 810 void SyncSchedulerImpl::NotifyThrottledTypesChanged(ModelTypeSet types) {
 811   FOR_EACH_OBSERVER(SyncEngineEventListener,
 812                     *session_context_->listeners(),
 813                     OnThrottledTypesChanged(types));
 814 }
 815
 816 bool SyncSchedulerImpl::IsBackingOff() const {
 817   DCHECK(CalledOnValidThread());
 818   return wait_interval_.get() && wait_interval_->mode ==
 819       WaitInterval::EXPONENTIAL_BACKOFF;
 820 }
 821
 822 void SyncSchedulerImpl::OnThrottled(const base::TimeDelta& throttle_duration) {
 823   DCHECK(CalledOnValidThread());
 824   wait_interval_.reset(new WaitInterval(WaitInterval::THROTTLED,
 825                                         throttle_duration));
 826   NotifyRetryTime(base::Time::Now() + wait_interval_->length);
 827   NotifyThrottledTypesChanged(ModelTypeSet::All());
 828 }
 829
 830 void SyncSchedulerImpl::OnTypesThrottled(
 831     ModelTypeSet types,
 832     const base::TimeDelta& throttle_duration) {
 833   base::TimeTicks now = base::TimeTicks::Now();
 834
 835   SDVLOG(1) << "Throttling " << ModelTypeSetToString(types) << " for "
 836             << throttle_duration.InMinutes() << " minutes.";
 837
 838   nudge_tracker_.SetTypesThrottledUntil(types, throttle_duration, now);
 839   base::TimeDelta time_until_next_unthrottle =
 840       nudge_tracker_.GetTimeUntilNextUnthrottle(now);
 841   type_unthrottle_timer_.Start(
 842       FROM_HERE,
 843       time_until_next_unthrottle,
 844       base::Bind(&SyncSchedulerImpl::TypeUnthrottle,
 845                  weak_ptr_factory_.GetWeakPtr(),
 846                  now + time_until_next_unthrottle));
 847   NotifyThrottledTypesChanged(nudge_tracker_.GetThrottledTypes());
 848 }
 849
 850 bool SyncSchedulerImpl::IsCurrentlyThrottled() {
 851   DCHECK(CalledOnValidThread());
 852   return wait_interval_.get() && wait_interval_->mode ==
 853       WaitInterval::THROTTLED;
 854 }
 855
 856 void SyncSchedulerImpl::OnReceivedShortPollIntervalUpdate(
 857     const base::TimeDelta& new_interval) {
 858   DCHECK(CalledOnValidThread());
 859   if (new_interval == syncer_short_poll_interval_seconds_)
 860     return;
 861   SDVLOG(1) << "Updating short poll interval to " << new_interval.InMinutes()
 862             << " minutes.";
 863   syncer_short_poll_interval_seconds_ = new_interval;
 864   AdjustPolling(UPDATE_INTERVAL);
 865 }
 866
 867 void SyncSchedulerImpl::OnReceivedLongPollIntervalUpdate(
 868     const base::TimeDelta& new_interval) {
 869   DCHECK(CalledOnValidThread());
 870   if (new_interval == syncer_long_poll_interval_seconds_)
 871     return;
 872   SDVLOG(1) << "Updating long poll interval to " << new_interval.InMinutes()
 873             << " minutes.";
 874   syncer_long_poll_interval_seconds_ = new_interval;
 875   AdjustPolling(UPDATE_INTERVAL);
 876 }
 877
 878 void SyncSchedulerImpl::OnReceivedCustomNudgeDelays(
 879     const std::map<ModelType, base::TimeDelta>& nudge_delays) {
 880   DCHECK(CalledOnValidThread());
 881   nudge_tracker_.OnReceivedCustomNudgeDelays(nudge_delays);
 882 }
 883
 884 void SyncSchedulerImpl::OnReceivedClientInvalidationHintBufferSize(int size) {
 885   if (size > 0)
 886     nudge_tracker_.SetHintBufferSize(size);
 887   else
 888     NOTREACHED() << "Hint buffer size should be > 0.";
 889 }
 890
 891 void SyncSchedulerImpl::OnSyncProtocolError(
 892     const SyncProtocolError& sync_protocol_error) {
 893   DCHECK(CalledOnValidThread());
 894   if (ShouldRequestEarlyExit(sync_protocol_error)) {
 895     SDVLOG(2) << "Sync Scheduler requesting early exit.";
 896     Stop();
 897   }
 898   if (IsActionableError(sync_protocol_error)) {
 899     SDVLOG(2) << "OnActionableError";
 900     FOR_EACH_OBSERVER(SyncEngineEventListener,
 901                       *session_context_->listeners(),
 902                       OnActionableError(sync_protocol_error));
 903   }
 904 }
 905
 906 void SyncSchedulerImpl::OnReceivedGuRetryDelay(const base::TimeDelta& delay) {
 907   nudge_tracker_.SetNextRetryTime(TimeTicks::Now() + delay);
 908   retry_timer_.Start(FROM_HERE, delay, this,
 909                      &SyncSchedulerImpl::RetryTimerCallback);
 910 }
 911
 912 void SyncSchedulerImpl::OnReceivedMigrationRequest(ModelTypeSet types) {
 913     FOR_EACH_OBSERVER(SyncEngineEventListener,
 914                       *session_context_->listeners(),
 915                       OnMigrationRequested(types));
 916 }
 917
 918 void SyncSchedulerImpl::SetNotificationsEnabled(bool notifications_enabled) {
 919   DCHECK(CalledOnValidThread());
 920   session_context_->set_notifications_enabled(notifications_enabled);
 921   if (notifications_enabled)
 922     nudge_tracker_.OnInvalidationsEnabled();
 923   else
 924     nudge_tracker_.OnInvalidationsDisabled();
 925 }
 926
 927 #undef SDVLOG_LOC
 928
 929 #undef SDVLOG
 930
 931 #undef SLOG
 932
 933 #undef ENUM_CASE
 934
 935 }  // namespace syncer