[Sync] Make BackendMigrator not wait for full sync cycles

chrome/browser/sync/engine/sync_scheduler.cc

 // Copyright (c) 2011 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 "chrome/browser/sync/engine/sync_scheduler.h"
 
 #include <algorithm>
 #include <cstring>
 
 #include "base/compiler_specific.h"
@@ skipping 40 matching lines @@
   return "";
 }
 
 SyncScheduler::SyncSessionJob::SyncSessionJob()
     : purpose(UNKNOWN),
       is_canary_job(false) {
 }
 
 SyncScheduler::SyncSessionJob::~SyncSessionJob() {}
 
-SyncScheduler::SyncSessionJob::SyncSessionJob(SyncSessionJobPurpose purpose,
+SyncScheduler::SyncSessionJob::SyncSessionJob(
+    SyncSessionJobPurpose purpose,
     base::TimeTicks start,
     linked_ptr<sessions::SyncSession> session, bool is_canary_job,
-    const tracked_objects::Location& from_here) : purpose(purpose),
-        scheduled_start(start),
-        session(session),
-        is_canary_job(is_canary_job),
-        from_here(from_here) {
+    const base::Closure& on_success,
+    const tracked_objects::Location& from_here)
+  : purpose(purpose),
+    scheduled_start(start),
+    session(session),
+    is_canary_job(is_canary_job),
+    on_success(on_success),
+    from_here(from_here) {
 }
 
 const char* SyncScheduler::SyncSessionJob::GetPurposeString(
     SyncScheduler::SyncSessionJob::SyncSessionJobPurpose purpose) {
   switch (purpose) {
     ENUM_CASE(UNKNOWN);
     ENUM_CASE(POLL);
     ENUM_CASE(NUDGE);
     ENUM_CASE(CLEAR_USER_DATA);
     ENUM_CASE(CONFIGURATION);
@@ skipping 113 matching lines @@
     DoCanaryJob();
   }
 
   if (old_server_connection_ok != server_connection_ok_) {
     const char* transition =
         server_connection_ok_ ? "down -> up" : "up -> down";
     SVLOG(2) << "Server connection changed: " << transition;
   }
 }
 
-void SyncScheduler::Start(Mode mode, ModeChangeCallback* callback) {
+void SyncScheduler::Start(Mode mode, const base::Closure& callback) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   std::string thread_name = MessageLoop::current()->thread_name();
   if (thread_name.empty())
     thread_name = "<Main thread>";
   SVLOG(2) << "Start called from thread "
            << thread_name << " with mode " << GetModeString(mode);
   if (!started_) {
     WatchConnectionManager();
     PostTask(FROM_HERE, "SendInitialSnapshot",
              method_factory_.NewRunnableMethod(
                  &SyncScheduler::SendInitialSnapshot));
   }
   started_ = true;
-  // TODO(sync): This will leak if StartImpl is never run.  Fix this.
-  // Might be easiest to just use base::Callback.
   PostTask(FROM_HERE, "StartImpl",
            method_factory_.NewRunnableMethod(
                &SyncScheduler::StartImpl, mode, callback));
 }
 
 void SyncScheduler::SendInitialSnapshot() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   scoped_ptr<SyncSession> dummy(new SyncSession(session_context_.get(), this,
       SyncSourceInfo(), ModelSafeRoutingInfo(),
       std::vector<ModelSafeWorker*>()));
   SyncEngineEvent event(SyncEngineEvent::STATUS_CHANGED);
   sessions::SyncSessionSnapshot snapshot(dummy->TakeSnapshot());
   event.snapshot = &snapshot;
   session_context_->NotifyListeners(event);
 }
 
 void SyncScheduler::WatchConnectionManager() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ServerConnectionManager* scm = session_context_->connection_manager();
   PostTask(FROM_HERE, "CheckServerConnectionManagerStatus",
            method_factory_.NewRunnableMethod(
                &SyncScheduler::CheckServerConnectionManagerStatus,
                scm->server_status()));
   scm->AddListener(this);
 }
 
