[SyncFS] Post tasks between SyncEngine and SyncWorker

chrome/browser/sync_file_system/drive_backend/sync_engine.cc

 // Copyright 2013 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 "chrome/browser/sync_file_system/drive_backend/sync_engine.h"
 
 #include <vector>
 
 #include "base/bind.h"
 #include "base/threading/sequenced_worker_pool.h"
@@ skipping 72 matching lines @@
   drive_service->Initialize(signin_manager->GetAuthenticatedAccountId());
 
   scoped_ptr<drive::DriveUploaderInterface> drive_uploader(
       new drive::DriveUploader(drive_service.get(), drive_task_runner.get()));
 
   drive::DriveNotificationManager* notification_manager =
       drive::DriveNotificationManagerFactory::GetForBrowserContext(context);
   ExtensionService* extension_service =
       extensions::ExtensionSystem::Get(context)->extension_service();
 
-  scoped_refptr<base::SequencedTaskRunner> task_runner(
+  scoped_refptr<base::SequencedTaskRunner> file_task_runner(
       worker_pool->GetSequencedTaskRunnerWithShutdownBehavior(
           worker_pool->GetSequenceToken(),
           base::SequencedWorkerPool::SKIP_ON_SHUTDOWN));
 
+  // TODO(peria): Create another task runner to manage SyncWorker.
+  base::SequencedTaskRunner*

[tzik, 2014/04/21 04:09:10]

s/base::SequencedTaskRunner/scoped_refptr<base::SingleThreadTaskRunner>/?
It's for ensuring the TaskRunner is not deleted soon after this statement ends.
Using SingleThreadTaskRunner is just for consistency to other usage.

[peria, 2014/04/21 04:50:06]

Done.

+      worker_task_runner = base::MessageLoopProxy::current();
+
   scoped_ptr<drive_backend::SyncEngine> sync_engine(
       new SyncEngine(drive_service.Pass(),
                      drive_uploader.Pass(),
+                     worker_task_runner,
                      notification_manager,
                      extension_service,
                      signin_manager));
-  sync_engine->Initialize(GetSyncFileSystemDir(context->GetPath()),
-                          task_runner.get(),
-                          NULL);
+  sync_engine->Initialize(
+      GetSyncFileSystemDir(context->GetPath()),
+      file_task_runner.get(),
+      NULL);
 
   return sync_engine.Pass();
 }
 
 void SyncEngine::AppendDependsOnFactories(
     std::set<BrowserContextKeyedServiceFactory*>* factories) {
   DCHECK(factories);
   factories->insert(drive::DriveNotificationManagerFactory::GetInstance());
   factories->insert(SigninManagerFactory::GetInstance());
   factories->insert(
       extensions::ExtensionsBrowserClient::Get()->GetExtensionSystemFactory());
 }
 
 SyncEngine::~SyncEngine() {
   net::NetworkChangeNotifier::RemoveNetworkChangeObserver(this);
   GetDriveService()->RemoveObserver(this);
   if (notification_manager_)
     notification_manager_->RemoveObserver(this);
 }
 
 void SyncEngine::Initialize(const base::FilePath& base_dir,
-                            base::SequencedTaskRunner* task_runner,
+                            base::SequencedTaskRunner* file_task_runner,
                             leveldb::Env* env_override) {
   scoped_ptr<SyncEngineContext> sync_engine_context(
       new SyncEngineContext(drive_service_.get(),
                             drive_uploader_.get(),
-                            task_runner));
-  // TODO(peria): Move this create function to thread pool.
+                            base::MessageLoopProxy::current(),

[tzik, 2014/04/21 04:09:10]

worker_task_runner_?

[peria, 2014/04/21 04:50:06]

UI task runner should be set here, so base::MessageLoopProxy::current() is
right. Isn't it?

[tzik, 2014/04/21 05:10:05]

Ah, right I misread that.

+                            file_task_runner));
+  // TODO(peria): Use PostTask on |worker_task_runner_| to call this function.
   sync_worker_ = SyncWorker::CreateOnWorker(weak_ptr_factory_.GetWeakPtr(),
                                             base_dir,
                                             sync_engine_context.Pass(),
                                             env_override);
 
   if (notification_manager_)
     notification_manager_->AddObserver(this);
   GetDriveService()->AddObserver(this);
   net::NetworkChangeNotifier::AddNetworkChangeObserver(this);
 }
 
 void SyncEngine::AddServiceObserver(SyncServiceObserver* observer) {
   service_observers_.AddObserver(observer);
 }
 
 void SyncEngine::AddFileStatusObserver(FileStatusObserver* observer) {
   file_status_observers_.AddObserver(observer);
 }
 
 void SyncEngine::RegisterOrigin(
     const GURL& origin, const SyncStatusCallback& callback) {
-  sync_worker_->RegisterOrigin(origin, callback);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::RegisterOrigin,
+                 base::Unretained(sync_worker_.get()),
+                 origin, callback));
 }
 
 void SyncEngine::EnableOrigin(
     const GURL& origin, const SyncStatusCallback& callback) {
-  sync_worker_->EnableOrigin(origin, callback);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::EnableOrigin,
+                 base::Unretained(sync_worker_.get()),
+                 origin, callback));
 }
 
 void SyncEngine::DisableOrigin(
     const GURL& origin, const SyncStatusCallback& callback) {
-  sync_worker_->DisableOrigin(origin, callback);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::DisableOrigin,
+                 base::Unretained(sync_worker_.get()),
+                 origin, callback));
 }
 
 void SyncEngine::UninstallOrigin(
     const GURL& origin,
     UninstallFlag flag,
     const SyncStatusCallback& callback) {
-  sync_worker_->UninstallOrigin(origin, flag, callback);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::UninstallOrigin,
+                 base::Unretained(sync_worker_.get()),
+                 origin, flag, callback));
 }
 
 void SyncEngine::ProcessRemoteChange(const SyncFileCallback& callback) {
-  sync_worker_->ProcessRemoteChange(callback);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::ProcessRemoteChange,
+                 base::Unretained(sync_worker_.get()),
+                 callback));
 }
 
