Home | History | Annotate | Download | only in drive_backend 
      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 "base/sequenced_task_runner.h"
     11 #include "chrome/browser/sync_file_system/drive_backend/sync_task.h"
     12 #include "chrome/browser/sync_file_system/drive_backend/sync_task_token.h"
     13 #include "chrome/browser/sync_file_system/sync_file_metadata.h"
     14 
     15 using fileapi::FileSystemURL;
     16 
     17 namespace sync_file_system {
     18 namespace drive_backend {
     19 
     20 namespace {
     21 
     22 class SyncTaskAdapter : public ExclusiveTask {
     23  public:
     24   explicit SyncTaskAdapter(const SyncTaskManager::Task& task) : task_(task) {}
     25   virtual ~SyncTaskAdapter() {}
     26 
     27   virtual void RunExclusive(const SyncStatusCallback& callback) OVERRIDE {
     28     task_.Run(callback);
     29   }
     30 
     31  private:
     32   SyncTaskManager::Task task_;
     33 
     34   DISALLOW_COPY_AND_ASSIGN(SyncTaskAdapter);
     35 };
     36 
     37 }  // namespace
     38 
     39 SyncTaskManager::PendingTask::PendingTask() {}
     40 
     41 SyncTaskManager::PendingTask::PendingTask(
     42     const base::Closure& task, Priority pri, int seq)
     43     : task(task), priority(pri), seq(seq) {}
     44 
     45 SyncTaskManager::PendingTask::~PendingTask() {}
     46 
     47 bool SyncTaskManager::PendingTaskComparator::operator()(
     48     const PendingTask& left,
     49     const PendingTask& right) const {
     50   if (left.priority != right.priority)
     51     return left.priority < right.priority;
     52   return left.seq > right.seq;
     53 }
     54 
     55 SyncTaskManager::SyncTaskManager(
     56     base::WeakPtr<Client> client,
     57     size_t maximum_background_task,
     58     base::SequencedTaskRunner* task_runner)
     59     : client_(client),
     60       maximum_background_task_(maximum_background_task),
     61       pending_task_seq_(0),
     62       task_token_seq_(SyncTaskToken::kMinimumBackgroundTaskTokenID),
     63       task_runner_(task_runner) {
     64 }
     65 
     66 SyncTaskManager::~SyncTaskManager() {
     67   client_.reset();
     68   token_.reset();
     69 }
     70 
     71 void SyncTaskManager::Initialize(SyncStatusCode status) {
     72   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
     73   DCHECK(!token_);
     74   NotifyTaskDone(SyncTaskToken::CreateForForegroundTask(AsWeakPtr()),
     75                  status);
     76 }
     77 
     78 void SyncTaskManager::ScheduleTask(
     79     const tracked_objects::Location& from_here,
     80     const Task& task,
     81     Priority priority,
     82     const SyncStatusCallback& callback) {
     83   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
     84 
     85   ScheduleSyncTask(from_here,
     86                    scoped_ptr<SyncTask>(new SyncTaskAdapter(task)),
     87                    priority,
     88                    callback);
     89 }
     90 
     91 void SyncTaskManager::ScheduleSyncTask(
     92     const tracked_objects::Location& from_here,
     93     scoped_ptr<SyncTask> task,
     94     Priority priority,
     95     const SyncStatusCallback& callback) {
     96   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
     97 
     98   scoped_ptr<SyncTaskToken> token(GetToken(from_here, callback));
     99   if (!token) {
    100     PushPendingTask(
    101         base::Bind(&SyncTaskManager::ScheduleSyncTask, AsWeakPtr(), from_here,
    102                    base::Passed(&task), priority, callback),
    103         priority);
    104     return;
    105   }
    106   RunTask(token.Pass(), task.Pass());
    107 }
    108 
    109 bool SyncTaskManager::ScheduleTaskIfIdle(
    110         const tracked_objects::Location& from_here,
    111         const Task& task,
    112         const SyncStatusCallback& callback) {
    113   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    114 
    115   return ScheduleSyncTaskIfIdle(
    116       from_here,
    117       scoped_ptr<SyncTask>(new SyncTaskAdapter(task)),
    118       callback);
    119 }
    120 
    121 bool SyncTaskManager::ScheduleSyncTaskIfIdle(
    122     const tracked_objects::Location& from_here,
    123     scoped_ptr<SyncTask> task,
    124     const SyncStatusCallback& callback) {
    125   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    126 
    127   scoped_ptr<SyncTaskToken> token(GetToken(from_here, callback));
    128   if (!token)
    129     return false;
    130   RunTask(token.Pass(), task.Pass());
    131   return true;
    132 }
    133 
    134 // static
    135 void SyncTaskManager::NotifyTaskDone(scoped_ptr<SyncTaskToken> token,
    136                                      SyncStatusCode status) {
    137   DCHECK(token);
    138 
    139   SyncTaskManager* manager = token->manager();
    140   if (token->token_id() == SyncTaskToken::kTestingTaskTokenID) {
    141     DCHECK(!manager);
    142     SyncStatusCallback callback = token->callback();
    143     token->clear_callback();
    144     callback.Run(status);
    145     return;
    146   }
    147 
    148   if (manager)
    149     manager->NotifyTaskDoneBody(token.Pass(), status);
    150 }
    151 
    152 // static
    153 void SyncTaskManager::UpdateBlockingFactor(
    154     scoped_ptr<SyncTaskToken> current_task_token,
    155     scoped_ptr<BlockingFactor> blocking_factor,
    156     const Continuation& continuation) {
    157   DCHECK(current_task_token);
    158 
    159   SyncTaskManager* manager = current_task_token->manager();
    160   if (current_task_token->token_id() == SyncTaskToken::kTestingTaskTokenID) {
    161     DCHECK(!manager);
    162     continuation.Run(current_task_token.Pass());
    163     return;
    164   }
    165 
    166   if (!manager)
    167     return;
    168 
    169   scoped_ptr<SyncTaskToken> foreground_task_token;
    170   scoped_ptr<SyncTaskToken> background_task_token;
    171   scoped_ptr<TaskLogger::TaskLog> task_log = current_task_token->PassTaskLog();
    172   if (current_task_token->token_id() == SyncTaskToken::kForegroundTaskTokenID)
    173     foreground_task_token = current_task_token.Pass();
    174   else
    175     background_task_token = current_task_token.Pass();
    176 
    177   manager->UpdateBlockingFactorBody(foreground_task_token.Pass(),
    178                                     background_task_token.Pass(),
    179                                     task_log.Pass(),
    180                                     blocking_factor.Pass(),
    181                                     continuation);
    182 }
    183 
    184 bool SyncTaskManager::IsRunningTask(int64 token_id) const {
    185   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    186 
    187   // If the client is gone, all task should be aborted.
    188   if (!client_)
    189     return false;
    190 
    191   if (token_id == SyncTaskToken::kForegroundTaskTokenID)
    192     return true;
    193 
    194   return ContainsKey(running_background_tasks_, token_id);
    195 }
    196 
    197 void SyncTaskManager::DetachFromSequence() {
    198   sequence_checker_.DetachFromSequence();
    199 }
    200 
    201 void SyncTaskManager::NotifyTaskDoneBody(scoped_ptr<SyncTaskToken> token,
    202                                          SyncStatusCode status) {
    203   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    204   DCHECK(token);
    205 
    206   DVLOG(3) << "NotifyTaskDone: " << "finished with status=" << status
    207            << " (" << SyncStatusCodeToString(status) << ")"
    208            << " " << token_->location().ToString();
    209 
    210   if (token->blocking_factor()) {
    211     dependency_manager_.Erase(token->blocking_factor());
    212     token->clear_blocking_factor();
    213   }
    214 
    215   if (client_) {
    216     if (token->has_task_log()) {
    217       token->FinalizeTaskLog(SyncStatusCodeToString(status));
    218       client_->RecordTaskLog(token->PassTaskLog());
    219     }
    220   }
    221 
    222   scoped_ptr<SyncTask> task;
    223   SyncStatusCallback callback = token->callback();
    224   token->clear_callback();
    225   if (token->token_id() == SyncTaskToken::kForegroundTaskTokenID) {
    226     token_ = token.Pass();
    227     task = running_foreground_task_.Pass();
    228   } else {
    229     task = running_background_tasks_.take_and_erase(token->token_id());
    230   }
    231 
    232   // Acquire the token to prevent a new task to jump into the queue.
    233   token = token_.Pass();
    234 
    235   bool task_used_network = false;
    236   if (task)
    237     task_used_network = task->used_network();
    238 
    239   if (client_)
    240     client_->NotifyLastOperationStatus(status, task_used_network);
    241 
    242   if (!callback.is_null())
    243     callback.Run(status);
    244 
    245   // Post MaybeStartNextForegroundTask rather than calling it directly to avoid
    246   // making the call-chaing longer.
    247   task_runner_->PostTask(
    248       FROM_HERE,
    249       base::Bind(&SyncTaskManager::MaybeStartNextForegroundTask,
    250                  AsWeakPtr(), base::Passed(&token)));
    251 }
    252 
    253 void SyncTaskManager::UpdateBlockingFactorBody(
    254     scoped_ptr<SyncTaskToken> foreground_task_token,
    255     scoped_ptr<SyncTaskToken> background_task_token,
    256     scoped_ptr<TaskLogger::TaskLog> task_log,
    257     scoped_ptr<BlockingFactor> blocking_factor,
    258     const Continuation& continuation) {
    259   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    260 
    261   // Run the task directly if the parallelization is disabled.
    262   if (!maximum_background_task_) {
    263     DCHECK(foreground_task_token);
    264     DCHECK(!background_task_token);
    265     foreground_task_token->SetTaskLog(task_log.Pass());
    266     continuation.Run(foreground_task_token.Pass());
    267     return;
    268   }
    269 
    270   // Clear existing |blocking_factor| from |dependency_manager_| before
    271   // getting |foreground_task_token|, so that we can avoid dead lock.
    272   if (background_task_token && background_task_token->blocking_factor()) {
    273     dependency_manager_.Erase(background_task_token->blocking_factor());
    274     background_task_token->clear_blocking_factor();
    275   }
    276 
    277   // Try to get |foreground_task_token|.  If it's not available, wait for
    278   // current foreground task to finish.
    279   if (!foreground_task_token) {
    280     DCHECK(background_task_token);
    281     foreground_task_token = GetToken(background_task_token->location(),
    282                                      SyncStatusCallback());
    283     if (!foreground_task_token) {
    284       PushPendingTask(
    285           base::Bind(&SyncTaskManager::UpdateBlockingFactorBody,
    286                      AsWeakPtr(),
    287                      base::Passed(&foreground_task_token),
    288                      base::Passed(&background_task_token),
    289                      base::Passed(&task_log),
    290                      base::Passed(&blocking_factor),
    291                      continuation),
    292           PRIORITY_HIGH);
    293       MaybeStartNextForegroundTask(scoped_ptr<SyncTaskToken>());
    294       return;
    295     }
    296   }
    297 
    298   // Check if the task can run as a background task now.
    299   // If there are too many task running or any other task blocks current
    300   // task, wait for any other task to finish.
    301   bool task_number_limit_exceeded =
    302       !background_task_token &&
    303       running_background_tasks_.size() >= maximum_background_task_;
    304   if (task_number_limit_exceeded ||
    305       !dependency_manager_.Insert(blocking_factor.get())) {
    306     DCHECK(!running_background_tasks_.empty());
    307     DCHECK(pending_backgrounding_task_.is_null());
    308 
    309     // Wait for NotifyTaskDone to release a |blocking_factor|.
    310     pending_backgrounding_task_ =
    311         base::Bind(&SyncTaskManager::UpdateBlockingFactorBody,
    312                    AsWeakPtr(),
    313                    base::Passed(&foreground_task_token),
    314                    base::Passed(&background_task_token),
    315                    base::Passed(&task_log),
    316                    base::Passed(&blocking_factor),
    317                    continuation);
    318     return;
    319   }
    320 
    321   if (background_task_token) {
    322     background_task_token->set_blocking_factor(blocking_factor.Pass());
    323   } else {
    324     tracked_objects::Location from_here = foreground_task_token->location();
    325     SyncStatusCallback callback = foreground_task_token->callback();
    326     foreground_task_token->clear_callback();
    327 
    328     background_task_token =
    329         SyncTaskToken::CreateForBackgroundTask(
    330             AsWeakPtr(),
    331             task_token_seq_++,
    332             blocking_factor.Pass());
    333     background_task_token->UpdateTask(from_here, callback);
    334     running_background_tasks_.set(background_task_token->token_id(),
    335                                   running_foreground_task_.Pass());
    336   }
    337 
    338   token_ = foreground_task_token.Pass();
    339   MaybeStartNextForegroundTask(scoped_ptr<SyncTaskToken>());
    340   background_task_token->SetTaskLog(task_log.Pass());
    341   continuation.Run(background_task_token.Pass());
    342 }
    343 
    344 scoped_ptr<SyncTaskToken> SyncTaskManager::GetToken(
    345     const tracked_objects::Location& from_here,
    346     const SyncStatusCallback& callback) {
    347   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    348 
    349   if (!token_)
    350     return scoped_ptr<SyncTaskToken>();
    351   token_->UpdateTask(from_here, callback);
    352   return token_.Pass();
    353 }
    354 
    355 void SyncTaskManager::PushPendingTask(
    356     const base::Closure& closure, Priority priority) {
    357   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    358 
    359   pending_tasks_.push(PendingTask(closure, priority, pending_task_seq_++));
    360 }
    361 
    362 void SyncTaskManager::RunTask(scoped_ptr<SyncTaskToken> token,
    363                               scoped_ptr<SyncTask> task) {
    364   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    365   DCHECK(!running_foreground_task_);
    366 
    367   running_foreground_task_ = task.Pass();
    368   running_foreground_task_->RunPreflight(token.Pass());
    369 }
    370 
    371 void SyncTaskManager::MaybeStartNextForegroundTask(
    372     scoped_ptr<SyncTaskToken> token) {
    373   DCHECK(sequence_checker_.CalledOnValidSequencedThread());
    374 
    375   if (token) {
    376     DCHECK(!token_);
    377     token_ = token.Pass();
    378   }
    379 
    380   if (!pending_backgrounding_task_.is_null()) {
    381     base::Closure closure = pending_backgrounding_task_;
    382     pending_backgrounding_task_.Reset();
    383     closure.Run();
    384     return;
    385   }
    386 
    387   if (!token_)
    388     return;
    389 
    390   if (!pending_tasks_.empty()) {
    391     base::Closure closure = pending_tasks_.top().task;
    392     pending_tasks_.pop();
    393     closure.Run();
    394     return;
    395   }
    396 
    397   if (client_)
    398     client_->MaybeScheduleNextTask();
    399 }
    400 
    401 }  // namespace drive_backend
    402 }  // namespace sync_file_system
    403