-void SyncScheduler::StartImpl(Mode mode, ModeChangeCallback* callback) {
+void SyncScheduler::StartImpl(Mode mode, const base::Closure& callback) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SVLOG(2) << "In StartImpl with mode " << GetModeString(mode);
 
-  scoped_ptr<ModeChangeCallback> scoped_callback(callback);
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(!session_context_->account_name().empty());
   DCHECK(syncer_.get());
   Mode old_mode = mode_;
   mode_ = mode;
   AdjustPolling(NULL);  // Will kick start poll timer if needed.
-  if (scoped_callback.get())
-    scoped_callback->Run();
+  if (!callback.is_null())
+    callback.Run();
 
   if (old_mode != mode_) {
     // We just changed our mode. See if there are any pending jobs that we could
     // execute in the new mode.
     DoPendingJobIfPossible(false);
   }
 }
 
 SyncScheduler::JobProcessDecision SyncScheduler::DecideWhileInWaitInterval(
     const SyncSessionJob& job) {
@@ skipping 69 matching lines @@
 void SyncScheduler::InitOrCoalescePendingJob(const SyncSessionJob& job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(job.purpose != SyncSessionJob::CONFIGURATION);
   if (pending_nudge_.get() == NULL) {
     SVLOG(2) << "Creating a pending nudge job";
     SyncSession* s = job.session.get();
     scoped_ptr<SyncSession> session(new SyncSession(s->context(),
         s->delegate(), s->source(), s->routing_info(), s->workers()));
 
     SyncSessionJob new_job(SyncSessionJob::NUDGE, job.scheduled_start,
-        make_linked_ptr(session.release()), false, job.from_here);
+                           make_linked_ptr(session.release()),
+                           false, base::Closure(), job.from_here);
     pending_nudge_.reset(new SyncSessionJob(new_job));
 
     return;
   }
 
   SVLOG(2) << "Coalescing a pending nudge";
   pending_nudge_->session->Coalesce(*(job.session.get()));
   pending_nudge_->scheduled_start = job.scheduled_start;
 
   // Unfortunately the nudge location cannot be modified. So it stores the
@@ skipping 27 matching lines @@
     InitOrCoalescePendingJob(job);
   } else if (job.purpose == SyncSessionJob::CONFIGURATION){
     SVLOG(2) << "Saving a configuration job";
     DCHECK(wait_interval_.get());
     DCHECK(mode_ == CONFIGURATION_MODE);
 
     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.from_here);
+                           make_linked_ptr(s), false, base::Closure(),
+                           job.from_here);
     wait_interval_->pending_configure_job.reset(new SyncSessionJob(new_job));
   } // drop the rest.
   // TODO(sync): Is it okay to drop the rest?  It's weird that
   // SaveJob() only does what it says sometimes.  (See
   // http://crbug.com/90868.)
 }
 
 // Functor for std::find_if to search by ModelSafeGroup.
 struct ModelSafeWorkerGroupIs {
   explicit ModelSafeWorkerGroupIs(ModelSafeGroup group) : group(group) {}
   bool operator()(ModelSafeWorker* w) {
     return group == w->GetModelSafeGroup();
   }
   ModelSafeGroup group;
 };
 
 void SyncScheduler::ScheduleClearUserData() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   PostTask(FROM_HERE, "ScheduleClearUserDataImpl",
            method_factory_.NewRunnableMethod(
                &SyncScheduler::ScheduleClearUserDataImpl));
 }
 
-void SyncScheduler::ScheduleCleanupDisabledTypes() {
+void SyncScheduler::ScheduleCleanupDisabledTypes(
+    const base::Closure& callback) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   PostTask(FROM_HERE, "ScheduleCleanupDisabledTypes",
            method_factory_.NewRunnableMethod(
-               &SyncScheduler::ScheduleCleanupDisabledTypesImpl));
+               &SyncScheduler::ScheduleCleanupDisabledTypesImpl, callback));
 }
 
 void SyncScheduler::ScheduleNudge(
     const TimeDelta& delay,
     NudgeSource source, const ModelTypeBitSet& types,
     const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SVLOG_LOC(nudge_location, 2)
       << "Nudge scheduled with delay " << delay.InMilliseconds() << " ms, "
       << "source " << GetNudgeSourceString(source) << ", "
@@ skipping 23 matching lines @@
            method_factory_.NewRunnableMethod(
                &SyncScheduler::ScheduleNudgeImpl, delay,
                GetUpdatesFromNudgeSource(source), types_with_payloads, false,
                nudge_location));
 }
 
 void SyncScheduler::ScheduleClearUserDataImpl() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ScheduleSyncSessionJob(
       TimeDelta::FromSeconds(0), SyncSessionJob::CLEAR_USER_DATA,
-      CreateSyncSession(SyncSourceInfo()), FROM_HERE);
+      CreateSyncSession(SyncSourceInfo()), base::Closure(), FROM_HERE);
 }
 
