src/sync/sessions/model_type_registry.cc

   1 // Copyright 2014 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/sessions/model_type_registry.h"
   6
   7 #include "base/bind.h"
   8 #include "base/observer_list.h"
   9 #include "base/thread_task_runner_handle.h"
  10 #include "sync/engine/directory_commit_contributor.h"
  11 #include "sync/engine/directory_update_handler.h"
  12 #include "sync/engine/model_type_sync_proxy.h"
  13 #include "sync/engine/model_type_sync_proxy_impl.h"
  14 #include "sync/engine/model_type_sync_worker.h"
  15 #include "sync/engine/model_type_sync_worker_impl.h"
  16 #include "sync/internal_api/public/non_blocking_sync_common.h"
  17 #include "sync/sessions/directory_type_debug_info_emitter.h"
  18 #include "sync/util/cryptographer.h"
  19
  20 namespace syncer {
  21
  22 namespace {
  23
  24 class ModelTypeSyncProxyWrapper : public ModelTypeSyncProxy {
  25  public:
  26   ModelTypeSyncProxyWrapper(
  27       const base::WeakPtr<ModelTypeSyncProxyImpl>& proxy,
  28       const scoped_refptr<base::SequencedTaskRunner>& processor_task_runner);
  29   virtual ~ModelTypeSyncProxyWrapper();
  30
  31   virtual void OnCommitCompleted(
  32       const DataTypeState& type_state,
  33       const CommitResponseDataList& response_list) OVERRIDE;
  34   virtual void OnUpdateReceived(
  35       const DataTypeState& type_state,
  36       const UpdateResponseDataList& response_list,
  37       const UpdateResponseDataList& pending_updates) OVERRIDE;
  38
  39  private:
  40   base::WeakPtr<ModelTypeSyncProxyImpl> processor_;
  41   scoped_refptr<base::SequencedTaskRunner> processor_task_runner_;
  42 };
  43
  44 ModelTypeSyncProxyWrapper::ModelTypeSyncProxyWrapper(
  45     const base::WeakPtr<ModelTypeSyncProxyImpl>& proxy,
  46     const scoped_refptr<base::SequencedTaskRunner>& processor_task_runner)
  47     : processor_(proxy), processor_task_runner_(processor_task_runner) {
  48 }
  49
  50 ModelTypeSyncProxyWrapper::~ModelTypeSyncProxyWrapper() {
  51 }
  52
  53 void ModelTypeSyncProxyWrapper::OnCommitCompleted(
  54     const DataTypeState& type_state,
  55     const CommitResponseDataList& response_list) {
  56   processor_task_runner_->PostTask(
  57       FROM_HERE,
  58       base::Bind(&ModelTypeSyncProxyImpl::OnCommitCompleted,
  59                  processor_,
  60                  type_state,
  61                  response_list));
  62 }
  63
  64 void ModelTypeSyncProxyWrapper::OnUpdateReceived(
  65     const DataTypeState& type_state,
  66     const UpdateResponseDataList& response_list,
  67     const UpdateResponseDataList& pending_updates) {
  68   processor_task_runner_->PostTask(
  69       FROM_HERE,
  70       base::Bind(&ModelTypeSyncProxyImpl::OnUpdateReceived,
  71                  processor_,
  72                  type_state,
  73                  response_list,
  74                  pending_updates));
  75 }
  76
  77 class ModelTypeSyncWorkerWrapper : public ModelTypeSyncWorker {
  78  public:
  79   ModelTypeSyncWorkerWrapper(
  80       const base::WeakPtr<ModelTypeSyncWorkerImpl>& worker,
  81       const scoped_refptr<base::SequencedTaskRunner>& sync_thread);
  82   virtual ~ModelTypeSyncWorkerWrapper();
  83
  84   virtual void EnqueueForCommit(const CommitRequestDataList& list) OVERRIDE;
  85
  86  private:
  87   base::WeakPtr<ModelTypeSyncWorkerImpl> worker_;
  88   scoped_refptr<base::SequencedTaskRunner> sync_thread_;
  89 };
  90
  91 ModelTypeSyncWorkerWrapper::ModelTypeSyncWorkerWrapper(
  92     const base::WeakPtr<ModelTypeSyncWorkerImpl>& worker,
  93     const scoped_refptr<base::SequencedTaskRunner>& sync_thread)
  94     : worker_(worker), sync_thread_(sync_thread) {
  95 }
  96
  97 ModelTypeSyncWorkerWrapper::~ModelTypeSyncWorkerWrapper() {
  98 }
  99
 100 void ModelTypeSyncWorkerWrapper::EnqueueForCommit(
 101     const CommitRequestDataList& list) {
 102   sync_thread_->PostTask(
 103       FROM_HERE,
 104       base::Bind(&ModelTypeSyncWorkerImpl::EnqueueForCommit, worker_, list));
 105 }
 106
 107 }  // namespace
 108
 109 ModelTypeRegistry::ModelTypeRegistry(
 110     const std::vector<scoped_refptr<ModelSafeWorker> >& workers,
 111     syncable::Directory* directory,
 112     NudgeHandler* nudge_handler)
 113     : directory_(directory),
 114       cryptographer_provider_(directory_),
 115       nudge_handler_(nudge_handler),
 116       weak_ptr_factory_(this) {
 117   for (size_t i = 0u; i < workers.size(); ++i) {
 118     workers_map_.insert(
 119         std::make_pair(workers[i]->GetModelSafeGroup(), workers[i]));
 120   }
 121 }
 122
 123 ModelTypeRegistry::~ModelTypeRegistry() {}
 124
 125 void ModelTypeRegistry::SetEnabledDirectoryTypes(
 126     const ModelSafeRoutingInfo& routing_info) {
 127   // Remove all existing directory processors and delete them.  The
 128   // DebugInfoEmitters are not deleted here, since we want to preserve their
 129   // counters.
 130   for (ModelTypeSet::Iterator it = enabled_directory_types_.First();
 131        it.Good(); it.Inc()) {
 132     size_t result1 = update_handler_map_.erase(it.Get());
 133     size_t result2 = commit_contributor_map_.erase(it.Get());
 134     DCHECK_EQ(1U, result1);
 135     DCHECK_EQ(1U, result2);
 136   }
 137
 138   // Clear the old instances of directory update handlers and commit
 139   // contributors, deleting their contents in the processs.
 140   directory_update_handlers_.clear();
 141   directory_commit_contributors_.clear();
 142
 143   // Create new ones and add them to the appropriate containers.
 144   for (ModelSafeRoutingInfo::const_iterator routing_iter = routing_info.begin();
 145        routing_iter != routing_info.end(); ++routing_iter) {
 146     ModelType type = routing_iter->first;
 147     ModelSafeGroup group = routing_iter->second;
 148     std::map<ModelSafeGroup, scoped_refptr<ModelSafeWorker> >::iterator
 149         worker_it = workers_map_.find(group);
 150     DCHECK(worker_it != workers_map_.end());
 151     scoped_refptr<ModelSafeWorker> worker = worker_it->second;
 152
 153     // DebugInfoEmitters are never deleted.  Use existing one if we have it.
 154     DirectoryTypeDebugInfoEmitter* emitter = NULL;
 155     DirectoryTypeDebugInfoEmitterMap::iterator it =
 156         directory_type_debug_info_emitter_map_.find(type);
 157     if (it != directory_type_debug_info_emitter_map_.end()) {
 158       emitter = it->second;
 159     } else {
 160       emitter = new DirectoryTypeDebugInfoEmitter(directory_, type,
 161                                                   &type_debug_info_observers_);
 162       directory_type_debug_info_emitter_map_.insert(
 163           std::make_pair(type, emitter));
 164       directory_type_debug_info_emitters_.push_back(emitter);
 165     }
 166
 167     DirectoryCommitContributor* committer =
 168         new DirectoryCommitContributor(directory_, type, emitter);
 169     DirectoryUpdateHandler* updater =
 170         new DirectoryUpdateHandler(directory_, type, worker, emitter);
 171
 172     // These containers take ownership of their contents.
 173     directory_commit_contributors_.push_back(committer);
 174     directory_update_handlers_.push_back(updater);
 175
 176     bool inserted1 =
 177         update_handler_map_.insert(std::make_pair(type, updater)).second;
 178     DCHECK(inserted1) << "Attempt to override existing type handler in map";
 179
 180     bool inserted2 =
 181         commit_contributor_map_.insert(std::make_pair(type, committer)).second;
 182     DCHECK(inserted2) << "Attempt to override existing type handler in map";
 183   }
 184
 185   enabled_directory_types_ = GetRoutingInfoTypes(routing_info);
 186   DCHECK(Intersection(GetEnabledDirectoryTypes(),
 187                       GetEnabledNonBlockingTypes()).Empty());
 188 }
 189
 190 void ModelTypeRegistry::ConnectSyncTypeToWorker(
 191     ModelType type,
 192     const DataTypeState& data_type_state,
 193     const UpdateResponseDataList& saved_pending_updates,
 194     const scoped_refptr<base::SequencedTaskRunner>& type_task_runner,
 195     const base::WeakPtr<ModelTypeSyncProxyImpl>& proxy_impl) {
 196   DVLOG(1) << "Enabling an off-thread sync type: " << ModelTypeToString(type);
 197
 198   // Initialize Worker -> Proxy communication channel.
 199   scoped_ptr<ModelTypeSyncProxy> proxy(
 200       new ModelTypeSyncProxyWrapper(proxy_impl, type_task_runner));
 201   scoped_ptr<ModelTypeSyncWorkerImpl> worker(
 202       new ModelTypeSyncWorkerImpl(type,
 203                                   data_type_state,
 204                                   saved_pending_updates,
 205                                   &cryptographer_provider_,
 206                                   nudge_handler_,
 207                                   proxy.Pass()));
 208
 209   // Initialize Proxy -> Worker communication channel.
 210   scoped_ptr<ModelTypeSyncWorker> wrapped_worker(
 211       new ModelTypeSyncWorkerWrapper(worker->AsWeakPtr(),
 212                                      scoped_refptr<base::SequencedTaskRunner>(
 213                                          base::ThreadTaskRunnerHandle::Get())));
 214   type_task_runner->PostTask(FROM_HERE,
 215                              base::Bind(&ModelTypeSyncProxyImpl::OnConnect,
 216                                         proxy_impl,
 217                                         base::Passed(&wrapped_worker)));
 218
 219   DCHECK(update_handler_map_.find(type) == update_handler_map_.end());
 220   DCHECK(commit_contributor_map_.find(type) == commit_contributor_map_.end());
 221
 222   update_handler_map_.insert(std::make_pair(type, worker.get()));
 223   commit_contributor_map_.insert(std::make_pair(type, worker.get()));
 224
 225   // The container takes ownership.
 226   model_type_sync_workers_.push_back(worker.release());
 227
 228   DCHECK(Intersection(GetEnabledDirectoryTypes(),
 229                       GetEnabledNonBlockingTypes()).Empty());
 230 }
 231
 232 void ModelTypeRegistry::DisconnectSyncWorker(ModelType type) {
 233   DVLOG(1) << "Disabling an off-thread sync type: " << ModelTypeToString(type);
 234   DCHECK(update_handler_map_.find(type) != update_handler_map_.end());
 235   DCHECK(commit_contributor_map_.find(type) != commit_contributor_map_.end());
 236
 237   size_t updaters_erased = update_handler_map_.erase(type);
 238   size_t committers_erased = commit_contributor_map_.erase(type);
 239
 240   DCHECK_EQ(1U, updaters_erased);
 241   DCHECK_EQ(1U, committers_erased);
 242
 243   // Remove from the ScopedVector, deleting the worker in the process.
 244   for (ScopedVector<ModelTypeSyncWorkerImpl>::iterator it =
 245            model_type_sync_workers_.begin();
 246        it != model_type_sync_workers_.end();
 247        ++it) {
 248     if ((*it)->GetModelType() == type) {
 249       model_type_sync_workers_.erase(it);
 250       break;
 251     }
 252   }
 253 }
 254
 255 ModelTypeSet ModelTypeRegistry::GetEnabledTypes() const {
 256   return Union(GetEnabledDirectoryTypes(), GetEnabledNonBlockingTypes());
 257 }
 258
 259 UpdateHandlerMap* ModelTypeRegistry::update_handler_map() {
 260   return &update_handler_map_;
 261 }
 262
 263 CommitContributorMap* ModelTypeRegistry::commit_contributor_map() {
 264   return &commit_contributor_map_;
 265 }
 266
 267 DirectoryTypeDebugInfoEmitterMap*
 268 ModelTypeRegistry::directory_type_debug_info_emitter_map() {
 269   return &directory_type_debug_info_emitter_map_;
 270 }
 271
 272 void ModelTypeRegistry::RegisterDirectoryTypeDebugInfoObserver(
 273     syncer::TypeDebugInfoObserver* observer) {
 274   if (!type_debug_info_observers_.HasObserver(observer))
 275     type_debug_info_observers_.AddObserver(observer);
 276 }
 277
 278 void ModelTypeRegistry::UnregisterDirectoryTypeDebugInfoObserver(
 279     syncer::TypeDebugInfoObserver* observer) {
 280   type_debug_info_observers_.RemoveObserver(observer);
 281 }
 282
 283 bool ModelTypeRegistry::HasDirectoryTypeDebugInfoObserver(
 284     syncer::TypeDebugInfoObserver* observer) {
 285   return type_debug_info_observers_.HasObserver(observer);
 286 }
 287
 288 void ModelTypeRegistry::RequestEmitDebugInfo() {
 289   for (DirectoryTypeDebugInfoEmitterMap::iterator it =
 290        directory_type_debug_info_emitter_map_.begin();
 291        it != directory_type_debug_info_emitter_map_.end(); ++it) {
 292     it->second->EmitCommitCountersUpdate();
 293     it->second->EmitUpdateCountersUpdate();
 294     it->second->EmitStatusCountersUpdate();
 295   }
 296 }
 297
 298 base::WeakPtr<SyncContext> ModelTypeRegistry::AsWeakPtr() {
 299   return weak_ptr_factory_.GetWeakPtr();
 300 }
 301
 302 ModelTypeSet ModelTypeRegistry::GetEnabledDirectoryTypes() const {
 303   return enabled_directory_types_;
 304 }
 305
 306 ModelTypeSet ModelTypeRegistry::GetEnabledNonBlockingTypes() const {
 307   ModelTypeSet enabled_off_thread_types;
 308   for (ScopedVector<ModelTypeSyncWorkerImpl>::const_iterator it =
 309            model_type_sync_workers_.begin();
 310        it != model_type_sync_workers_.end();
 311        ++it) {
 312     enabled_off_thread_types.Put((*it)->GetModelType());
 313   }
 314   return enabled_off_thread_types;
 315 }
 316
 317 }  // namespace syncer