src/chrome/browser/sync_file_system/drive_backend/sync_task_manager.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "chrome/browser/sync_file_system/drive_backend/sync_task_manager.h"
   6
   7 #include "base/bind.h"
   8 #include "base/location.h"
   9 #include "base/memory/scoped_ptr.h"
  10 #include "chrome/browser/sync_file_system/drive_backend/sync_task_token.h"
  11 #include "chrome/browser/sync_file_system/sync_file_metadata.h"
  12
  13 using fileapi::FileSystemURL;
  14
  15 namespace sync_file_system {
  16 namespace drive_backend {
  17
  18 namespace {
  19
  20 class SyncTaskAdapter : public ExclusiveTask {
  21  public:
  22   explicit SyncTaskAdapter(const SyncTaskManager::Task& task) : task_(task) {}
  23   virtual ~SyncTaskAdapter() {}
  24
  25   virtual void RunExclusive(const SyncStatusCallback& callback) OVERRIDE {
  26     task_.Run(callback);
  27   }
  28
  29  private:
  30   SyncTaskManager::Task task_;
  31
  32   DISALLOW_COPY_AND_ASSIGN(SyncTaskAdapter);
  33 };
  34
  35 }  // namespace
  36
  37 SyncTaskManager::PendingTask::PendingTask() {}
  38
  39 SyncTaskManager::PendingTask::PendingTask(
  40     const base::Closure& task, Priority pri, int seq)
  41     : task(task), priority(pri), seq(seq) {}
  42
  43 SyncTaskManager::PendingTask::~PendingTask() {}
  44
  45 bool SyncTaskManager::PendingTaskComparator::operator()(
  46     const PendingTask& left,
  47     const PendingTask& right) const {
  48   if (left.priority != right.priority)
  49     return left.priority < right.priority;
  50   return left.seq > right.seq;
  51 }
  52
  53 SyncTaskManager::SyncTaskManager(
  54     base::WeakPtr<Client> client,
  55     size_t maximum_background_task)
  56     : client_(client),
  57       maximum_background_task_(maximum_background_task),
  58       pending_task_seq_(0),
  59       task_token_seq_(SyncTaskToken::kMinimumBackgroundTaskTokenID) {
  60 }
  61
  62 SyncTaskManager::~SyncTaskManager() {
  63   client_.reset();
  64   token_.reset();
  65 }
  66
  67 void SyncTaskManager::Initialize(SyncStatusCode status) {
  68   DCHECK(!token_);
  69   NotifyTaskDone(SyncTaskToken::CreateForForegroundTask(AsWeakPtr()),
  70                  status);
  71 }
  72
  73 void SyncTaskManager::ScheduleTask(
  74     const tracked_objects::Location& from_here,
  75     const Task& task,
  76     Priority priority,
  77     const SyncStatusCallback& callback) {
  78   ScheduleSyncTask(from_here,
  79                    scoped_ptr<SyncTask>(new SyncTaskAdapter(task)),
  80                    priority,
  81                    callback);
  82 }
  83
  84 void SyncTaskManager::ScheduleSyncTask(
  85     const tracked_objects::Location& from_here,
  86     scoped_ptr<SyncTask> task,
  87     Priority priority,
  88     const SyncStatusCallback& callback) {
  89   scoped_ptr<SyncTaskToken> token(GetToken(from_here, callback));
  90   if (!token) {
  91     PushPendingTask(
  92         base::Bind(&SyncTaskManager::ScheduleSyncTask, AsWeakPtr(), from_here,
  93                    base::Passed(&task), priority, callback),
  94         priority);
  95     return;
  96   }
  97   RunTask(token.Pass(), task.Pass());
  98 }
  99
 100 bool SyncTaskManager::ScheduleTaskIfIdle(
 101         const tracked_objects::Location& from_here,
 102         const Task& task,
 103         const SyncStatusCallback& callback) {
 104   return ScheduleSyncTaskIfIdle(
 105       from_here,
 106       scoped_ptr<SyncTask>(new SyncTaskAdapter(task)),
 107       callback);
 108 }
 109
 110 bool SyncTaskManager::ScheduleSyncTaskIfIdle(
 111     const tracked_objects::Location& from_here,
 112     scoped_ptr<SyncTask> task,
 113     const SyncStatusCallback& callback) {
 114   scoped_ptr<SyncTaskToken> token(GetToken(from_here, callback));
 115   if (!token)
 116     return false;
 117   RunTask(token.Pass(), task.Pass());
 118   return true;
 119 }
 120
 121 // static
 122 void SyncTaskManager::NotifyTaskDone(scoped_ptr<SyncTaskToken> token,
 123                                      SyncStatusCode status) {
 124   DCHECK(token);
 125
 126   SyncTaskManager* manager = token->manager();
 127   if (manager)
 128     manager->NotifyTaskDoneBody(token.Pass(), status);
 129 }
 130
 131 // static
 132 void SyncTaskManager::UpdateBlockingFactor(
 133     scoped_ptr<SyncTaskToken> current_task_token,
 134     scoped_ptr<BlockingFactor> blocking_factor,
 135     const Continuation& continuation) {
 136   DCHECK(current_task_token);
 137
 138   SyncTaskManager* manager = current_task_token->manager();
 139   if (!manager)
 140     return;
 141
 142   scoped_ptr<SyncTaskToken> foreground_task_token;
 143   scoped_ptr<SyncTaskToken> background_task_token;
 144   if (current_task_token->token_id() == SyncTaskToken::kForegroundTaskTokenID)
 145     foreground_task_token = current_task_token.Pass();
 146   else
 147     background_task_token = current_task_token.Pass();
 148
 149   manager->UpdateBlockingFactorBody(foreground_task_token.Pass(),
 150                                     background_task_token.Pass(),
 151                                     blocking_factor.Pass(),
 152                                     continuation);
 153 }
 154
 155 bool SyncTaskManager::IsRunningTask(int64 token_id) const {
 156   // If the client is gone, all task should be aborted.
 157   if (!client_)
 158     return false;
 159
 160   if (token_id == SyncTaskToken::kForegroundTaskTokenID)
 161     return true;
 162
 163   return ContainsKey(running_background_tasks_, token_id);
 164 }
 165
 166 void SyncTaskManager::NotifyTaskDoneBody(scoped_ptr<SyncTaskToken> token,
 167                                          SyncStatusCode status) {
 168   DCHECK(token);
 169
 170   DVLOG(3) << "NotifyTaskDone: " << "finished with status=" << status
 171            << " (" << SyncStatusCodeToString(status) << ")"
 172            << " " << token_->location().ToString();
 173
 174   if (token->blocking_factor()) {
 175     dependency_manager_.Erase(token->blocking_factor());
 176     token->clear_blocking_factor();
 177   }
 178
 179   scoped_ptr<SyncTask> task;
 180   SyncStatusCallback callback = token->callback();
 181   token->clear_callback();
 182   if (token->token_id() == SyncTaskToken::kForegroundTaskTokenID) {
 183     token_ = token.Pass();
 184     task = running_foreground_task_.Pass();
 185   } else {
 186     task = running_background_tasks_.take_and_erase(token->token_id());
 187   }
 188
 189   bool task_used_network = false;
 190   if (task)
 191     task_used_network = task->used_network();
 192
 193   if (client_)
 194     client_->NotifyLastOperationStatus(status, task_used_network);
 195
 196   if (!callback.is_null())
 197     callback.Run(status);
 198
 199   StartNextTask();
 200 }
 201
 202 void SyncTaskManager::UpdateBlockingFactorBody(
 203     scoped_ptr<SyncTaskToken> foreground_task_token,
 204     scoped_ptr<SyncTaskToken> background_task_token,
 205     scoped_ptr<BlockingFactor> blocking_factor,
 206     const Continuation& continuation) {
 207   // Run the task directly if the parallelization is disabled.
 208   if (!maximum_background_task_) {
 209     DCHECK(foreground_task_token);
 210     DCHECK(!background_task_token);
 211     continuation.Run(foreground_task_token.Pass());
 212     return;
 213   }
 214
 215   // Clear existing |blocking_factor| from |dependency_manager_| before
 216   // getting |foreground_task_token|, so that we can avoid dead lock.
 217   if (background_task_token && background_task_token->blocking_factor()) {
 218     dependency_manager_.Erase(background_task_token->blocking_factor());
 219     background_task_token->clear_blocking_factor();
 220   }
 221
 222   // Try to get |foreground_task_token|.  If it's not available, wait for
 223   // current foreground task to finish.
 224   if (!foreground_task_token) {
 225     DCHECK(background_task_token);
 226     foreground_task_token = GetToken(background_task_token->location(),
 227                                      SyncStatusCallback());
 228     if (!foreground_task_token) {
 229       PushPendingTask(
 230           base::Bind(&SyncTaskManager::UpdateBlockingFactorBody,
 231                      AsWeakPtr(),
 232                      base::Passed(&foreground_task_token),
 233                      base::Passed(&background_task_token),
 234                      base::Passed(&blocking_factor),
 235                      continuation),
 236           PRIORITY_HIGH);
 237       StartNextTask();
 238       return;
 239     }
 240   }
 241
 242   // Check if the task can run as a background task now.
 243   // If there are too many task running or any other task blocks current
 244   // task, wait for any other task to finish.
 245   bool task_number_limit_exceeded =
 246       !background_task_token &&
 247       running_background_tasks_.size() >= maximum_background_task_;
 248   if (task_number_limit_exceeded ||
 249       !dependency_manager_.Insert(blocking_factor.get())) {
 250     DCHECK(!running_background_tasks_.empty());
 251     DCHECK(pending_backgrounding_task_.is_null());
 252
 253     // Wait for NotifyTaskDone to release a |blocking_factor|.
 254     pending_backgrounding_task_ =
 255         base::Bind(&SyncTaskManager::UpdateBlockingFactorBody,
 256                    AsWeakPtr(),
 257                    base::Passed(&foreground_task_token),
 258                    base::Passed(&background_task_token),
 259                    base::Passed(&blocking_factor),
 260                    continuation);
 261     return;
 262   }
 263
 264   if (background_task_token) {
 265     background_task_token->set_blocking_factor(blocking_factor.Pass());
 266   } else {
 267     tracked_objects::Location from_here = foreground_task_token->location();
 268     SyncStatusCallback callback = foreground_task_token->callback();
 269     foreground_task_token->clear_callback();
 270
 271     background_task_token =
 272         SyncTaskToken::CreateForBackgroundTask(
 273             AsWeakPtr(),
 274             task_token_seq_++,
 275             blocking_factor.Pass());
 276     background_task_token->UpdateTask(from_here, callback);
 277     running_background_tasks_.set(background_task_token->token_id(),
 278                                   running_foreground_task_.Pass());
 279   }
 280
 281   token_ = foreground_task_token.Pass();
 282   StartNextTask();
 283   continuation.Run(background_task_token.Pass());
 284 }
 285
 286 scoped_ptr<SyncTaskToken> SyncTaskManager::GetToken(
 287     const tracked_objects::Location& from_here,
 288     const SyncStatusCallback& callback) {
 289   if (!token_)
 290     return scoped_ptr<SyncTaskToken>();
 291   token_->UpdateTask(from_here, callback);
 292   return token_.Pass();
 293 }
 294
 295 void SyncTaskManager::PushPendingTask(
 296     const base::Closure& closure, Priority priority) {
 297   pending_tasks_.push(PendingTask(closure, priority, pending_task_seq_++));
 298 }
 299
 300 void SyncTaskManager::RunTask(scoped_ptr<SyncTaskToken> token,
 301                               scoped_ptr<SyncTask> task) {
 302   DCHECK(!running_foreground_task_);
 303   running_foreground_task_ = task.Pass();
 304   running_foreground_task_->RunPreflight(token.Pass());
 305 }
 306
 307 void SyncTaskManager::StartNextTask() {
 308   if (!pending_backgrounding_task_.is_null()) {
 309     base::Closure closure = pending_backgrounding_task_;
 310     pending_backgrounding_task_.Reset();
 311     closure.Run();
 312     return;
 313   }
 314
 315   if (!pending_tasks_.empty()) {
 316     base::Closure closure = pending_tasks_.top().task;
 317     pending_tasks_.pop();
 318     closure.Run();
 319     return;
 320   }
 321
 322   if (client_)
 323     client_->MaybeScheduleNextTask();
 324 }
 325
 326 }  // namespace drive_backend
 327 }  // namespace sync_file_system