-void SyncScheduler::ScheduleCleanupDisabledTypesImpl() {
+void SyncScheduler::ScheduleCleanupDisabledTypesImpl(
+    const base::Closure& callback) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ScheduleSyncSessionJob(
       TimeDelta::FromSeconds(0), SyncSessionJob::CLEANUP_DISABLED_TYPES,
-      CreateSyncSession(SyncSourceInfo()), FROM_HERE);
+      CreateSyncSession(SyncSourceInfo()), callback, FROM_HERE);
 }
 
 void SyncScheduler::ScheduleNudgeImpl(
     const TimeDelta& delay,
     GetUpdatesCallerInfo::GetUpdatesSource source,
     const ModelTypePayloadMap& types_with_payloads,
     bool is_canary_job, const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   SVLOG_LOC(nudge_location, 2)
       << "In ScheduleNudgeImpl with delay "
       << delay.InMilliseconds() << " ms, "
       << "source " << GetUpdatesSourceString(source) << ", "
       << "payloads "
       << syncable::ModelTypePayloadMapToString(types_with_payloads)
       << (is_canary_job ? " (canary)" : "");
 
   SyncSourceInfo info(source, types_with_payloads);
 
   SyncSession* session(CreateSyncSession(info));
   SyncSessionJob job(SyncSessionJob::NUDGE, TimeTicks::Now() + delay,
                      make_linked_ptr(session), is_canary_job,
-                     nudge_location);
+                     base::Closure(), nudge_location);
 
   session = NULL;
   if (!ShouldRunJob(job))
     return;
 
   if (pending_nudge_.get()) {
     if (IsBackingOff() && delay > TimeDelta::FromSeconds(1)) {
       SVLOG(2) << "Dropping the nudge because we are in backoff";
       return;
     }
 
     SVLOG(2) << "Coalescing pending nudge";
     pending_nudge_->session->Coalesce(*(job.session.get()));
 
     if (!IsBackingOff()) {
       SVLOG(2) << "Dropping a nudge because"
                << " we are not in backoff and the job was coalesced";
       return;
     } else {
       SVLOG(2) << "Rescheduling pending nudge";
       SyncSession* s = pending_nudge_->session.get();
       job.session.reset(new SyncSession(s->context(), s->delegate(),
           s->source(), s->routing_info(), s->workers()));
       pending_nudge_.reset();
     }
   }
 
   // TODO(lipalani) - pass the job itself to ScheduleSyncSessionJob.
   ScheduleSyncSessionJob(delay, SyncSessionJob::NUDGE, job.session.release(),
-      nudge_location);
+                         base::Closure(), nudge_location);
 }
 
 // Helper to extract the routing info and workers corresponding to types in
 // |types| from |registrar|.
 void GetModelSafeParamsForTypes(const ModelTypeBitSet& types,
     ModelSafeWorkerRegistrar* registrar, ModelSafeRoutingInfo* routes,
     std::vector<ModelSafeWorker*>* workers) {
   ModelSafeRoutingInfo r_tmp;
   std::vector<ModelSafeWorker*> w_tmp;
   registrar->GetModelSafeRoutingInfo(&r_tmp);
@@ skipping 56 matching lines @@
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   SVLOG(2) << "In ScheduleConfigImpl";
   // TODO(tim): config-specific GetUpdatesCallerInfo value?
   SyncSession* session = new SyncSession(session_context_.get(), this,
       SyncSourceInfo(source,
           syncable::ModelTypePayloadMapFromRoutingInfo(
               routing_info, std::string())),
       routing_info, workers);
   ScheduleSyncSessionJob(TimeDelta::FromSeconds(0),
-    SyncSessionJob::CONFIGURATION, session, FROM_HERE);
+                         SyncSessionJob::CONFIGURATION,
+                         session, base::Closure(), FROM_HERE);
 }
 
 const char* SyncScheduler::GetModeString(SyncScheduler::Mode mode) {
   switch (mode) {
     ENUM_CASE(CONFIGURATION_MODE);
     ENUM_CASE(NORMAL_MODE);
   }
   return "";
 }
 
