[Sync] Refactor sync configuration logic.

sync/engine/sync_scheduler.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.h"
 
 #include <algorithm>
 #include <cstring>
 
 #include "base/bind.h"
@@ skipping 49 matching lines @@
       return false;
   }
 }
 
 bool IsActionableError(
     const browser_sync::SyncProtocolError& error) {
   return (error.action != browser_sync::UNKNOWN_ACTION);
 }
 }  // namespace
 
+ConfigureParams::ConfigureParams()
+    : source(GetUpdatesCallerInfo::UNKNOWN),
+      need_encryption_key(false) {}
+ConfigureParams::ConfigureParams(
+    const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource& source,
+    const syncable::ModelTypeSet& types_to_config,
+    const browser_sync::ModelSafeRoutingInfo& routing_info,
+    bool need_encryption_key,
+    const base::Closure& ready_task)
+    : source(source),
+      types_to_config(types_to_config),
+      routing_info(routing_info),
+      need_encryption_key(need_encryption_key),
+      ready_task(ready_task) {
+  DCHECK(!ready_task.is_null());
+}
+ConfigureParams::~ConfigureParams() {}
+
 SyncScheduler::DelayProvider::DelayProvider() {}
 SyncScheduler::DelayProvider::~DelayProvider() {}
 
 SyncScheduler::WaitInterval::WaitInterval()
     : mode(UNKNOWN),
       had_nudge(false) {
 }
 
 SyncScheduler::WaitInterval::~WaitInterval() {}
 
@@ skipping 12 matching lines @@
 SyncScheduler::SyncSessionJob::SyncSessionJob()
     : purpose(UNKNOWN),
       is_canary_job(false) {
 }
 
 SyncScheduler::SyncSessionJob::~SyncSessionJob() {}
 
 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) {
+    ConfigureParams config_params, const tracked_objects::Location& from_here)
+    : purpose(purpose),
+      scheduled_start(start),
+      session(session),
+      is_canary_job(is_canary_job),
+      config_params(config_params),
+      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 122 matching lines @@
 
 void SyncScheduler::UpdateServerConnectionManagerStatus(
     HttpResponse::ServerConnectionCode code) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "New server connection code: "
             << HttpResponse::GetServerConnectionCodeString(code);
 
   connection_code_ = code;
 }
 
-void SyncScheduler::Start(Mode mode, const base::Closure& callback) {
+void SyncScheduler::Start(Mode mode) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   std::string thread_name = 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(NULL);  // Will kick start poll timer if needed.
-  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);
   }
 }
 
 void SyncScheduler::SendInitialSnapshot() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   scoped_ptr<SyncSession> dummy(new SyncSession(session_context_, this,
       SyncSourceInfo(), ModelSafeRoutingInfo(),
       std::vector<ModelSafeWorker*>()));
   SyncEngineEvent event(SyncEngineEvent::STATUS_CHANGED);
   event.snapshot = dummy->TakeSnapshot();
   session_context_->NotifyListeners(event);
 }
 
+namespace {
+
+// Helper to extract the routing info and workers corresponding to types in
+// |types| from |current_routes| and |current_workers|.
+void BuildModelSafeParams(
+    const ModelTypeSet& types_to_config,
+    const ModelSafeRoutingInfo& current_routes,
+    const std::vector<ModelSafeWorker*>& current_workers,
+    ModelSafeRoutingInfo* result_routes,
+    std::vector<ModelSafeWorker*>* result_workers) {
+  std::set<ModelSafeGroup> active_groups;
+  active_groups.insert(GROUP_PASSIVE);
+  for (ModelTypeSet::Iterator iter = types_to_config.First(); iter.Good();
+       iter.Inc()) {
+    syncable::ModelType type = iter.Get();
+    ModelSafeRoutingInfo::const_iterator route = current_routes.find(type);
+    DCHECK(route != current_routes.end());
+    ModelSafeGroup group = route->second;
+    (*result_routes)[type] = group;
+    active_groups.insert(group);
+  }
+
+  for(std::vector<ModelSafeWorker*>::const_iterator iter =
+          current_workers.begin(); iter != current_workers.end(); ++iter) {
+    if (active_groups.count((*iter)->GetModelSafeGroup()) > 0)
+      result_workers->push_back(*iter);
+  }
+}
+
+}  // namespace.
+
+bool SyncScheduler::ScheduleConfiguration(const ConfigureParams& params) {
+  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
+  DCHECK_EQ(CONFIGURATION_MODE, mode_);
+  DCHECK(!params.ready_task.is_null());
+  SDVLOG(2) << "Reconfiguring syncer.";
+
+  // Only one configuration is allowed at a time. Verify we're not waiting
+  // for a pending configure job.
+  DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job.get());
+
+  // TODO(sync): now that ModelChanging commands only use those workers within
+  // the routing info, we don't really need |restricted_workers|. Remove it.
+  browser_sync::ModelSafeRoutingInfo restricted_routes;
+  std::vector<ModelSafeWorker*> restricted_workers;
+  BuildModelSafeParams(params.types_to_config,
+                       params.routing_info,
+                       session_context_->workers(),
+                       &restricted_routes,
+                       &restricted_workers);
+  session_context_->set_routing_info(params.routing_info);
+
+  // TODO(sync): if it's confirmed that Cleanup has no effect on non-configures,
+  // remove this command and add a call to PurgeEntriesWithTypeIn here. This

