Keep sync tasks alive as long as it's not finished

chrome/browser/chromeos/drive/sync_client.cc

 // Copyright (c) 2012 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/chromeos/drive/sync_client.h"
 
 #include <vector>
 
 #include "base/bind.h"
 #include "base/message_loop/message_loop_proxy.h"
@@ skipping 103 matching lines @@
     if (error != FILE_ERROR_OK) {
       LOG(WARNING) << "Failed to pin cache entry: " << local_id;
       continue;
     }
 
     local_ids->push_back(local_id);
   }
   DCHECK(!it->HasError());
 }
 
-// Runs the task and returns a dummy cancel closure.
-base::Closure RunTaskAndReturnDummyCancelClosure(const base::Closure& task) {
-  task.Run();
-  return base::Closure();
-}
-
 }  // namespace
 
-SyncClient::SyncTask::SyncTask() : state(PENDING), should_run_again(false) {}
+SyncClient::SyncTask::SyncTask()
+    : state(SUSPENDED), context(BACKGROUND), should_run_again(false) {}
 SyncClient::SyncTask::~SyncTask() {}
 
 SyncClient::SyncClient(base::SequencedTaskRunner* blocking_task_runner,
                        file_system::OperationObserver* observer,
                        JobScheduler* scheduler,
                        ResourceMetadata* metadata,
                        FileCache* cache,
                        LoaderController* loader_controller,
                        const base::FilePath& temporary_file_directory)
     : blocking_task_runner_(blocking_task_runner),
@@ skipping 59 matching lines @@
 
 void SyncClient::RemoveFetchTask(const std::string& local_id) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 
   SyncTasks::iterator it = tasks_.find(SyncTasks::key_type(FETCH, local_id));
   if (it == tasks_.end())
     return;
 
   SyncTask* task = &it->second;
   switch (task->state) {
+    case SUSPENDED:
     case PENDING:
-      tasks_.erase(it);
+      OnTaskComplete(FETCH, local_id, FILE_ERROR_ABORT);
       break;
     case RUNNING:
       if (!task->cancel_closure.is_null())
         task->cancel_closure.Run();
       break;
   }
 }
 
 void SyncClient::AddUpdateTask(const ClientContext& context,
                                const std::string& local_id) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
   AddUpdateTaskInternal(context, local_id, delay_);
 }
 
+base::Closure SyncClient::PerformFetchTask(const std::string& local_id,
+                                           const ClientContext& context) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  return download_operation_->EnsureFileDownloadedByLocalId(
+      local_id,
+      context,
+      GetFileContentInitializedCallback(),
+      google_apis::GetContentCallback(),
+      base::Bind(&SyncClient::OnFetchFileComplete,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 local_id));
+}
+
 void SyncClient::AddFetchTaskInternal(const std::string& local_id,
                                       const base::TimeDelta& delay) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 
   SyncTask task;
-  task.task = base::Bind(
-      &file_system::DownloadOperation::EnsureFileDownloadedByLocalId,
-      base::Unretained(download_operation_.get()),
+  task.state = PENDING;
+  task.context = ClientContext(BACKGROUND);
+  task.task = base::Bind(&SyncClient::PerformFetchTask,
+                         base::Unretained(this),
+                         local_id);
+  AddTask(SyncTasks::key_type(FETCH, local_id), task, delay);
+}
+
+base::Closure SyncClient::PerformUpdateTask(const std::string& local_id,
+                                            const ClientContext& context) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  entry_update_performer_->UpdateEntry(
       local_id,
-      ClientContext(BACKGROUND),
-      GetFileContentInitializedCallback(),
-      google_apis::GetContentCallback(),
-      base::Bind(&SyncClient::OnFetchFileComplete,
+      context,
+      base::Bind(&SyncClient::OnTaskComplete,
                  weak_ptr_factory_.GetWeakPtr(),
+                 UPDATE,
                  local_id));
-  AddTask(SyncTasks::key_type(FETCH, local_id), task, delay);
+  return base::Closure();
 }
 
 void SyncClient::AddUpdateTaskInternal(const ClientContext& context,
                                        const std::string& local_id,
                                        const base::TimeDelta& delay) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
   SyncTask task;