-void SyncEngine::SetRemoteChangeProcessor(
-    RemoteChangeProcessor* processor) {
-  sync_worker_->SetRemoteChangeProcessor(processor);
+void SyncEngine::SetRemoteChangeProcessor(RemoteChangeProcessor* processor) {
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::SetRemoteChangeProcessor,
+                 base::Unretained(sync_worker_.get()),
+                 processor));
 }
 
 LocalChangeProcessor* SyncEngine::GetLocalChangeProcessor() {
   return this;
 }
 
 bool SyncEngine::IsConflicting(const fileapi::FileSystemURL& url) {
   // TODO(tzik): Implement this before we support manual conflict resolution.
   return false;
 }
 
 RemoteServiceState SyncEngine::GetCurrentState() const {
+  // TODO(peria): Post task
   return sync_worker_->GetCurrentState();
 }
 
 void SyncEngine::GetOriginStatusMap(OriginStatusMap* status_map) {
+  // TODO(peria): Make this route asynchronous.
   sync_worker_->GetOriginStatusMap(status_map);
 }
 
 scoped_ptr<base::ListValue> SyncEngine::DumpFiles(const GURL& origin) {
+  // TODO(peria): Make this route asynchronous.
   return sync_worker_->DumpFiles(origin);
 }
 
 scoped_ptr<base::ListValue> SyncEngine::DumpDatabase() {
+  // TODO(peria): Make this route asynchronous.
   return sync_worker_->DumpDatabase();
 }
 
 void SyncEngine::SetSyncEnabled(bool enabled) {
-  sync_worker_->SetSyncEnabled(enabled);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::SetSyncEnabled,
+                 base::Unretained(sync_worker_.get()),
+                 enabled));
 }
 
 void SyncEngine::UpdateSyncEnabled(bool enabled) {
   const char* status_message = enabled ? "Sync is enabled" : "Sync is disabled";
   FOR_EACH_OBSERVER(
       Observer, service_observers_,
       OnRemoteServiceStateUpdated(GetCurrentState(), status_message));
 }
 
 SyncStatusCode SyncEngine::SetDefaultConflictResolutionPolicy(
     ConflictResolutionPolicy policy) {
+  // TODO(peria): Make this route asynchronous.
   return sync_worker_->SetDefaultConflictResolutionPolicy(policy);
 }
 
 SyncStatusCode SyncEngine::SetConflictResolutionPolicy(
     const GURL& origin,
     ConflictResolutionPolicy policy) {
+  // TODO(peria): Make this route asynchronous.
   return sync_worker_->SetConflictResolutionPolicy(origin, policy);
 }
 
 ConflictResolutionPolicy SyncEngine::GetDefaultConflictResolutionPolicy()
     const {
+  // TODO(peria): Make this route asynchronous.
   return sync_worker_->GetDefaultConflictResolutionPolicy();
 }
 
 ConflictResolutionPolicy SyncEngine::GetConflictResolutionPolicy(
     const GURL& origin) const {
+  // TODO(peria): Make this route asynchronous.
   return sync_worker_->GetConflictResolutionPolicy(origin);
 }
 
 void SyncEngine::GetRemoteVersions(
     const fileapi::FileSystemURL& url,
     const RemoteVersionsCallback& callback) {
   // TODO(tzik): Implement this before we support manual conflict resolution.
   callback.Run(SYNC_STATUS_FAILED, std::vector<Version>());
 }
 
