[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
 
+SyncScheduler::ConfigureParams::ConfigureParams()
+    : source(GetUpdatesCallerInfo::UNKNOWN),
+      need_cleanup(false),
+      need_encryption_key(false) {}
+SyncScheduler::ConfigureParams::ConfigureParams(
+    const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource& source,
+    const syncable::ModelTypeSet& types_to_config,
+    const browser_sync::ModelSafeRoutingInfo& routing_info,
+    bool need_cleanup,
+    bool need_encryption_key,
+    const base::Closure& ready_task)
+    : source(source),
+      types_to_config(types_to_config),
+      routing_info(routing_info),
+      need_cleanup(need_cleanup),
+      need_encryption_key(need_encryption_key),
+      ready_task(ready_task) {}
+SyncScheduler::ConfigureParams::~ConfigureParams() {}
+
 SyncScheduler::DelayProvider::DelayProvider() {}
 SyncScheduler::DelayProvider::~DelayProvider() {}
 
 SyncScheduler::WaitInterval::WaitInterval()
     : mode(UNKNOWN),
       had_nudge(false) {
 }
 
 SyncScheduler::WaitInterval::~WaitInterval() {}
 
@@ skipping 159 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;
-    PostTask(FROM_HERE, "SendInitialSnapshot",
-             base::Bind(&SyncScheduler::SendInitialSnapshot,
-                        weak_ptr_factory_.GetWeakPtr()));
+    SendInitialSnapshot();
   }
-  PostTask(FROM_HERE, "StartImpl",
-           base::Bind(&SyncScheduler::StartImpl,
-                      weak_ptr_factory_.GetWeakPtr(), mode, callback));
+  StartImpl(mode);
 }
 
 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);
 }
 
-void SyncScheduler::StartImpl(Mode mode, const base::Closure& callback) {
+void SyncScheduler::StartImpl(Mode mode) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "In StartImpl with mode " << GetModeString(mode);
 
   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 (!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);
   }
 }
 
+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::Configure(const ConfigureParams& params) {
+  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
+  DCHECK_EQ(CONFIGURATION_MODE, mode_);
+  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());
+
+  // We set the routing info for all enabled types in the session context, but
+  // the session for this configuration only knows about the routing info for
+  // those types_to_config (set via restricted_routes and restricted_workers).
+  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);
+  const browser_sync::ModelSafeRoutingInfo& previous_routing_info =
+      session_context_->routing_info();
+  // TODO(zea): ensure that the routing info can't be set anywhere but here.
+  session_context_->set_routing_info(params.routing_info);
+
+  if (params.need_cleanup) {

[rlarocque, 2012/06/04 23:01:38]

CleanupDisabledTypesCommand has some strange logic built around the
SyncSessionContext's previous_session_routing_info member.  Because you're
custom-building a sync session for this job anyway, maybe you could eliminate
the need for that logic?

In an ideal world, CleanupDisabledTypes would be a function that takes a
'ModelTypeSet types_to_clean' parameter...

[tim (not reviewing), 2012/06/05 04:02:09]

We do have a function that takes a 'types to clean' parameter --
PurgeEntriesWithTypeIn on Directory.  CleanupDisabledTypes is built around that
to figure out what should be cleaned up and when. I think ideally, cleanup would
be performed by one or as few pieces of code as possible.  Splitting up the
logic across more components makes it harder to write targeted unit tests.  

I'm wondering, do we actually have to go through the extra trouble here? What if
we just schedule a cleanup and let the command impl figure out if it needs to
run or not?

[Nicolas Zea, 2012/06/07 19:25:42]

Discussed offline. I've removed this logic and now just trigger cleanup with
every configuration cycle (the logic in the command itself prevents us from
doing extra work).

+    SyncSessionJob job(SyncSessionJob::CLEANUP_DISABLED_TYPES, TimeTicks::Now(),
+                       make_linked_ptr(CreateSyncSession(SyncSourceInfo())),
+                       false,
+                       FROM_HERE);
+    DoSyncSessionJob(job);
+  }
+
+  if (params.need_encryption_key) {
+    // TODO(zea): implement in such a way that we can handle failures and the
+    // subsequent retry's the scheduler might perform.
+    NOTIMPLEMENTED();
+  }
+
+  // If the set of enabled types hasn't changed there's no need to configure.
+  if (params.routing_info != previous_routing_info) {
+    DCHECK(!restricted_routes.empty());
+    // TODO(tim): config-specific GetUpdatesCallerInfo value?
+    SyncSession* session = new SyncSession(
+        session_context_,
+        this,
+        SyncSourceInfo(params.source,
+                       syncable::ModelTypePayloadMapFromRoutingInfo(
+                           restricted_routes,
+                           std::string())),
+        restricted_routes,
+        restricted_workers);
+    SyncSessionJob job(SyncSessionJob::CONFIGURATION,

[rlarocque, 2012/06/04 23:01:38]

The arguments for converting SyncSessionJobs into real functions keep piling
up...

+                       TimeTicks::Now(),
+                       make_linked_ptr(session),
+                       false,
+                       FROM_HERE);
+    job.config_params = params;
+    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 (wait_interval_.get() && wait_interval_->pending_configure_job.get())
+      return false;
+  } else {
+    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 124 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);
     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;
 };
 