-  task.task = base::Bind(
-      &RunTaskAndReturnDummyCancelClosure,
-      base::Bind(&EntryUpdatePerformer::UpdateEntry,
-                 base::Unretained(entry_update_performer_.get()),
-                 local_id,
-                 context,
-                 base::Bind(&SyncClient::OnUpdateComplete,
-                            weak_ptr_factory_.GetWeakPtr(),
-                            local_id)));
+  task.state = PENDING;
+  task.context = context;
+  task.task = base::Bind(&SyncClient::PerformUpdateTask,
+                         base::Unretained(this),
+                         local_id);
   AddTask(SyncTasks::key_type(UPDATE, local_id), task, delay);
 }
 
 void SyncClient::AddTask(const SyncTasks::key_type& key,
                          const SyncTask& task,
                          const base::TimeDelta& delay) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 
   SyncTasks::iterator it = tasks_.find(key);
   if (it != tasks_.end()) {
     switch (it->second.state) {
+      case SUSPENDED:
+        // Activate the task.
+        it->second.state = PENDING;
+        break;
       case PENDING:
         // The same task will run, do nothing.
-        break;
+        return;
       case RUNNING:
         // Something has changed since the task started. Schedule rerun.
         it->second.should_run_again = true;
-        break;
+        return;
     }
-    return;
+  } else {
+    tasks_[key] = task;
   }
-
   DCHECK_EQ(PENDING, task.state);
-  tasks_[key] = task;
-
   base::MessageLoopProxy::current()->PostDelayedTask(
       FROM_HERE,
       base::Bind(&SyncClient::StartTask, weak_ptr_factory_.GetWeakPtr(), key),
       delay);
 }
 
 void SyncClient::StartTask(const SyncTasks::key_type& key) {
   SyncTasks::iterator it = tasks_.find(key);
   if (it == tasks_.end())
     return;
 
   SyncTask* task = &it->second;
   switch (task->state) {
+    case SUSPENDED:
     case PENDING:
       task->state = RUNNING;
-      task->cancel_closure = task->task.Run();
+      task->cancel_closure = task->task.Run(task->context);
       break;
     case RUNNING:  // Do nothing.
       break;
   }
 }
 
 void SyncClient::OnGetLocalIdsOfBacklog(
     const std::vector<std::string>* to_fetch,
     const std::vector<std::string>* to_update) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
@@ skipping 13 matching lines @@
   }
 }
 
 void SyncClient::AddFetchTasks(const std::vector<std::string>* local_ids) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 
   for (size_t i = 0; i < local_ids->size(); ++i)
     AddFetchTask((*local_ids)[i]);
 }
 
