chrome/browser/chromeos/file_system_provider/queue.cc

   1 // Copyright 2015 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 "base/bind.h"
   6 #include "base/location.h"
   7 #include "base/logging.h"
   8 #include "base/single_thread_task_runner.h"
   9 #include "base/thread_task_runner_handle.h"
  10 #include "chrome/browser/chromeos/file_system_provider/queue.h"
  11
  12 namespace chromeos {
  13 namespace file_system_provider {
  14
  15 Queue::Task::Task() : token(0) {
  16 }
  17
  18 Queue::Task::Task(size_t token, const AbortableCallback& callback)
  19     : token(token), callback(callback) {
  20 }
  21
  22 Queue::Task::~Task() {
  23 }
  24
  25 Queue::Queue(size_t max_in_parallel)
  26     : max_in_parallel_(max_in_parallel),
  27       next_token_(1),
  28       weak_ptr_factory_(this) {
  29   CHECK_LT(0u, max_in_parallel);
  30 }
  31
  32 Queue::~Queue() {
  33 }
  34
  35 size_t Queue::NewToken() {
  36   return next_token_++;
  37 }
  38
  39 void Queue::Enqueue(size_t token, const AbortableCallback& callback) {
  40 #if !NDEBUG
  41   CHECK(executed_.find(token) == executed_.end());
  42   for (auto& task : pending_) {
  43     CHECK(token != task.token);
  44   }
  45 #endif
  46   pending_.push_back(Task(token, callback));
  47   base::ThreadTaskRunnerHandle::Get()->PostTask(
  48       FROM_HERE, base::Bind(&Queue::MaybeRun, weak_ptr_factory_.GetWeakPtr()));
  49 }
  50
  51 void Queue::Complete(size_t token) {
  52   const auto it = executed_.find(token);
  53   CHECK(it != executed_.end());
  54   completed_[token] = it->second;
  55   executed_.erase(it);
  56 }
  57
  58 void Queue::Remove(size_t token) {
  59   const auto it = completed_.find(token);
  60   if (it != completed_.end()) {
  61     completed_.erase(it);
  62     base::ThreadTaskRunnerHandle::Get()->PostTask(
  63         FROM_HERE,
  64         base::Bind(&Queue::MaybeRun, weak_ptr_factory_.GetWeakPtr()));
  65     return;
  66   }
  67
  68   // If not completed, then it must have been aborted.
  69   const auto aborted_it = aborted_.find(token);
  70   CHECK(aborted_it != aborted_.end());
  71   aborted_.erase(aborted_it);
  72
  73   base::ThreadTaskRunnerHandle::Get()->PostTask(
  74       FROM_HERE, base::Bind(&Queue::MaybeRun, weak_ptr_factory_.GetWeakPtr()));
  75 }
  76
  77 void Queue::MaybeRun() {
  78   if (executed_.size() + completed_.size() == max_in_parallel_ ||
  79       !pending_.size()) {
  80     return;
  81   }
  82
  83   CHECK_GT(max_in_parallel_, executed_.size() + completed_.size());
  84   Task task = pending_.front();
  85   pending_.pop_front();
  86
  87   executed_[task.token] = task;
  88   AbortCallback abort_callback = task.callback.Run();
  89
  90   // It may happen that the task is completed and removed synchronously. Hence,
  91   // we need to check if the task is still in the executed collection.
  92   const auto executed_task_it = executed_.find(task.token);
  93   if (executed_task_it != executed_.end())
  94     executed_task_it->second.abort_callback = abort_callback;
  95 }
  96
  97 void Queue::Abort(size_t token) {
  98   // Check if it's running.
  99   const auto it = executed_.find(token);
 100   if (it != executed_.end()) {
 101     Task task = it->second;
 102     aborted_[token] = task;
 103     executed_.erase(it);
 104     CHECK(!task.abort_callback.is_null());
 105     task.abort_callback.Run();
 106     base::ThreadTaskRunnerHandle::Get()->PostTask(
 107         FROM_HERE,
 108         base::Bind(&Queue::MaybeRun, weak_ptr_factory_.GetWeakPtr()));
 109     return;
 110   }
 111
 112   // Aborting not running tasks is linear. TODO(mtomasz): Optimize if feasible.
 113   for (auto it = pending_.begin(); it != pending_.end(); ++it) {
 114     if (token == it->token) {
 115       aborted_[token] = *it;
 116       pending_.erase(it);
 117       base::ThreadTaskRunnerHandle::Get()->PostTask(
 118           FROM_HERE,
 119           base::Bind(&Queue::MaybeRun, weak_ptr_factory_.GetWeakPtr()));
 120       return;
 121     }
 122   }
 123
 124   // The task is already removed, marked as completed or aborted.
 125   NOTREACHED();
 126 }
 127
 128 bool Queue::IsAborted(size_t token) {
 129 #if !NDEBUG
 130   bool in_queue = executed_.find(token) != executed_.end() ||
 131                   completed_.find(token) != completed_.end() ||
 132                   aborted_.find(token) != aborted_.end();
 133   for (auto& task : pending_) {
 134     if (token == task.token) {
 135       in_queue = true;
 136       break;
 137     }
 138   }
 139   DCHECK(in_queue);
 140 #endif
 141   return aborted_.find(token) != aborted_.end();
 142 }
 143
 144 }  // namespace file_system_provider
 145 }  // namespace chromeos