[tim (not reviewing), 2012/06/14 00:26:53]

nit - well, we don't want to poke the directory directly here

[Nicolas Zea, 2012/06/14 22:35:52]

Removed comment. Note that this is being remove anyways as part of
http://codereview.chromium.org/10541079/.

+  // will also allow us to get rid of previous_session_routing_info.
+  SyncSessionJob cleanup_job(
+      SyncSessionJob::CLEANUP_DISABLED_TYPES,
+      TimeTicks::Now(),
+      make_linked_ptr(CreateSyncSession(SyncSourceInfo())),
+      false,
+      ConfigureParams(),
+      FROM_HERE);
+  DoSyncSessionJob(cleanup_job);

[tim (not reviewing), 2012/06/14 00:26:53]

Hmm. So, this might choose not to actually execute the job, but we don't do any
error checking and continue through.  That seems bad.  Of course, I'm pretty
sure we're relying on the implicit fact that cleanup_disabled_types will just
always run :)  A comment explaining would be nice.

[Nicolas Zea, 2012/06/14 22:35:52]

Done, and mentioned bug number for getting rid of this.

+
+  if (params.need_encryption_key) {
+    // TODO(zea): implement in such a way that we can handle failures and the
+    // subsequent retrys the scheduler might perform.
+    NOTIMPLEMENTED();

[tim (not reviewing), 2012/06/14 00:26:53]

bug?

[Nicolas Zea, 2012/06/14 22:35:52]

Done.

+  }
+
+  // Only reconfigure if we have types to config.

[tim (not reviewing), 2012/06/14 00:26:53]

It seems weird that this whole flow is called config, but this comment finds
itself half way through.  Should |types_to_config| be called
|types_to_fetch/download|?

[Nicolas Zea, 2012/06/14 22:35:52]

Done.

