Make sync's change processors sequence-affine.

components/sync/driver/generic_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/generic_change_processor.h"
 
 #include <stddef.h>
 
 #include <algorithm>
 #include <utility>
 
 #include "base/location.h"
 #include "base/memory/ptr_util.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/utf_string_conversions.h"
-#include "base/threading/thread_task_runner_handle.h"
+#include "base/threading/sequenced_task_runner_handle.h"
 #include "components/sync/base/unrecoverable_error_handler.h"
 #include "components/sync/driver/sync_api_component_factory.h"
 #include "components/sync/driver/sync_client.h"
 #include "components/sync/model/local_change_observer.h"
 #include "components/sync/model/sync_change.h"
 #include "components/sync/model/sync_error.h"
 #include "components/sync/model/syncable_service.h"
 #include "components/sync/syncable/base_node.h"
 #include "components/sync/syncable/change_record.h"
 #include "components/sync/syncable/entry.h"  // TODO(tim): Bug 123674.
@@ skipping 85 matching lines @@
     const base::WeakPtr<SyncMergeResult>& merge_result,
     UserShare* user_share,
     SyncClient* sync_client,
     std::unique_ptr<AttachmentStoreForSync> attachment_store)
     : ChangeProcessor(std::move(error_handler)),
       type_(type),
       local_service_(local_service),
       merge_result_(merge_result),
       share_handle_(user_share),
       weak_ptr_factory_(this) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   DCHECK_NE(type_, UNSPECIFIED);
   if (attachment_store) {
     std::string store_birthday;
     {
       ReadTransaction trans(FROM_HERE, share_handle());
       store_birthday = trans.GetStoreBirthday();
     }
     attachment_service_ =
         sync_client->GetSyncApiComponentFactory()->CreateAttachmentService(
             std::move(attachment_store), *user_share, store_birthday, type,
             this);
     attachment_service_weak_ptr_factory_ =
         base::MakeUnique<base::WeakPtrFactory<AttachmentService>>(
             attachment_service_.get());
     attachment_service_proxy_ = AttachmentServiceProxy(
-        base::ThreadTaskRunnerHandle::Get(),
+        base::SequencedTaskRunnerHandle::Get(),
         attachment_service_weak_ptr_factory_->GetWeakPtr());
     UploadAllAttachmentsNotOnServer();
   } else {
     attachment_service_proxy_ =
-        AttachmentServiceProxy(base::ThreadTaskRunnerHandle::Get(),
+        AttachmentServiceProxy(base::SequencedTaskRunnerHandle::Get(),
                                base::WeakPtr<AttachmentService>());
   }
 }
 
 GenericChangeProcessor::~GenericChangeProcessor() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
 }
 
 void GenericChangeProcessor::ApplyChangesFromSyncModel(
     const BaseTransaction* trans,
     int64_t model_version,
     const ImmutableChangeRecordList& changes) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   DCHECK(syncer_changes_.empty());
   for (ChangeRecordList::const_iterator it = changes.Get().begin();
        it != changes.Get().end(); ++it) {
     if (it->action == ChangeRecord::ACTION_DELETE) {
       std::unique_ptr<sync_pb::EntitySpecifics> specifics;
       if (it->specifics.has_password()) {
         DCHECK(it->extra.get());
         specifics = base::MakeUnique<sync_pb::EntitySpecifics>(it->specifics);
         specifics->mutable_password()
             ->mutable_client_only_encrypted_data()
@@ skipping 20 matching lines @@
         return;
       }
       syncer_changes_.push_back(SyncChange(
           FROM_HERE, action,
           BuildRemoteSyncData(it->id, read_node, attachment_service_proxy_)));
     }
   }
 }
 
 void GenericChangeProcessor::CommitChangesFromSyncModel() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   if (syncer_changes_.empty())
     return;
   if (!local_service_.get()) {
     ModelType type = syncer_changes_[0].sync_data().GetDataType();
     SyncError error(FROM_HERE, SyncError::DATATYPE_ERROR,
                     "Local service destroyed.", type);
     error_handler()->OnUnrecoverableError(error);
     return;
   }
   SyncError error =