-bool SyncClient::OnTaskComplete(SyncType type, const std::string& local_id) {
+void SyncClient::OnTaskComplete(SyncType type,
+                                const std::string& local_id,
+                                FileError error) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 
   const SyncTasks::key_type key(type, local_id);
   SyncTasks::iterator it = tasks_.find(key);
   DCHECK(it != tasks_.end());
 
+  bool should_erase_task = true;
+  base::TimeDelta retry_delay = base::TimeDelta::FromSeconds(0);
+
+  switch (error) {
+    case FILE_ERROR_OK:
+      DVLOG(1) << "Completed: type = " << type << ", id = " << local_id;
+      break;
+    case FILE_ERROR_ABORT:
+      // Ignore it because this is caused by user's cancel operations.
+      break;
+    case FILE_ERROR_NO_CONNECTION:
+      // Run the task again so that we'll retry once the connection is back.
+      it->second.should_run_again = true;
+      it->second.context = ClientContext(BACKGROUND);
+      break;
+    case FILE_ERROR_SERVICE_UNAVAILABLE:
+      // Run the task again so that we'll retry once the service is back.
+      it->second.should_run_again = true;
+      it->second.context = ClientContext(BACKGROUND);
+      retry_delay = long_delay_;
+      operation_observer_->OnDriveSyncError(
+          file_system::DRIVE_SYNC_ERROR_SERVICE_UNAVAILABLE, local_id);
+      break;
+    case FILE_ERROR_PARENT_NEEDS_TO_BE_SYNCED: {
+      // Suspend the task and register it as a dependent task of the parent.
+      should_erase_task = false;
+      it->second.state = SUSPENDED;
+      ResourceEntry* entry = new ResourceEntry;
+      base::PostTaskAndReplyWithResult(
+          blocking_task_runner_.get(),
+          FROM_HERE,
+          base::Bind(&ResourceMetadata::GetResourceEntryById,
+                     base::Unretained(metadata_),
+                     local_id,
+                     entry),
+          base::Bind(&SyncClient::AddDependentTask,
+                     weak_ptr_factory_.GetWeakPtr(),
+                     type,
+                     local_id,
+                     base::Owned(entry)));
+      break;
+    }
+    default:
+      operation_observer_->OnDriveSyncError(
+          file_system::DRIVE_SYNC_ERROR_MISC, local_id);
+      LOG(WARNING) << "Failed: type = " << type << ", id = " << local_id
+                   << ": " << FileErrorToString(error);
+  }
+
   if (it->second.should_run_again) {
     DVLOG(1) << "Running again: type = " << type << ", id = " << local_id;
+    it->second.state = PENDING;
     it->second.should_run_again = false;
-    it->second.task.Run();
-    return false;
+    base::MessageLoopProxy::current()->PostDelayedTask(
+        FROM_HERE,
+        base::Bind(&SyncClient::StartTask, weak_ptr_factory_.GetWeakPtr(), key),
+        retry_delay);
+  } else if (should_erase_task) {
+    for (size_t i = 0; i < it->second.dependent_tasks.size(); ++i)
+      StartTask(it->second.dependent_tasks[i]);
+    tasks_.erase(it);
   }
-
-  tasks_.erase(it);
-  return true;
 }
 
 void SyncClient::OnFetchFileComplete(const std::string& local_id,
                                      FileError error,
                                      const base::FilePath& local_path,
                                      scoped_ptr<ResourceEntry> entry) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
-
-  if (!OnTaskComplete(FETCH, local_id))
-    return;
-
-  if (error == FILE_ERROR_OK) {
-    DVLOG(1) << "Fetched " << local_id << ": " << local_path.value();
-  } else {
-    switch (error) {
-      case FILE_ERROR_ABORT:
-        // If user cancels download, unpin the file so that we do not sync the
-        // file again.
-        base::PostTaskAndReplyWithResult(
-            blocking_task_runner_,
-            FROM_HERE,
-            base::Bind(&FileCache::Unpin, base::Unretained(cache_), local_id),
-            base::Bind(&util::EmptyFileOperationCallback));
-        break;
-      case FILE_ERROR_NO_CONNECTION:
-        // Add the task again so that we'll retry once the connection is back.
-        AddFetchTaskInternal(local_id, delay_);
-        break;
-      case FILE_ERROR_SERVICE_UNAVAILABLE:
-        // Add the task again so that we'll retry once the service is back.
-        AddFetchTaskInternal(local_id, long_delay_);
-        operation_observer_->OnDriveSyncError(
-            file_system::DRIVE_SYNC_ERROR_SERVICE_UNAVAILABLE,
-            local_id);
-        break;
-      default:
-        operation_observer_->OnDriveSyncError(
-            file_system::DRIVE_SYNC_ERROR_MISC,
-            local_id);
-        LOG(WARNING) << "Failed to fetch " << local_id
-                     << ": " << FileErrorToString(error);
-    }
+  OnTaskComplete(FETCH, local_id, error);
+  if (error == FILE_ERROR_ABORT) {
+    // If user cancels download, unpin the file so that we do not sync the file
+    // again.
+    base::PostTaskAndReplyWithResult(
+        blocking_task_runner_,
+        FROM_HERE,
+        base::Bind(&FileCache::Unpin, base::Unretained(cache_), local_id),
+        base::Bind(&util::EmptyFileOperationCallback));
   }
 }
 
