task_queue.cc 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196
  1. // Copyright 2016 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. #include "components/offline_pages/task/task_queue.h"
  5. #include <utility>
  6. #include "base/bind.h"
  7. #include "base/logging.h"
  8. #include "base/memory/scoped_refptr.h"
  9. #include "base/memory/weak_ptr.h"
  10. #include "base/notreached.h"
  11. #include "base/task/single_thread_task_runner.h"
  12. #include "base/threading/thread_task_runner_handle.h"
  13. namespace offline_pages {
  14. struct TaskQueue::Entry {
  15. Entry() = default;
  16. explicit Entry(std::unique_ptr<Task> task) : task(std::move(task)) {}
  17. Entry(const base::Location& location, std::unique_ptr<Task> task)
  18. : task(std::move(task)), from_here(location) {}
  19. Entry(std::unique_ptr<Task> task, base::OnceClosure resume_callback)
  20. : task(std::move(task)), resume_callback(std::move(resume_callback)) {}
  21. std::unique_ptr<Task> task;
  22. base::OnceClosure resume_callback;
  23. base::Location from_here;
  24. };
  25. TaskQueue::TaskQueue(Delegate* delegate)
  26. : task_runner_(base::ThreadTaskRunnerHandle::Get()), delegate_(delegate) {
  27. DCHECK(delegate_);
  28. }
  29. TaskQueue::~TaskQueue() {}
  30. void TaskQueue::AddTask(std::unique_ptr<Task> task) {
  31. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  32. task->task_queue_ = this;
  33. tasks_.emplace_back(std::move(task));
  34. StartTaskIfAvailable();
  35. }
  36. void TaskQueue::AddTask(const base::Location& from_here,
  37. std::unique_ptr<Task> task) {
  38. DVLOG(2) << "Adding task " << from_here.ToString();
  39. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  40. task->task_queue_ = this;
  41. tasks_.emplace_back(from_here, std::move(task));
  42. StartTaskIfAvailable();
  43. }
  44. bool TaskQueue::HasPendingTasks() const {
  45. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  46. return !tasks_.empty() || HasRunningTask();
  47. }
  48. bool TaskQueue::HasRunningTask() const {
  49. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  50. return current_task_ != nullptr;
  51. }
  52. void TaskQueue::StartTaskIfAvailable() {
  53. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  54. DVLOG(2) << "running? " << HasRunningTask() << ", pending? "
  55. << HasPendingTasks() << " " << __func__;
  56. if (HasRunningTask())
  57. return;
  58. if (!HasPendingTasks()) {
  59. task_runner_->PostTask(FROM_HERE,
  60. base::BindOnce(&TaskQueue::InformTaskQueueIsIdle,
  61. weak_ptr_factory_.GetWeakPtr()));
  62. return;
  63. }
  64. current_task_ = std::move(tasks_.front().task);
  65. current_task_location_ = tasks_.front().from_here;
  66. base::OnceClosure resume_callback = std::move(tasks_.front().resume_callback);
  67. tasks_.pop_front();
  68. if (resume_callback) {
  69. task_runner_->PostTask(FROM_HERE,
  70. base::BindOnce(&TaskQueue::ResumeCurrentTask,
  71. weak_ptr_factory_.GetWeakPtr(),
  72. std::move(resume_callback)));
  73. } else {
  74. task_runner_->PostTask(FROM_HERE,
  75. base::BindOnce(&TaskQueue::RunCurrentTask,
  76. weak_ptr_factory_.GetWeakPtr()));
  77. }
  78. }
  79. void TaskQueue::RunCurrentTask() {
  80. DVLOG(2) << "Running task " << current_task_location_.ToString();
  81. current_task_->Execute(base::BindOnce(&TaskCompletedCallback, task_runner_,
  82. weak_ptr_factory_.GetWeakPtr(),
  83. current_task_.get()));
  84. }
  85. void TaskQueue::ResumeCurrentTask(base::OnceClosure on_resume) {
  86. DVLOG(2) << "Resuming task " << current_task_location_.ToString();
  87. DCHECK_EQ(Task::TaskState::kPendingResume, current_task_->state_);
  88. current_task_->state_ = Task::TaskState::kRunning;
  89. std::move(on_resume).Run();
  90. }
  91. // static
  92. void TaskQueue::TaskCompletedCallback(
  93. scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  94. base::WeakPtr<TaskQueue> task_queue,
  95. Task* task) {
  96. task_runner->PostTask(
  97. FROM_HERE, base::BindOnce(&TaskQueue::TaskCompleted, task_queue, task));
  98. }
  99. void TaskQueue::TaskCompleted(Task* task) {
  100. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  101. // Normally, the completed task is the current task.
  102. if (task == current_task_.get()) {
  103. current_task_.reset(nullptr);
  104. DVLOG(2) << "Current task completed " << current_task_location_.ToString();
  105. StartTaskIfAvailable();
  106. return;
  107. }
  108. // If the task is in the suspended_tasks_ list, remove it.
  109. for (auto iter = suspended_tasks_.begin(); iter != suspended_tasks_.end();
  110. ++iter) {
  111. if (iter->task.get() == task) {
  112. DVLOG(2) << "Suspended task completed " << iter->from_here.ToString();
  113. suspended_tasks_.erase(iter);
  114. return;
  115. }
  116. }
  117. NOTREACHED() << "TaskCompleted: cannot find task";
  118. }
  119. void TaskQueue::SuspendTask(Task* task) {
  120. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  121. // Task::Suspend() sets state to kSuspended.
  122. DCHECK_EQ(Task::TaskState::kSuspended, task->state_);
  123. DCHECK_EQ(task, current_task_.get());
  124. suspended_tasks_.emplace_back(current_task_location_,
  125. std::move(current_task_));
  126. StartTaskIfAvailable();
  127. }
  128. void TaskQueue::ResumeTask(Task* task, base::OnceClosure on_resume) {
  129. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  130. DCHECK_EQ(Task::TaskState::kSuspended, task->state_);
  131. for (auto iter = suspended_tasks_.begin(); iter != suspended_tasks_.end();
  132. ++iter) {
  133. if (iter->task.get() == task) {
  134. iter->resume_callback = std::move(on_resume);
  135. tasks_.push_back(std::move(*iter));
  136. suspended_tasks_.erase(iter);
  137. task->state_ = Task::TaskState::kPendingResume;
  138. StartTaskIfAvailable();
  139. return;
  140. }
  141. }
  142. NOTREACHED() << "Trying to resume task that's not suspended";
  143. }
  144. void TaskQueue::InformTaskQueueIsIdle() {
  145. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  146. delegate_->OnTaskQueueIsIdle();
  147. }
  148. // Returns a human-readable string describing the contents of the task queue.
  149. std::string TaskQueue::GetStateForTesting() const {
  150. std::stringstream ss;
  151. if (current_task_) {
  152. ss << "Current task: " << current_task_location_.ToString() << '\n';
  153. } else {
  154. ss << "No current task\n";
  155. }
  156. int number = 1;
  157. for (const auto& entry : tasks_) {
  158. ss << "Pending task " << number++ << ": " << entry.from_here.ToString()
  159. << '\n';
  160. }
  161. number = 1;
  162. for (const auto& entry : suspended_tasks_) {
  163. ss << "Suspended task " << number++ << ": " << entry.from_here.ToString()
  164. << '\n';
  165. }
  166. return ss.str();
  167. }
  168. } // namespace offline_pages