+  if (!params.types_to_config.Empty()) {
+    DCHECK(!restricted_routes.empty());
+    // TODO(tim): config-specific GetUpdatesCallerInfo value?

[tim (not reviewing), 2012/06/14 00:26:53]

We have a config specific GetUpdatesCallerInfo... 

[Nicolas Zea, 2012/06/14 22:35:52]

Woops, I removed this is the other patch, but it came back here. Done.

+    linked_ptr<SyncSession> session(new SyncSession(
+        session_context_,
+        this,
+        SyncSourceInfo(params.source,
+                       syncable::ModelTypePayloadMapFromRoutingInfo(
+                           restricted_routes,
+                           std::string())),
+        restricted_routes,
+        restricted_workers));
+    SyncSessionJob job(SyncSessionJob::CONFIGURATION,
+                       TimeTicks::Now(),
+                       session,
+                       false,
+                       params,
+                       FROM_HERE);
+    DoSyncSessionJob(job);
+
+    // If we failed, the job would have been saved as the pending configure
+    // job and a wait interval would have been set.
+    if (!session->Succeeded()) {
+      DCHECK(wait_interval_.get() &&
+             wait_interval_->pending_configure_job.get());
+      return false;
+    }
+  } else {
+    SDVLOG(2) << "No change in routing info, calling ready task directly.";
+    params.ready_task.Run();
+  }
+
+  return true;
+}
+
 SyncScheduler::JobProcessDecision SyncScheduler::DecideWhileInWaitInterval(
     const SyncSessionJob& job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(wait_interval_.get());
   DCHECK_NE(job.purpose, SyncSessionJob::CLEAR_USER_DATA);
   DCHECK_NE(job.purpose, SyncSessionJob::CLEANUP_DISABLED_TYPES);
 
   SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
             << WaitInterval::GetModeString(wait_interval_->mode)
             << (wait_interval_->had_nudge ? " (had nudge)" : "")
@@ skipping 78 matching lines @@
 void SyncScheduler::InitOrCoalescePendingJob(const SyncSessionJob& job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(job.purpose != SyncSessionJob::CONFIGURATION);
   if (pending_nudge_.get() == NULL) {
     SDVLOG(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,
+        ConfigureParams(), job.from_here);
     pending_nudge_.reset(new SyncSessionJob(new_job));
 
     return;
   }
 
   SDVLOG(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 25 matching lines @@
   // TODO(sync): Should we also check that job.purpose !=
   // CLEANUP_DISABLED_TYPES?  (See http://crbug.com/90868.)
   if (job.purpose == SyncSessionJob::NUDGE) {
     SDVLOG(2) << "Saving a nudge job";
     InitOrCoalescePendingJob(job);
   } else if (job.purpose == SyncSessionJob::CONFIGURATION){
     SDVLOG(2) << "Saving a configuration job";
     DCHECK(wait_interval_.get());
     DCHECK(mode_ == CONFIGURATION_MODE);
 
+    // Config params should always get set.
+    DCHECK(!job.config_params.ready_task.is_null());
     SyncSession* old = job.session.get();
     SyncSession* s(new SyncSession(session_context_, this, old->source(),
                                    old->routing_info(), old->workers()));
-    SyncSessionJob new_job(job.purpose, TimeTicks::Now(),
-                           make_linked_ptr(s), false, job.from_here);
+    SyncSessionJob new_job(job.purpose,
+                           TimeTicks::Now(),
+                           make_linked_ptr(s),
+                           false,
+                           job.config_params,
+                           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::ClearUserData() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SyncSessionJob job(SyncSessionJob::CLEAR_USER_DATA, TimeTicks::Now(),
+  SyncSessionJob job(SyncSessionJob::CLEAR_USER_DATA,
+                     TimeTicks::Now(),
                      make_linked_ptr(CreateSyncSession(SyncSourceInfo())),
                      false,
+                     ConfigureParams(),
                      FROM_HERE);
 
   DoSyncSessionJob(job);
 }
 
-void SyncScheduler::CleanupDisabledTypes() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SyncSessionJob job(SyncSessionJob::CLEANUP_DISABLED_TYPES, TimeTicks::Now(),
-                     make_linked_ptr(CreateSyncSession(SyncSourceInfo())),
-                     false,
-                     FROM_HERE);
-  DoSyncSessionJob(job);
-}
-
 void SyncScheduler::ScheduleNudgeAsync(
     const TimeDelta& delay,
     NudgeSource source, ModelTypeSet types,
     const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG_LOC(nudge_location, 2)
       << "Nudge scheduled with delay " << delay.InMilliseconds() << " ms, "
       << "source " << GetNudgeSourceString(source) << ", "
       << "types " << ModelTypeSetToString(types);
 
@@ skipping 37 matching lines @@
       << "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);
+                     ConfigureParams(), nudge_location);
 
   session = NULL;
   if (!ShouldRunJob(job))
     return;
 
   if (pending_nudge_.get()) {
     if (IsBackingOff() && delay > TimeDelta::FromSeconds(1)) {
       SDVLOG(2) << "Dropping the nudge because we are in backoff";
       return;
     }
@@ skipping 10 matching lines @@
     job.scheduled_start = std::min(job.scheduled_start,
                                    pending_nudge_->scheduled_start);
     pending_nudge_.reset();
   }
 
   // TODO(zea): Consider adding separate throttling/backoff for datatype
   // refresh requests.
   ScheduleSyncSessionJob(job);
 }
 
-// Helper to extract the routing info and workers corresponding to types in
-// |types| from |current_routes| and |current_workers|.
-void GetModelSafeParamsForTypes(ModelTypeSet types,
-    const ModelSafeRoutingInfo& current_routes,
-    const std::vector<ModelSafeWorker*>& current_workers,
-    ModelSafeRoutingInfo* result_routes,
-    std::vector<ModelSafeWorker*>* result_workers) {
-  bool passive_group_added = false;
-
-  typedef std::vector<ModelSafeWorker*>::const_iterator iter;
-  for (ModelTypeSet::Iterator it = types.First();
-       it.Good(); it.Inc()) {
-    const syncable::ModelType t = it.Get();
-    ModelSafeRoutingInfo::const_iterator route = current_routes.find(t);
-    DCHECK(route != current_routes.end());
-    ModelSafeGroup group = route->second;
-
-    (*result_routes)[t] = group;
-    iter w_tmp_it = std::find_if(current_workers.begin(), current_workers.end(),
-                                 ModelSafeWorkerGroupIs(group));
-    if (w_tmp_it != current_workers.end()) {
-      iter result_workers_it = std::find_if(
-          result_workers->begin(), result_workers->end(),
-          ModelSafeWorkerGroupIs(group));
-      if (result_workers_it == result_workers->end())
-        result_workers->push_back(*w_tmp_it);
-
-      if (group == GROUP_PASSIVE)
-        passive_group_added = true;
-    } else {
-        NOTREACHED();
-    }
-  }
-
-  // Always add group passive.
-  if (passive_group_added == false) {
-    iter it = std::find_if(current_workers.begin(), current_workers.end(),
-                           ModelSafeWorkerGroupIs(GROUP_PASSIVE));
-    if (it != current_workers.end())
-      result_workers->push_back(*it);
-    else
-      NOTREACHED();
-  }
-}
-
-void SyncScheduler::ScheduleConfiguration(
-    ModelTypeSet types,
-    GetUpdatesCallerInfo::GetUpdatesSource source) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  DCHECK(IsConfigRelatedUpdateSourceValue(source));
-  SDVLOG(2) << "Scheduling a config";
-
-  ModelSafeRoutingInfo routes;
-  std::vector<ModelSafeWorker*> workers;
-  GetModelSafeParamsForTypes(types,
-                             session_context_->routing_info(),
-                             session_context_->workers(),
-                             &routes, &workers);
-
-  SyncSession* session = new SyncSession(session_context_, this,
-      SyncSourceInfo(source,
-                     syncable::ModelTypePayloadMapFromRoutingInfo(
-                         routes, std::string())),
-      routes, workers);
-  SyncSessionJob job(SyncSessionJob::CONFIGURATION, TimeTicks::Now(),
-                     make_linked_ptr(session),
-                     false,
-                     FROM_HERE);
-  DoSyncSessionJob(job);
-}
-
 const char* SyncScheduler::GetModeString(SyncScheduler::Mode mode) {
   switch (mode) {
     ENUM_CASE(CONFIGURATION_MODE);
     ENUM_CASE(NORMAL_MODE);
   }
   return "";
 }
 
 const char* SyncScheduler::GetDecisionString(
     SyncScheduler::JobProcessDecision mode) {
@@ skipping 199 matching lines @@
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(!old_job.session->HasMoreToSync());
 
   AdjustPolling(&old_job);
 
   if (old_job.session->Succeeded()) {
     // Only reset backoff if we actually reached the server.
     if (old_job.session->SuccessfullyReachedServer())
       wait_interval_.reset();
     SDVLOG(2) << "Job succeeded so not scheduling more jobs";
+
+    // If this was a configuration job with a ready task, invoke it now that
+    // we finished successfully.

[tim (not reviewing), 2012/06/14 00:26:53]

Why do this in 'ScheduleNextSync'?  It's not really part of the contract... I
don't think there's a reason we have to do this after AdjustPolling, unless I'm
really missing something?  Can we do it before calling ScheduleNextSync instead?

[Nicolas Zea, 2012/06/14 22:35:52]

Primarily just because this is where we were already checking Succeeded(). But,
there's not reason we can't check it in more than one place. Done.

+    if (!old_job.config_params.ready_task.is_null())
+      old_job.config_params.ready_task.Run();
     return;
   }
 
   if (old_job.purpose == SyncSessionJob::POLL) {
     return; // We don't retry POLL jobs.
   }
 
   // TODO(rlarocque): There's no reason why we should blindly backoff and retry
   // if we don't succeed.  Some types of errors are not likely to disappear on
   // their own.  With the return values now available in the old_job.session, we
@@ skipping 65 matching lines @@
 
   SDVLOG(2) << "In handle continuation error with "
             << SyncSessionJob::GetPurposeString(old_job.purpose)
             << " job. The time delta(ms) is "
             << 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) {
+    SDVLOG(2) << "Configuration did not succeed, scheduling retry.";
+    // Config params should always get set.
+    DCHECK(!old_job.config_params.ready_task.is_null());
     SyncSession* old = old_job.session.get();
     SyncSession* s(new SyncSession(session_context_, 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, old_job.config_params,
+                       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 101 matching lines @@
   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);
 
   SyncSessionJob job(SyncSessionJob::POLL, TimeTicks::Now(),
                      make_linked_ptr(s),
                      false,
+                     ConfigureParams(),
                      FROM_HERE);
 
   ScheduleSyncSessionJob(job);
 }
 
 void SyncScheduler::Unthrottle() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
   SDVLOG(2) << "Unthrottled.";
   DoCanaryJob();
@@ skipping 82 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // browser_sync