-void SyncClient::OnUpdateComplete(const std::string& local_id,
+void SyncClient::AddDependentTask(SyncType type,
+                                  const std::string& local_id,
+                                  const ResourceEntry* entry,
                                   FileError error) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  if (error != FILE_ERROR_OK) {
+    OnTaskComplete(type, local_id, error);
+    return;
+  }
 
-  if (!OnTaskComplete(UPDATE, local_id))
+  const SyncTasks::key_type key_parent(type, entry->parent_local_id());
+  SyncTasks::iterator it_parent = tasks_.find(key_parent);
+  if (it_parent == tasks_.end()) {

[kinaba, 2014/07/17 06:07:46]

Can't there be a danger that the parent task has completed during the blocking
GetResourceEntryById operation? Is it still ok to fail in such a case?

[hashimoto, 2014/07/18 06:12:16]

Extremely good catch.
Changed EntryUpdatePerformer to return the parent's local ID when necessary.

The new patch set fixes the addressed race problem while keeping the following
two contracts:
1. No AddTask() call should not result in a chain of an unlimited number of
tasks.
2. No task should stay in the SUSPENDED state forever.

+    OnTaskComplete(type, local_id, FILE_ERROR_NOT_FOUND);
     return;
+  }
 
-  if (error == FILE_ERROR_OK) {
-    DVLOG(1) << "Updated " << local_id;
-
-    // Add update tasks for child entries which may be waiting for the parent to
-    // be updated.
-    ResourceEntryVector* entries = new ResourceEntryVector;
-    base::PostTaskAndReplyWithResult(
-        blocking_task_runner_.get(),
-        FROM_HERE,
-        base::Bind(&ResourceMetadata::ReadDirectoryById,
-                   base::Unretained(metadata_), local_id, entries),
-        base::Bind(&SyncClient::AddChildUpdateTasks,
-                   weak_ptr_factory_.GetWeakPtr(), base::Owned(entries)));
-  } else {
-    switch (error) {
-      case FILE_ERROR_ABORT:
-        // Ignore it because this is caused by user's cancel operations.
-        break;
-      case FILE_ERROR_NO_CONNECTION:
-        // Add the task again so that we'll retry once the connection is back.
-        AddUpdateTaskInternal(ClientContext(BACKGROUND), local_id,
-                              base::TimeDelta::FromSeconds(0));
-        break;
-      case FILE_ERROR_SERVICE_UNAVAILABLE:
-        // Add the task again so that we'll retry once the service is back.
-        AddUpdateTaskInternal(ClientContext(BACKGROUND), local_id, long_delay_);
-        operation_observer_->OnDriveSyncError(
-            file_system::DRIVE_SYNC_ERROR_SERVICE_UNAVAILABLE,
-            local_id);
-        break;
-      default:
-        operation_observer_->OnDriveSyncError(
-            file_system::DRIVE_SYNC_ERROR_MISC,
-            local_id);
-        LOG(WARNING) << "Failed to update " << local_id << ": "
-                     << FileErrorToString(error);
-    }
-  }
-}
-
-void SyncClient::AddChildUpdateTasks(const ResourceEntryVector* entries,
-                                     FileError error) {
-  if (error != FILE_ERROR_OK)
-    return;
-
-  for (size_t i = 0; i < entries->size(); ++i) {
-    const ResourceEntry& entry = (*entries)[i];
-    if (entry.metadata_edit_state() != ResourceEntry::CLEAN) {
-      AddUpdateTaskInternal(ClientContext(BACKGROUND), entry.local_id(),
-                            base::TimeDelta::FromSeconds(0));
-    }
-  }
+  it_parent->second.dependent_tasks.push_back(
+      SyncTasks::key_type(type, local_id));
 }
 
 }  // namespace internal
 }  // namespace drive