net/base/prioritized_dispatcher.cc

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