sync: Add NonBlockingSyncTypeProcessorCore (again)

This is a re-land of r258035, which was mistakenly reverted with
r258266.  Original CL description follows.

Adds the class for handling sync updates and commits on the sync thread.
The current implementation is mostly a bunch of stubs.

In the future, this will be hooked up to a NonBlockingSyncTypeProcessor
that lives on the model type's thread.  These two classes will
collaborate to implement the non-blocking sync system.

This CL also adds some code to the ModelTypeRegistry and
SyncSessionContext to prepare for managing these
NonBlockingSyncTypeProcessor objects.  This new code is used only in
tests.

TBR=zea@chromium.org
BUG=351005

Review URL: https://codereview.chromium.org/206693002

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@258390 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 170 insertions, 24 deletions

diff --git a/sync/sessions/model_type_registry.cc b/sync/sessions/model_type_registry.cc
index 7035755..ab93c68 100644
--- a/sync/sessions/model_type_registry.cc
+++ b/sync/sessions/model_type_registry.cc
@@ -6,15 +6,14 @@
 
 #include "sync/engine/directory_commit_contributor.h"
 #include "sync/engine/directory_update_handler.h"
+#include "sync/engine/non_blocking_type_processor_core.h"
 
 namespace syncer {
 
 ModelTypeRegistry::ModelTypeRegistry(
     const std::vector<scoped_refptr<ModelSafeWorker> >& workers,
     syncable::Directory* directory)
-    : update_handler_deleter_(&update_handler_map_),
-      commit_contributor_deleter_(&commit_contributor_map_),
-      directory_(directory) {
+    : directory_(directory) {
   for (size_t i = 0u; i < workers.size(); ++i) {
     workers_map_.insert(
         std::make_pair(workers[i]->GetModelSafeGroup(), workers[i]));
@@ -25,11 +24,21 @@ ModelTypeRegistry::~ModelTypeRegistry() {}
 
 void ModelTypeRegistry::SetEnabledDirectoryTypes(
     const ModelSafeRoutingInfo& routing_info) {
-  STLDeleteValues(&update_handler_map_);
-  STLDeleteValues(&commit_contributor_map_);
-  update_handler_map_.clear();
-  commit_contributor_map_.clear();
+  // Remove all existing directory processors and delete them.
+  for (ModelTypeSet::Iterator it = enabled_directory_types_.First();
+       it.Good(); it.Inc()) {
+    size_t result1 = update_handler_map_.erase(it.Get());
+    size_t result2 = commit_contributor_map_.erase(it.Get());
+    DCHECK_EQ(1U, result1);
+    DCHECK_EQ(1U, result2);
+  }
+
+  // Clear the old instances of directory update handlers and commit
+  // contributors, deleting their contents in the processs.
+  directory_update_handlers_.clear();
+  directory_commit_contributors_.clear();
 
+  // Create new ones and add them to the appropriate containers.
   for (ModelSafeRoutingInfo::const_iterator routing_iter = routing_info.begin();
        routing_iter != routing_info.end(); ++routing_iter) {
     ModelType type = routing_iter->first;
@@ -44,6 +53,10 @@ void ModelTypeRegistry::SetEnabledDirectoryTypes(
     DirectoryUpdateHandler* updater =
         new DirectoryUpdateHandler(directory_, type, worker);
 
+    // These containers take ownership of their contents.
+    directory_commit_contributors_.push_back(committer);
+    directory_update_handlers_.push_back(updater);
+
     bool inserted1 =
         update_handler_map_.insert(std::make_pair(type, updater)).second;
     DCHECK(inserted1) << "Attempt to override existing type handler in map";
@@ -51,10 +64,53 @@ void ModelTypeRegistry::SetEnabledDirectoryTypes(
     bool inserted2 =
         commit_contributor_map_.insert(std::make_pair(type, committer)).second;
     DCHECK(inserted2) << "Attempt to override existing type handler in map";
+  }
+
+  enabled_directory_types_ = GetRoutingInfoTypes(routing_info);
+}
+
+void ModelTypeRegistry::InitializeNonBlockingType(syncer::ModelType type) {
+  DVLOG(1) << "Enabling an off-thread sync type: " << ModelTypeToString(type);
+
+  scoped_ptr<NonBlockingTypeProcessorCore> core(
+      new NonBlockingTypeProcessorCore(type));
+
+  DCHECK(update_handler_map_.find(type) == update_handler_map_.end());
+  DCHECK(commit_contributor_map_.find(type) == commit_contributor_map_.end());
+
+  update_handler_map_.insert(std::make_pair(type, core.get()));
+  commit_contributor_map_.insert(std::make_pair(type, core.get()));
+
+  // The container takes ownership.
+  non_blocking_type_processor_cores_.push_back(core.release());
+}
+
+void ModelTypeRegistry::RemoveNonBlockingType(ModelType type) {
+  DVLOG(1) << "Disabling an off-thread sync type: " << ModelTypeToString(type);
+  DCHECK(update_handler_map_.find(type) != update_handler_map_.end());
+  DCHECK(commit_contributor_map_.find(type) != commit_contributor_map_.end());
+
+  size_t updaters_erased = update_handler_map_.erase(type);
+  size_t committers_erased = commit_contributor_map_.erase(type);
 
+  DCHECK_EQ(1U, updaters_erased);
+  DCHECK_EQ(1U, committers_erased);
+
+  // Remove from the ScopedVector, deleting the core in the process.
+  for (ScopedVector<NonBlockingTypeProcessorCore>::iterator it =
+       non_blocking_type_processor_cores_.begin();
+       it != non_blocking_type_processor_cores_.end(); ++it) {
+    if ((*it)->GetModelType() == type) {
+      non_blocking_type_processor_cores_.erase(it);
+      break;
+    }
   }
 }
 
+ModelTypeSet ModelTypeRegistry::GetEnabledTypes() const {
+  return Union(GetEnabledDirectoryTypes(), GetEnabledNonBlockingTypes());
+}
+
 UpdateHandlerMap* ModelTypeRegistry::update_handler_map() {
   return &update_handler_map_;
 }
@@ -63,4 +119,18 @@ CommitContributorMap* ModelTypeRegistry::commit_contributor_map() {
   return &commit_contributor_map_;
 }
 
+ModelTypeSet ModelTypeRegistry::GetEnabledDirectoryTypes() const {
+  return enabled_directory_types_;
+}
+
+ModelTypeSet ModelTypeRegistry::GetEnabledNonBlockingTypes() const {
+  ModelTypeSet enabled_off_thread_types;
+  for (ScopedVector<NonBlockingTypeProcessorCore>::const_iterator it =
+           non_blocking_type_processor_cores_.begin();
+       it != non_blocking_type_processor_cores_.end(); ++it) {
+    enabled_off_thread_types.Put((*it)->GetModelType());
+  }
+  return enabled_off_thread_types;
+}
+
 }  // namespace syncer
diff --git a/sync/sessions/model_type_registry.h b/sync/sessions/model_type_registry.h
index ad34ada..65445d4 100644
--- a/sync/sessions/model_type_registry.h
+++ b/sync/sessions/model_type_registry.h
@@ -9,7 +9,7 @@
 
 #include "base/basictypes.h"
 #include "base/memory/scoped_ptr.h"
-#include "base/stl_util.h"
+#include "base/memory/scoped_vector.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/engine/model_safe_worker.h"
@@ -20,8 +20,11 @@ namespace syncable {
 class Directory;
 }  // namespace syncable
 
-class UpdateHandler;
 class CommitContributor;
+class DirectoryCommitContributor;
+class DirectoryUpdateHandler;
+class NonBlockingTypeProcessorCore;
+class UpdateHandler;
 
 typedef std::map<ModelType, UpdateHandler*> UpdateHandlerMap;
 typedef std::map<ModelType, CommitContributor*> CommitContributorMap;
@@ -37,20 +40,37 @@ class SYNC_EXPORT_PRIVATE ModelTypeRegistry {
   // Sets the set of enabled types.
   void SetEnabledDirectoryTypes(const ModelSafeRoutingInfo& routing_info);
 
+  // Enables an off-thread type for syncing.
+  //
+  // Expects that the type is not currently enabled.
+  void InitializeNonBlockingType(syncer::ModelType type);
+
+  // Disables the syncing of an off-thread type.
+  //
+  // Expects that the type is currently enabled.
+  // Deletes the processor core associated with the type.
+  void RemoveNonBlockingType(syncer::ModelType type);
+
+  // Gets the set of enabled types.
+  ModelTypeSet GetEnabledTypes() const;
+
   // Simple getters.
   UpdateHandlerMap* update_handler_map();
   CommitContributorMap* commit_contributor_map();
 
  private:
-  // Classes to manage the types hooked up to receive and commit sync data.
-  UpdateHandlerMap update_handler_map_;
-  CommitContributorMap commit_contributor_map_;
+  ModelTypeSet GetEnabledNonBlockingTypes() const;
+  ModelTypeSet GetEnabledDirectoryTypes() const;
 
-  // Deleter for the |update_handler_map_|.
-  STLValueDeleter<UpdateHandlerMap> update_handler_deleter_;
+  // Sets of handlers and contributors.
+  ScopedVector<DirectoryCommitContributor> directory_commit_contributors_;
+  ScopedVector<DirectoryUpdateHandler> directory_update_handlers_;
+  ScopedVector<NonBlockingTypeProcessorCore> non_blocking_type_processor_cores_;
 
-  // Deleter for the |commit_contributor_map_|.
-  STLValueDeleter<CommitContributorMap> commit_contributor_deleter_;
+  // Maps of UpdateHandlers and CommitContributors.
+  // They do not own any of the objects they point to.
+  UpdateHandlerMap update_handler_map_;
+  CommitContributorMap commit_contributor_map_;
 
   // The known ModelSafeWorkers.
   std::map<ModelSafeGroup, scoped_refptr<ModelSafeWorker> > workers_map_;
@@ -58,6 +78,9 @@ class SYNC_EXPORT_PRIVATE ModelTypeRegistry {
   // The directory.  Not owned.
   syncable::Directory* directory_;
 
+  // The set of enabled directory types.
+  ModelTypeSet enabled_directory_types_;
+
   DISALLOW_COPY_AND_ASSIGN(ModelTypeRegistry);
 };
 
diff --git a/sync/sessions/model_type_registry_unittest.cc b/sync/sessions/model_type_registry_unittest.cc
index a063758..7f174e5 100644
--- a/sync/sessions/model_type_registry_unittest.cc
+++ b/sync/sessions/model_type_registry_unittest.cc
@@ -5,6 +5,7 @@
 #include <vector>
 
 #include "base/message_loop/message_loop.h"
+#include "sync/engine/non_blocking_type_processor_core.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/sessions/model_type_registry.h"
 #include "sync/test/engine/fake_model_worker.h"
@@ -111,4 +112,59 @@ TEST_F(ModelTypeRegistryTest, SetEnabledDirectoryTypes_Clear) {
   registry()->SetEnabledDirectoryTypes(routing_info2);
 }
 
+TEST_F(ModelTypeRegistryTest, NonBlockingTypes) {
+  EXPECT_TRUE(registry()->GetEnabledTypes().Empty());
+
+  registry()->InitializeNonBlockingType(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
+      ModelTypeSet(syncer::THEMES)));
+
+  registry()->InitializeNonBlockingType(syncer::SESSIONS);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
+      ModelTypeSet(syncer::THEMES, syncer::SESSIONS)));
+
+  registry()->RemoveNonBlockingType(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
+      ModelTypeSet(syncer::SESSIONS)));
+
+  // Allow ModelTypeRegistry destruction to delete the
+  // Sessions' NonBlockingTypeProcessorCore.
+}
+
+TEST_F(ModelTypeRegistryTest, NonBlockingTypesWithDirectoryTypes) {
+  ModelSafeRoutingInfo routing_info1;
+  routing_info1.insert(std::make_pair(NIGORI, GROUP_PASSIVE));
+  routing_info1.insert(std::make_pair(BOOKMARKS, GROUP_UI));
+  routing_info1.insert(std::make_pair(AUTOFILL, GROUP_DB));
+
+  ModelTypeSet current_types;
+  EXPECT_TRUE(registry()->GetEnabledTypes().Empty());
+
+  // Add the themes non-blocking type.
+  registry()->InitializeNonBlockingType(syncer::THEMES);
+  current_types.Put(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Add some directory types.
+  registry()->SetEnabledDirectoryTypes(routing_info1);
+  current_types.PutAll(GetRoutingInfoTypes(routing_info1));
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Add sessions non-blocking type.
+  registry()->InitializeNonBlockingType(syncer::SESSIONS);
+  current_types.Put(syncer::SESSIONS);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Remove themes non-blocking type.
+  registry()->RemoveNonBlockingType(syncer::THEMES);
+  current_types.Remove(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Clear all directory types.
+  ModelSafeRoutingInfo routing_info2;
+  registry()->SetEnabledDirectoryTypes(routing_info2);
+  current_types.RemoveAll(GetRoutingInfoTypes(routing_info1));
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+}
+
 }  // namespace syncer
diff --git a/sync/sessions/sync_session_context.cc b/sync/sessions/sync_session_context.cc
index 36e6343..54ae473 100644
--- a/sync/sessions/sync_session_context.cc
+++ b/sync/sessions/sync_session_context.cc
@@ -42,9 +42,12 @@ SyncSessionContext::SyncSessionContext(
 SyncSessionContext::~SyncSessionContext() {
 }
 
+ModelTypeSet SyncSessionContext::GetEnabledTypes() const {
+  return model_type_registry_->GetEnabledTypes();
+}
+
 void SyncSessionContext::SetRoutingInfo(
     const ModelSafeRoutingInfo& routing_info) {
-  enabled_types_ = GetRoutingInfoTypes(routing_info);
   model_type_registry_->SetEnabledDirectoryTypes(routing_info);
 }
 
diff --git a/sync/sessions/sync_session_context.h b/sync/sessions/sync_session_context.h
index f1afa7d..c7661ce 100644
--- a/sync/sessions/sync_session_context.h
+++ b/sync/sessions/sync_session_context.h
@@ -62,9 +62,7 @@ class SYNC_EXPORT_PRIVATE SyncSessionContext {
     return directory_;
   }
 
-  ModelTypeSet enabled_types() const {
-    return enabled_types_;
-  }
+  ModelTypeSet GetEnabledTypes() const;
 
   void SetRoutingInfo(const ModelSafeRoutingInfo& routing_info);
 
@@ -142,10 +140,6 @@ class SYNC_EXPORT_PRIVATE SyncSessionContext {
   ServerConnectionManager* const connection_manager_;
   syncable::Directory* const directory_;
 
-  // The set of enabled types.  Derrived from the routing info set with
-  // set_routing_info().
-  ModelTypeSet enabled_types_;
-
   // We use this to stuff extensions activity into CommitMessages so the server
   // can correlate commit traffic with extension-related bookmark mutations.
   scoped_refptr<ExtensionsActivity> extensions_activity_;