Make sync's change processors sequence-affine.

components/sync/driver/shared_change_processor.cc

 // Copyright 2014 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 "components/sync/driver/shared_change_processor.h"
 
 #include <utility>
 
+#include "base/threading/sequenced_task_runner_handle.h"
 #include "base/threading/thread_task_runner_handle.h"

[gab, 2016/12/20 20:11:29]

rm?

[fdoray, 2016/12/20 20:34:39]

Still needed to initialize |frontend_task_runner_|.

 #include "components/sync/base/data_type_histogram.h"
 #include "components/sync/driver/generic_change_processor.h"
 #include "components/sync/driver/generic_change_processor_factory.h"
 #include "components/sync/driver/shared_change_processor_ref.h"
 #include "components/sync/driver/sync_client.h"
 #include "components/sync/model/sync_change.h"
 #include "components/sync/model/syncable_service.h"
 
 using base::AutoLock;
 
 namespace syncer {
 
 class AttachmentService;
 
 SharedChangeProcessor::SharedChangeProcessor(ModelType type)
     : disconnected_(false),
       type_(type),
       frontend_task_runner_(base::ThreadTaskRunnerHandle::Get()),
       generic_change_processor_(nullptr) {
   DCHECK_NE(type_, UNSPECIFIED);
 }
 
 SharedChangeProcessor::~SharedChangeProcessor() {
-  // We can either be deleted when the DTC is destroyed (on UI
-  // thread), or when the SyncableService stops syncing (datatype
-  // thread).  |generic_change_processor_|, if non-null, must be
-  // deleted on |backend_loop_|.
+  // We can either be deleted when the DTC is destroyed (on UI thread), or when
+  // the SyncableService stops syncing (on |backend_task_runner_|).
+  // |generic_change_processor_|, if non-null, must be deleted on
+  // |backend_task_runner_|.
   if (backend_task_runner_.get()) {
-    if (backend_task_runner_->BelongsToCurrentThread()) {
+    if (backend_task_runner_->RunsTasksOnCurrentThread()) {
       delete generic_change_processor_;
     } else {
       DCHECK(frontend_task_runner_->BelongsToCurrentThread());
       if (!backend_task_runner_->DeleteSoon(FROM_HERE,
                                             generic_change_processor_)) {
         NOTREACHED();
       }
     }
   } else {
     DCHECK(!generic_change_processor_);
@@ skipping 89 matching lines @@
 }
 
 base::WeakPtr<SyncableService> SharedChangeProcessor::Connect(
     SyncClient* sync_client,
     GenericChangeProcessorFactory* processor_factory,
     UserShare* user_share,
     std::unique_ptr<DataTypeErrorHandler> error_handler,
     const base::WeakPtr<SyncMergeResult>& merge_result) {
   DCHECK(sync_client);
   DCHECK(error_handler);
-  backend_task_runner_ = base::ThreadTaskRunnerHandle::Get();
+  backend_task_runner_ = base::SequencedTaskRunnerHandle::Get();
   AutoLock lock(monitor_lock_);
   if (disconnected_)
     return base::WeakPtr<SyncableService>();
   error_handler_ = std::move(error_handler);
   base::WeakPtr<SyncableService> local_service =
       sync_client->GetSyncableServiceForType(type_);
   if (!local_service.get()) {
     LOG(WARNING) << "SyncableService destroyed before DTC was stopped.";
     disconnected_ = true;
     return base::WeakPtr<SyncableService>();
@@ skipping 22 matching lines @@
   error_handler_.reset();
   return was_connected;
 }
 
 ChangeProcessor* SharedChangeProcessor::generic_change_processor() {
   return generic_change_processor_;
 }
 
 int SharedChangeProcessor::GetSyncCount() {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
   AutoLock lock(monitor_lock_);
   if (disconnected_) {
     LOG(ERROR) << "Change processor disconnected.";
     return 0;
   }
   return generic_change_processor_->GetSyncCount();
 }
 
 SyncError SharedChangeProcessor::ProcessSyncChanges(
     const tracked_objects::Location& from_here,
     const SyncChangeList& list_of_changes) {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
   AutoLock lock(monitor_lock_);
   if (disconnected_) {
     // The DTC that disconnects us must ensure it posts a StopSyncing task.
     // If we reach this, it means it just hasn't executed yet.
     SyncError error(FROM_HERE, SyncError::DATATYPE_ERROR,
                     "Change processor disconnected.", type_);
     return error;
   }
   return generic_change_processor_->ProcessSyncChanges(from_here,
                                                        list_of_changes);
 }
 
 SyncDataList SharedChangeProcessor::GetAllSyncData(ModelType type) const {
   SyncDataList data;
   GetAllSyncDataReturnError(type, &data);  // Handles the disconnect case.
   return data;
 }
 
 SyncError SharedChangeProcessor::GetAllSyncDataReturnError(
     ModelType type,
     SyncDataList* data) const {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
   AutoLock lock(monitor_lock_);
   if (disconnected_) {
     SyncError error(FROM_HERE, SyncError::DATATYPE_ERROR,
                     "Change processor disconnected.", type_);
     return error;
   }
   return generic_change_processor_->GetAllSyncDataReturnError(data);
 }
 
 SyncError SharedChangeProcessor::UpdateDataTypeContext(
     ModelType type,
     SyncChangeProcessor::ContextRefreshStatus refresh_status,
     const std::string& context) {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
   AutoLock lock(monitor_lock_);
   if (disconnected_) {
     SyncError error(FROM_HERE, SyncError::DATATYPE_ERROR,
                     "Change processor disconnected.", type_);
     return error;
   }
   return generic_change_processor_->UpdateDataTypeContext(type, refresh_status,
                                                           context);
 }
 
 void SharedChangeProcessor::AddLocalChangeObserver(
     LocalChangeObserver* observer) {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
 
   generic_change_processor_->AddLocalChangeObserver(observer);
 }
 
 void SharedChangeProcessor::RemoveLocalChangeObserver(
     LocalChangeObserver* observer) {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
 
   generic_change_processor_->RemoveLocalChangeObserver(observer);
 }
 
 bool SharedChangeProcessor::SyncModelHasUserCreatedNodes(bool* has_nodes) {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
   AutoLock lock(monitor_lock_);
   if (disconnected_) {
     LOG(ERROR) << "Change processor disconnected.";
     return false;
   }
   return generic_change_processor_->SyncModelHasUserCreatedNodes(has_nodes);
 }
 
 bool SharedChangeProcessor::CryptoReadyIfNecessary() {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
   AutoLock lock(monitor_lock_);
   if (disconnected_) {
     LOG(ERROR) << "Change processor disconnected.";
     return true;  // Otherwise we get into infinite spin waiting.
   }
   return generic_change_processor_->CryptoReadyIfNecessary();
 }
 
 bool SharedChangeProcessor::GetDataTypeContext(std::string* context) const {
   DCHECK(backend_task_runner_.get());
-  DCHECK(backend_task_runner_->BelongsToCurrentThread());
+  DCHECK(backend_task_runner_->RunsTasksOnCurrentThread());
   AutoLock lock(monitor_lock_);
   if (disconnected_) {
     LOG(ERROR) << "Change processor disconnected.";
     return false;
   }
   return generic_change_processor_->GetDataTypeContext(context);
 }
 
 SyncError SharedChangeProcessor::CreateAndUploadError(
     const tracked_objects::Location& location,
@@ skipping 13 matching lines @@
 #undef PER_DATA_TYPE_MACRO
 }
 
 void SharedChangeProcessor::StopLocalService() {
   if (local_service_.get())
     local_service_->StopSyncing(type_);
   local_service_.reset();
 }
 
 }  // namespace syncer