@@ skipping 19 matching lines @@
     const char* name, Task* task, int64 delay_ms) {
   SVLOG_LOC(from_here, 3) << "Posting " << name << " task with "
                           << delay_ms << " ms delay";
   sync_loop_->PostDelayedTask(from_here, task, delay_ms);
 }
 
 void SyncScheduler::ScheduleSyncSessionJob(
     const base::TimeDelta& delay,
     SyncSessionJob::SyncSessionJobPurpose purpose,
     sessions::SyncSession* session,
+    const base::Closure& on_success,
     const tracked_objects::Location& from_here) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SVLOG_LOC(from_here, 2)
       << "In ScheduleSyncSessionJob with "
       << SyncSessionJob::GetPurposeString(purpose)
       << " job and " << delay.InMilliseconds() << " ms delay";
 
   SyncSessionJob job(purpose, TimeTicks::Now() + delay,
-                     make_linked_ptr(session), false, from_here);
+                     make_linked_ptr(session), false, on_success, from_here);
   if (purpose == SyncSessionJob::NUDGE) {
     SVLOG_LOC(from_here, 2) << "Resetting pending_nudge";
     DCHECK(!pending_nudge_.get() || pending_nudge_->session.get() == session);
     pending_nudge_.reset(new SyncSessionJob(job));
   }
   PostDelayedTask(from_here, "DoSyncSessionJob",
                   method_factory_.NewRunnableMethod(
                       &SyncScheduler::DoSyncSessionJob, job),
                   delay.InMilliseconds());
 }
@@ skipping 113 matching lines @@
   if (IsSyncingCurrentlySilenced()) {
     SVLOG(2) << "We are currently throttled; not scheduling the next sync.";
     // TODO(sync): Investigate whether we need to check job.purpose
     // here; see DCHECKs in SaveJob().  (See http://crbug.com/90868.)
     SaveJob(job);
     return;  // Nothing to do.
   }
 
   SVLOG(2) << "Updating the next polling time after SyncMain";
   ScheduleNextSync(job);
+  if (!job.on_success.is_null())
+    job.on_success.Run();
 }
 
 void SyncScheduler::ScheduleNextSync(const SyncSessionJob& old_job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(!old_job.session->HasMoreToSync());
   // Note: |num_server_changes_remaining| > 0 here implies that we received a
   // broken response while trying to download all updates, because the Syncer
   // will loop until this value is exhausted. Also, if unsynced_handles exist
   // but HasMoreToSync is false, this implies that the Syncer determined no
   // forward progress was possible at this time (an error, such as an HTTP
@@ skipping 108 matching lines @@
            << length.InMilliseconds();
 
   // This will reset the had_nudge variable as well.
   wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
                                         length));
   if (old_job.purpose == SyncSessionJob::CONFIGURATION) {
     SyncSession* old = old_job.session.get();
     SyncSession* s(new SyncSession(session_context_.get(), this,
         old->source(), old->routing_info(), old->workers()));
     SyncSessionJob job(old_job.purpose, TimeTicks::Now() + length,
-                        make_linked_ptr(s), false, FROM_HERE);
+                       make_linked_ptr(s), false, base::Closure(),
+                       FROM_HERE);
     wait_interval_->pending_configure_job.reset(new SyncSessionJob(job));
   } else {
     // We are not in configuration mode. So wait_interval's pending job
     // should be null.
     DCHECK(wait_interval_->pending_configure_job.get() == NULL);
 
     // TODO(lipalani) - handle clear user data.
     InitOrCoalescePendingJob(old_job);
   }
   RestartWaiting();
@@ skipping 75 matching lines @@
     copy.is_canary_job = is_canary_job;
     DoSyncSessionJob(copy);
   }
 }
 
 SyncSession* SyncScheduler::CreateSyncSession(const SyncSourceInfo& source) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ModelSafeRoutingInfo routes;
   std::vector<ModelSafeWorker*> workers;
   session_context_->registrar()->GetModelSafeRoutingInfo(&routes);
+  VLOG(2) << "Creating sync session with routes "
+          << ModelSafeRoutingInfoToString(routes);
   session_context_->registrar()->GetWorkers(&workers);
   SyncSourceInfo info(source);
 
   SyncSession* session(new SyncSession(session_context_.get(), this, info,
       routes, workers));
 
   return session;
 }
 
 void SyncScheduler::PollTimerCallback() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ModelSafeRoutingInfo r;
   ModelTypePayloadMap types_with_payloads =
       syncable::ModelTypePayloadMapFromRoutingInfo(r, std::string());
   SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, types_with_payloads);
   SyncSession* s = CreateSyncSession(info);
   ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), SyncSessionJob::POLL, s,
-      FROM_HERE);
+                         base::Closure(), FROM_HERE);
 }
 
 void SyncScheduler::Unthrottle() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
   SVLOG(2) << "Unthrottled.";
   DoCanaryJob();
   wait_interval_.reset();
 }
 
@@ skipping 68 matching lines @@
 
 #undef SVLOG_LOC
 
 #undef SVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // browser_sync