@@ skipping 43 matching lines @@
 }
 
 void GenericChangeProcessor::OnAttachmentUploaded(
     const AttachmentId& attachment_id) {
   WriteTransaction trans(FROM_HERE, share_handle());
   trans.UpdateEntriesMarkAttachmentAsOnServer(attachment_id);
 }
 
 SyncError GenericChangeProcessor::GetAllSyncDataReturnError(
     SyncDataList* current_sync_data) const {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   std::string type_name = ModelTypeToString(type_);
   ReadTransaction trans(FROM_HERE, share_handle());
   ReadNode root(&trans);
   if (root.InitTypeRoot(type_) != BaseNode::INIT_OK) {
     SyncError error(FROM_HERE, SyncError::DATATYPE_ERROR,
                     "Server did not create the top-level " + type_name +
                         " node. We might be running against an out-of-"
                         "date server.",
                     type_);
     return error;
@@ skipping 145 matching lines @@
   if (IsActOnceDataType(type))
     node->Drop();
   else
     node->Tombstone();
   return SyncError();
 }
 
 SyncError GenericChangeProcessor::ProcessSyncChanges(
     const tracked_objects::Location& from_here,
     const SyncChangeList& list_of_changes) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
 
   if (list_of_changes.empty()) {
     // No work. Exit without entering WriteTransaction.
     return SyncError();
   }
 
   // Keep track of brand new attachments so we can persist them on this device
   // and upload them to the server.
   AttachmentIdSet new_attachments;
 
@@ skipping 199 matching lines @@
   if (merge_result_.get()) {
     merge_result_->set_num_items_modified(merge_result_->num_items_modified() +
                                           1);
   }
   // TODO(sync): Support updating other parts of the sync node (title,
   // successor, parent, etc.).
   return SyncError();
 }
 
 bool GenericChangeProcessor::SyncModelHasUserCreatedNodes(bool* has_nodes) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   DCHECK(has_nodes);
   std::string type_name = ModelTypeToString(type_);
   std::string err_str =
       "Server did not create the top-level " + type_name +
       " node. We might be running against an out-of-date server.";
   *has_nodes = false;
   ReadTransaction trans(FROM_HERE, share_handle());
   ReadNode type_root_node(&trans);
   if (type_root_node.InitTypeRoot(type_) != BaseNode::INIT_OK) {
     LOG(ERROR) << err_str;
     return false;
   }
 
   // The sync model has user created nodes if the type's root node has any
   // children.
   *has_nodes = type_root_node.HasChildren();
   return true;
 }
 
 bool GenericChangeProcessor::CryptoReadyIfNecessary() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   // We only access the cryptographer while holding a transaction.
   ReadTransaction trans(FROM_HERE, share_handle());
   const ModelTypeSet encrypted_types = trans.GetEncryptedTypes();
   return !encrypted_types.Has(type_) || trans.GetCryptographer()->is_ready();
 }
 
 void GenericChangeProcessor::StartImpl() {}
 
 UserShare* GenericChangeProcessor::share_handle() const {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   return share_handle_;
 }
 
 void GenericChangeProcessor::UploadAllAttachmentsNotOnServer() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   DCHECK(attachment_service_.get());
   AttachmentIdList ids;
   {
     ReadTransaction trans(FROM_HERE, share_handle());
     trans.GetAttachmentIdsToUpload(type_, &ids);
   }
   if (!ids.empty()) {
     attachment_service_->UploadAttachments(ids);
   }
 }
 
 void GenericChangeProcessor::NotifyLocalChangeObservers(
     const syncable::Entry* current_entry,
     const SyncChange& change) {
   for (auto& observer : local_change_observers_)
     observer.OnLocalChange(current_entry, change);
 }
 
 std::unique_ptr<AttachmentService>
 GenericChangeProcessor::GetAttachmentService() const {
   return std::unique_ptr<AttachmentService>(
       new AttachmentServiceProxy(attachment_service_proxy_));
 }
 
 }  // namespace syncer