@@ skipping 15 matching lines @@
         OnRemoteChangeQueueUpdated(metadata_db->CountDirtyTracker()));
   }
 }
 
 void SyncEngine::ApplyLocalChange(
     const FileChange& local_change,
     const base::FilePath& local_path,
     const SyncFileMetadata& local_metadata,
     const fileapi::FileSystemURL& url,
     const SyncStatusCallback& callback) {
-  sync_worker_->ApplyLocalChange(
-      local_change, local_path, local_metadata, url, callback);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::ApplyLocalChange,
+                 base::Unretained(sync_worker_.get()),
+                 local_change,
+                 local_path,
+                 local_metadata,
+                 url,
+                 callback));
 }
 
 SyncTaskManager* SyncEngine::GetSyncTaskManagerForTesting() {
+  // TODO(peria): Post task
   return sync_worker_->GetSyncTaskManager();
 }
 
 void SyncEngine::OnNotificationReceived() {
-  sync_worker_->OnNotificationReceived();
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::OnNotificationReceived,
+                 base::Unretained(sync_worker_.get())));
 }
 
 void SyncEngine::OnPushNotificationEnabled(bool) {}
 
 void SyncEngine::OnReadyToSendRequests() {
-  sync_worker_->OnReadyToSendRequests(
-      signin_manager_ ? signin_manager_->GetAuthenticatedAccountId() : "");
+  const std::string account_id =
+      signin_manager_ ? signin_manager_->GetAuthenticatedAccountId() : "";
+
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::OnReadyToSendRequests,
+                 base::Unretained(sync_worker_.get()),
+                 account_id));
 }
 
 void SyncEngine::OnRefreshTokenInvalid() {
-  sync_worker_->OnRefreshTokenInvalid();
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::OnRefreshTokenInvalid,
+                 base::Unretained(sync_worker_.get())));
 }
 
 void SyncEngine::OnNetworkChanged(
     net::NetworkChangeNotifier::ConnectionType type) {
-  sync_worker_->OnNetworkChanged(type);
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SyncWorker::OnNetworkChanged,
+                 base::Unretained(sync_worker_.get()),
+                 type));
 }
 
 drive::DriveServiceInterface* SyncEngine::GetDriveService() {
-  return sync_worker_->GetDriveService();
+  return drive_service_.get();
 }
 
 drive::DriveUploaderInterface* SyncEngine::GetDriveUploader() {
-  return sync_worker_->GetDriveUploader();
+  return drive_uploader_.get();
 }
 
 MetadataDatabase* SyncEngine::GetMetadataDatabase() {
+  // TODO(peria): Post task
   return sync_worker_->GetMetadataDatabase();
 }
 
 SyncEngine::SyncEngine(
     scoped_ptr<drive::DriveServiceInterface> drive_service,
     scoped_ptr<drive::DriveUploaderInterface> drive_uploader,
+    base::SequencedTaskRunner* worker_task_runner,
     drive::DriveNotificationManager* notification_manager,
     ExtensionServiceInterface* extension_service,
     SigninManagerBase* signin_manager)
     : drive_service_(drive_service.Pass()),
       drive_uploader_(drive_uploader.Pass()),
       notification_manager_(notification_manager),
       extension_service_(extension_service),
       signin_manager_(signin_manager),
+      worker_task_runner_(worker_task_runner),
       weak_ptr_factory_(this) {}
 
 void SyncEngine::DidProcessRemoteChange(RemoteToLocalSyncer* syncer) {
   if (syncer->sync_action() != SYNC_ACTION_NONE && syncer->url().is_valid()) {
     FOR_EACH_OBSERVER(FileStatusObserver,
                       file_status_observers_,
                       OnFileStatusChanged(syncer->url(),
                                           SYNC_FILE_STATUS_SYNCED,
                                           syncer->sync_action(),
                                           SYNC_DIRECTION_REMOTE_TO_LOCAL));
@@ skipping 67 matching lines @@
 void SyncEngine::NotifyLastOperationStatus() {
   FOR_EACH_OBSERVER(
       Observer,
       service_observers_,
       OnRemoteChangeQueueUpdated(
           GetMetadataDatabase()->CountDirtyTracker()));
 }
 
 }  // namespace drive_backend
 }  // namespace sync_file_system