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   DCHECK(it != executed_.end());
  54   executed_.erase(it);
  55   base::ThreadTaskRunnerHandle::Get()->PostTask(
  56       FROM_HERE, base::Bind(&Queue::MaybeRun, weak_ptr_factory_.GetWeakPtr()));
  57 }
  58
  59 void Queue::MaybeRun() {
  60   if (executed_.size() == max_in_parallel_ || !pending_.size())
  61     return;
  62
  63   CHECK_GT(max_in_parallel_, executed_.size());
  64   Task task = pending_.front();
  65   pending_.pop_front();
  66
  67   executed_[task.token] = task;
  68   AbortCallback abort_callback = task.callback.Run();
  69
  70   // It may happen that the task is completed and removed synchronously. Hence,
  71   // we need to check if the task is still in the executed collection.
  72   const auto executed_task_it = executed_.find(task.token);
  73   if (executed_task_it != executed_.end())
  74     executed_task_it->second.abort_callback = abort_callback;
  75 }
  76
  77 void Queue::Abort(size_t token) {
  78   // Check if it's running. If so, then abort and expect a Complete() call soon.
  79   const auto it = executed_.find(token);
  80   if (it != executed_.end()) {
  81     Task& task = it->second;
  82     AbortCallback abort_callback = task.abort_callback;
  83     task.abort_callback = AbortCallback();
  84     DCHECK(!abort_callback.is_null());
  85     abort_callback.Run();
  86     return;
  87   }
  88
  89   // Aborting not running tasks is linear. TODO(mtomasz): Optimize if feasible.
  90   for (auto it = pending_.begin(); it != pending_.end(); ++it) {
  91     if (token == it->token) {
  92       pending_.erase(it);
  93       base::ThreadTaskRunnerHandle::Get()->PostTask(
  94           FROM_HERE,
  95           base::Bind(&Queue::MaybeRun, weak_ptr_factory_.GetWeakPtr()));
  96       return;
  97     }
  98   }
  99
 100   // The task is already removed, marked as completed or aborted.
 101   NOTREACHED();
 102 }
 103
 104 }  // namespace file_system_provider
 105 }  // namespace chromeos