+// TODO(sync): Remove the *Impl methods for the other Schedule*
+// functions, too.
 void SyncScheduler::ScheduleClearUserData() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   PostTask(FROM_HERE, "ScheduleClearUserDataImpl",
            base::Bind(&SyncScheduler::ScheduleClearUserDataImpl,
                       weak_ptr_factory_.GetWeakPtr()));
 }
 
-// TODO(sync): Remove the *Impl methods for the other Schedule*
-// functions, too.
-void SyncScheduler::ScheduleCleanupDisabledTypes() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SyncSessionJob job(SyncSessionJob::CLEANUP_DISABLED_TYPES, TimeTicks::Now(),
-                     make_linked_ptr(CreateSyncSession(SyncSourceInfo())),
-                     false,
-                     FROM_HERE);
-  ScheduleSyncSessionJob(job);
-}
-
 void SyncScheduler::ScheduleNudge(
     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 84 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::ScheduleConfig(
-    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);
-
-  PostTask(FROM_HERE, "ScheduleConfigImpl",
-           base::Bind(&SyncScheduler::ScheduleConfigImpl,
-                      weak_ptr_factory_.GetWeakPtr(),
-                      routes,
-                      workers,
-                      source));
-}
-
-void SyncScheduler::ScheduleConfigImpl(
-    const ModelSafeRoutingInfo& routing_info,
-    const std::vector<ModelSafeWorker*>& workers,
-    const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-
-  SDVLOG(2) << "In ScheduleConfigImpl";
-  // TODO(tim): config-specific GetUpdatesCallerInfo value?
-  SyncSession* session = new SyncSession(session_context_, this,
-      SyncSourceInfo(source,
-          syncable::ModelTypePayloadMapFromRoutingInfo(
-              routing_info, std::string())),
-      routing_info, workers);
-  SyncSessionJob job(SyncSessionJob::CONFIGURATION, TimeTicks::Now(),
-                     make_linked_ptr(session),
-                     false,
-                     FROM_HERE);
-  ScheduleSyncSessionJob(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 201 matching lines @@
 
   if (old_job.session->Succeeded()) {
     // Success implies backoff relief.  Note that if this was a
     // "one-off" job (i.e. purpose ==
     // SyncSessionJob::{CLEAR_USER_DATA,CLEANUP_DISABLED_TYPES}), if
     // there was work to do before it ran this wont have changed, as
     // jobs like this don't run a full sync cycle.  So we don't need
     // special code here.
     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.
+    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) {
+    // 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);
     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);
@@ skipping 208 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // browser_sync