prioritized_dispatcher.cc 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148
  1. // Copyright (c) 2012 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 "net/base/prioritized_dispatcher.h"
  5. #include <ostream>
  6. #include "base/check_op.h"
  7. namespace net {
  8. PrioritizedDispatcher::Limits::Limits(Priority num_priorities,
  9. size_t total_jobs)
  10. : total_jobs(total_jobs), reserved_slots(num_priorities) {}
  11. PrioritizedDispatcher::Limits::Limits(const Limits& other) = default;
  12. PrioritizedDispatcher::Limits::~Limits() = default;
  13. PrioritizedDispatcher::PrioritizedDispatcher(const Limits& limits)
  14. : queue_(limits.reserved_slots.size()),
  15. max_running_jobs_(limits.reserved_slots.size()) {
  16. SetLimits(limits);
  17. }
  18. PrioritizedDispatcher::~PrioritizedDispatcher() = default;
  19. PrioritizedDispatcher::Handle PrioritizedDispatcher::Add(
  20. Job* job, Priority priority) {
  21. DCHECK(job);
  22. DCHECK_LT(priority, num_priorities());
  23. if (num_running_jobs_ < max_running_jobs_[priority]) {
  24. ++num_running_jobs_;
  25. job->Start();
  26. return Handle();
  27. }
  28. return queue_.Insert(job, priority);
  29. }
  30. PrioritizedDispatcher::Handle PrioritizedDispatcher::AddAtHead(
  31. Job* job, Priority priority) {
  32. DCHECK(job);
  33. DCHECK_LT(priority, num_priorities());
  34. if (num_running_jobs_ < max_running_jobs_[priority]) {
  35. ++num_running_jobs_;
  36. job->Start();
  37. return Handle();
  38. }
  39. return queue_.InsertAtFront(job, priority);
  40. }
  41. void PrioritizedDispatcher::Cancel(const Handle& handle) {
  42. queue_.Erase(handle);
  43. }
  44. PrioritizedDispatcher::Job* PrioritizedDispatcher::EvictOldestLowest() {
  45. Handle handle = queue_.FirstMin();
  46. if (handle.is_null())
  47. return nullptr;
  48. Job* job = handle.value();
  49. Cancel(handle);
  50. return job;
  51. }
  52. PrioritizedDispatcher::Handle PrioritizedDispatcher::ChangePriority(
  53. const Handle& handle, Priority priority) {
  54. DCHECK(!handle.is_null());
  55. DCHECK_LT(priority, num_priorities());
  56. DCHECK_GE(num_running_jobs_, max_running_jobs_[handle.priority()]) <<
  57. "Job should not be in queue when limits permit it to start.";
  58. if (handle.priority() == priority)
  59. return handle;
  60. if (MaybeDispatchJob(handle, priority))
  61. return Handle();
  62. Job* job = handle.value();
  63. queue_.Erase(handle);
  64. return queue_.Insert(job, priority);
  65. }
  66. void PrioritizedDispatcher::OnJobFinished() {
  67. DCHECK_GT(num_running_jobs_, 0u);
  68. --num_running_jobs_;
  69. MaybeDispatchNextJob();
  70. }
  71. PrioritizedDispatcher::Limits PrioritizedDispatcher::GetLimits() const {
  72. size_t num_priorities = max_running_jobs_.size();
  73. Limits limits(num_priorities, max_running_jobs_.back());
  74. // Calculate the number of jobs reserved for each priority and higher. Leave
  75. // the number of jobs reserved for the lowest priority or higher as 0.
  76. for (size_t i = 1; i < num_priorities; ++i) {
  77. limits.reserved_slots[i] = max_running_jobs_[i] - max_running_jobs_[i - 1];
  78. }
  79. return limits;
  80. }
  81. void PrioritizedDispatcher::SetLimits(const Limits& limits) {
  82. DCHECK_EQ(queue_.num_priorities(), limits.reserved_slots.size());
  83. size_t total = 0;
  84. for (size_t i = 0; i < limits.reserved_slots.size(); ++i) {
  85. total += limits.reserved_slots[i];
  86. max_running_jobs_[i] = total;
  87. }
  88. // Unreserved slots are available for all priorities.
  89. DCHECK_LE(total, limits.total_jobs) << "sum(reserved_slots) <= total_jobs";
  90. size_t spare = limits.total_jobs - total;
  91. for (size_t i = limits.reserved_slots.size(); i > 0; --i) {
  92. max_running_jobs_[i - 1] += spare;
  93. }
  94. // Start pending jobs, if limits permit.
  95. while (true) {
  96. if (!MaybeDispatchNextJob())
  97. break;
  98. }
  99. }
  100. void PrioritizedDispatcher::SetLimitsToZero() {
  101. SetLimits(Limits(queue_.num_priorities(), 0));
  102. }
  103. bool PrioritizedDispatcher::MaybeDispatchJob(const Handle& handle,
  104. Priority job_priority) {
  105. DCHECK_LT(job_priority, num_priorities());
  106. if (num_running_jobs_ >= max_running_jobs_[job_priority])
  107. return false;
  108. Job* job = handle.value();
  109. queue_.Erase(handle);
  110. ++num_running_jobs_;
  111. job->Start();
  112. return true;
  113. }
  114. bool PrioritizedDispatcher::MaybeDispatchNextJob() {
  115. Handle handle = queue_.FirstMax();
  116. if (handle.is_null()) {
  117. DCHECK_EQ(0u, queue_.size());
  118. return false;
  119. }
  120. return MaybeDispatchJob(handle, handle.priority());
  121. }
  122. } // namespace net