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"
#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 (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_->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_);
  }
}

void SharedChangeProcessor::StartAssociation(
    StartDoneCallback start_done,
    SyncClient* const sync_client,
    GenericChangeProcessorFactory* processor_factory,
    UserShare* user_share,
    std::unique_ptr<DataTypeErrorHandler> error_handler) {
  DCHECK(user_share);
  SyncMergeResult local_merge_result(type_);
  SyncMergeResult syncer_merge_result(type_);
  base::WeakPtrFactory<SyncMergeResult> weak_ptr_factory(&syncer_merge_result);

  // Connect |shared_change_processor| to the syncer and get the
  // SyncableService associated with type_.
  // Note that it's possible the shared_change_processor has already been
  // disconnected at this point, so all our accesses to the syncer from this
  // point on are through it.
  local_service_ =
      Connect(sync_client, processor_factory, user_share,
              std::move(error_handler), weak_ptr_factory.GetWeakPtr());
  if (!local_service_.get()) {
    SyncError error(FROM_HERE, SyncError::DATATYPE_ERROR,
                    "Failed to connect to syncer.", type_);
    local_merge_result.set_error(error);
    start_done.Run(DataTypeController::ASSOCIATION_FAILED, local_merge_result,
                   syncer_merge_result);
    return;
  }

  if (!CryptoReadyIfNecessary()) {
    SyncError error(FROM_HERE, SyncError::CRYPTO_ERROR, "", type_);
    local_merge_result.set_error(error);
    start_done.Run(DataTypeController::NEEDS_CRYPTO, local_merge_result,
                   syncer_merge_result);
    return;
  }

  bool sync_has_nodes = false;
  if (!SyncModelHasUserCreatedNodes(&sync_has_nodes)) {
    SyncError error(FROM_HERE, SyncError::UNRECOVERABLE_ERROR,
                    "Failed to load sync nodes", type_);
    local_merge_result.set_error(error);
    start_done.Run(DataTypeController::UNRECOVERABLE_ERROR, local_merge_result,
                   syncer_merge_result);
    return;
  }

  // Scope for |initial_sync_data| which might be expensive, so we don't want
  // to keep it in memory longer than necessary.
  {
    SyncDataList initial_sync_data;

    base::TimeTicks start_time = base::TimeTicks::Now();
    SyncError error = GetAllSyncDataReturnError(type_, &initial_sync_data);
    if (error.IsSet()) {
      local_merge_result.set_error(error);
      start_done.Run(DataTypeController::ASSOCIATION_FAILED, local_merge_result,
                     syncer_merge_result);
      return;
    }

    std::string datatype_context;
    if (GetDataTypeContext(&datatype_context)) {
      local_service_->UpdateDataTypeContext(
          type_, SyncChangeProcessor::NO_REFRESH, datatype_context);
    }

    syncer_merge_result.set_num_items_before_association(
        initial_sync_data.size());
    // Passes a reference to |shared_change_processor|.
    local_merge_result = local_service_->MergeDataAndStartSyncing(
        type_, initial_sync_data, std::unique_ptr<SyncChangeProcessor>(
                                      new SharedChangeProcessorRef(this)),
        std::unique_ptr<SyncErrorFactory>(new SharedChangeProcessorRef(this)));
    RecordAssociationTime(base::TimeTicks::Now() - start_time);
    if (local_merge_result.error().IsSet()) {
      start_done.Run(DataTypeController::ASSOCIATION_FAILED, local_merge_result,
                     syncer_merge_result);
      return;
    }
  }

  syncer_merge_result.set_num_items_after_association(GetSyncCount());

  start_done.Run(!sync_has_nodes ? DataTypeController::OK_FIRST_RUN
                                 : DataTypeController::OK,
                 local_merge_result, syncer_merge_result);
}

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::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>();
  }

  generic_change_processor_ = processor_factory
                                  ->CreateGenericChangeProcessor(
                                      type_, user_share, error_handler_->Copy(),
                                      local_service, merge_result, sync_client)
                                  .release();
  // If available, propagate attachment service to the syncable service.
  std::unique_ptr<AttachmentService> attachment_service =
      generic_change_processor_->GetAttachmentService();
  if (attachment_service) {
    local_service->SetAttachmentService(std::move(attachment_service));
  }
  return local_service;
}

bool SharedChangeProcessor::Disconnect() {
  // May be called from any thread.
  DVLOG(1) << "Disconnecting change processor.";
  AutoLock lock(monitor_lock_);
  bool was_connected = !disconnected_;
  disconnected_ = true;
  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_->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_->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_->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_->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_->RunsTasksOnCurrentThread());

  generic_change_processor_->AddLocalChangeObserver(observer);
}

void SharedChangeProcessor::RemoveLocalChangeObserver(
    LocalChangeObserver* observer) {
  DCHECK(backend_task_runner_.get());
  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_->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_->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_->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,
    const std::string& message) {
  AutoLock lock(monitor_lock_);
  if (!disconnected_) {
    return error_handler_->CreateAndUploadError(location, message, type_);
  } else {
    return SyncError(location, SyncError::DATATYPE_ERROR, message, type_);
  }
}

void SharedChangeProcessor::RecordAssociationTime(base::TimeDelta time) {
#define PER_DATA_TYPE_MACRO(type_str) \
  UMA_HISTOGRAM_TIMES("Sync." type_str "AssociationTime", time);
  SYNC_DATA_TYPE_HISTOGRAM(type_);
#undef PER_DATA_TYPE_MACRO
}

void SharedChangeProcessor::StopLocalService() {
  if (local_service_.get())
    local_service_->StopSyncing(type_);
  local_service_.reset();
}

}  // namespace syncer