Adds PriorityQueue and PrioritizedDispatcher.

This will be used in HostResolverImpl to dispatch Jobs.

BUG=109426
TEST=./net_unittests --gtest_filter=PriorityQueue*:PrioritizedDispatcher*

Review URL: http://codereview.chromium.org/9113022

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@116736 0039d316-1c4b-4281-b951-d872f2087c98

6 files changed, 849 insertions, 1 deletions

diff --git a/net/base/prioritized_dispatcher.cc b/net/base/prioritized_dispatcher.cc
new file mode 100644
index 0000000..afa12f8
--- /dev/null
+++ b/net/base/prioritized_dispatcher.cc
@@ -0,0 +1,103 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "net/base/prioritized_dispatcher.h"
+
+#include "base/logging.h"
+
+namespace net {
+
+PrioritizedDispatcher::Limits::Limits(Priority num_priorities,
+                                      size_t total_jobs)
+    : total_jobs(total_jobs), reserved_slots(num_priorities) {}
+
+PrioritizedDispatcher::Limits::~Limits() {}
+
+PrioritizedDispatcher::PrioritizedDispatcher(const Limits& limits)
+    : queue_(limits.reserved_slots.size()),
+      max_running_jobs_(limits.reserved_slots.size()),
+      num_running_jobs_(0) {
+  size_t total = 0;
+  for (size_t i = limits.reserved_slots.size(); i > 0; --i) {
+    total += limits.reserved_slots[i - 1];
+    max_running_jobs_[i - 1] = total;
+  }
+  // Unreserved slots are available for all priorities.
+  DCHECK_LE(total, limits.total_jobs) << "sum(reserved_slots) <= total_jobs";
+  size_t spare = limits.total_jobs - total;
+  for (size_t i = 0; i < max_running_jobs_.size(); ++i) {
+    max_running_jobs_[i] += spare;
+  }
+}
+
+PrioritizedDispatcher::~PrioritizedDispatcher() {}
+
+PrioritizedDispatcher::Handle PrioritizedDispatcher::Add(
+    Job* job, Priority priority) {
+  DCHECK(job);
+  DCHECK_LT(priority, num_priorities());
+  if (num_running_jobs_ < max_running_jobs_[priority]) {
+    ++num_running_jobs_;
+    job->Start();
+    return Handle();
+  }
+  return queue_.Insert(job, priority);
+}
+
+void PrioritizedDispatcher::Cancel(const Handle& handle) {
+  queue_.Erase(handle);
+}
+
+PrioritizedDispatcher::Job* PrioritizedDispatcher::EvictOldestLowest() {
+  Handle handle = queue_.FirstMax();
+  if (handle.is_null())
+    return NULL;
+  Job* job = handle.value();
+  Cancel(handle);
+  return job;
+}
+
+PrioritizedDispatcher::Handle PrioritizedDispatcher::ChangePriority(
+    const Handle& handle, Priority priority) {
+  DCHECK(!handle.is_null());
+  DCHECK_LT(priority, num_priorities());
+  DCHECK_GE(num_running_jobs_, max_running_jobs_[handle.priority()]) <<
+      "Job should not be in queue when limits permit it to start.";
+
+  if (handle.priority() == priority)
+    return handle;
+
+  if (MaybeDispatchJob(handle, priority))
+    return Handle();
+  Job* job = handle.value();
+  queue_.Erase(handle);
+  return queue_.Insert(job, priority);
+}
+
+void PrioritizedDispatcher::OnJobFinished() {
+  DCHECK_GT(num_running_jobs_, 0u);
+  --num_running_jobs_;
+  Handle handle = queue_.FirstMin();
+  if (handle.is_null()) {
+    DCHECK_EQ(0u, queue_.size());
+    return;
+  }
+  MaybeDispatchJob(handle, handle.priority());
+}
+
+bool PrioritizedDispatcher::MaybeDispatchJob(const Handle& handle,
+                                             Priority job_priority) {
+  DCHECK_LT(job_priority, num_priorities());
+  if (num_running_jobs_ >= max_running_jobs_[job_priority])
+    return false;
+  Job* job = handle.value();
+  queue_.Erase(handle);
+  ++num_running_jobs_;
+  job->Start();
+  return true;
+}
+
+}  // namespace net
+
+
diff --git a/net/base/prioritized_dispatcher.h b/net/base/prioritized_dispatcher.h
new file mode 100644
index 0000000..c10061b
--- /dev/null
+++ b/net/base/prioritized_dispatcher.h
@@ -0,0 +1,115 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef NET_BASE_PRIORITY_DISPATCH_H_
+#define NET_BASE_PRIORITY_DISPATCH_H_
+#pragma once
+
+#include <vector>
+
+#include "net/base/net_export.h"
+#include "net/base/priority_queue.h"
+
+namespace net {
+
+// A priority-based dispatcher of jobs. Dispatch order is by priority (lowest
+// first) and then FIFO. The dispatcher enforces limits on the number of running
+// jobs. It never revokes a job once started. The job must call OnJobFinished
+// once it finishes in order to dispatch further jobs.
+//
+// All operations are O(p) time for p priority levels. The class is fully
+// reentrant: it is safe to execute any method (incl. destructor) from within
+// Job callbacks. However, this class is NOT thread-safe, which is enforced
+// by the underlying non-thread-safe PriorityQueue.
+//
+class NET_EXPORT_PRIVATE PrioritizedDispatcher {
+ public:
+  class Job;
+  typedef PriorityQueue<Job*>::Priority Priority;
+
+  // Describes the limits for the number of jobs started by the dispatcher.
+  // For example, |total_jobs| = 30 and |reserved_slots| = { 5, 10, 5 }
+  // allow for at most 30 running jobs in total. If there are already 24 jobs
+  // running, then there can be 6 more jobs started of which at most 1 can be
+  // at priority 1 or 0, but the rest has to be at 0.
+  struct NET_EXPORT_PRIVATE Limits {
+    Limits(Priority num_priorities, size_t total_jobs);
+    ~Limits();
+
+    // Total allowed running jobs.
+    size_t total_jobs;
+    // Number of slots reserved for each priority and higher.
+    // Sum of |reserved_slots| must be no greater than |total_jobs|.
+    std::vector<size_t> reserved_slots;
+  };
+
+  // An interface to the job dispatched by PrioritizedDispatcher. The dispatcher
+  // does not own the Job but expects it to live as long as the Job is queued.
+  // Use Cancel to remove Job from queue before it is dispatched. The Job can be
+  // deleted after it is dispatched or canceled, or the dispatcher is destroyed.
+  class Job {
+   public:
+    // Note: PriorityDispatch will never delete a Job.
+    virtual ~Job() {}
+    // Called when the dispatcher starts the job. Must call OnJobFinished when
+    // done.
+    virtual void Start() = 0;
+  };
+
+  // A handle to the enqueued job. The handle becomes invalid when the job is
+  // canceled, updated, or started.
+  typedef PriorityQueue<Job*>::Pointer Handle;
+
+  // Creates a dispatcher enforcing |limits| on number of running jobs.
+  PrioritizedDispatcher(const Limits& limits);
+
+  ~PrioritizedDispatcher();
+
+  size_t num_running_jobs() const { return num_running_jobs_; }
+  size_t num_queued_jobs() const { return queue_.size(); }
+  size_t num_priorities() const { return max_running_jobs_.size(); }
+
+  // Adds |job| with |priority| to the dispatcher. If limits permit, |job| is
+  // started immediately. Returns handle to the job or null-handle if the job is
+  // started.
+  Handle Add(Job* job, Priority priority);
+
+  // Removes the job with |handle| from the queue. Invalidates |handle|.
+  // Note: a Handle is valid iff the job is in the queue, i.e. has not Started.
+  void Cancel(const Handle& handle);
+
+  // Removes and returns the oldest-lowest Job from the queue invalidating any
+  // handles to it. Returns NULL if the queue is empty.
+  Job* EvictOldestLowest();
+
+  // Moves the queued job with |handle| to the end of all values with priority
+  // |priority| and returns the updated handle, or null-handle if it starts the
+  // job. Invalidates |handle|. No-op if priority did not change.
+  Handle ChangePriority(const Handle& handle, Priority priority);
+
+  // Notifies the dispatcher that a running job has finished. Could start a job.
+  void OnJobFinished();
+
+ private:
+  // Attempts to dispatch the job with |handle| at priority |priority| (might be
+  // different than |handle.priority()|. Returns true if successful. If so
+  // the |handle| becomes invalid.
+  bool MaybeDispatchJob(const Handle& handle, Priority priority);
+
+  // Queue for jobs that need to wait for a spare slot.
+  PriorityQueue<Job*> queue_;
+  // Maximum total number of running jobs allowed after a job at a particular
+  // priority is started. If a greater or equal number of jobs are running, then
+  // another job cannot be started.
+  std::vector<size_t> max_running_jobs_;
+  // Total number of running jobs.
+  size_t num_running_jobs_;
+
+  DISALLOW_COPY_AND_ASSIGN(PrioritizedDispatcher);
+};
+
+}  // namespace net
+
+#endif  // NET_BASE_PRIORITY_DISPATCH_H_
+
diff --git a/net/base/prioritized_dispatcher_unittest.cc b/net/base/prioritized_dispatcher_unittest.cc
new file mode 100644
index 0000000..ff32003
--- /dev/null
+++ b/net/base/prioritized_dispatcher_unittest.cc
@@ -0,0 +1,279 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include <ctype.h>
+#include <string>
+
+#include "base/compiler_specific.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/memory/scoped_vector.h"
+#include "net/base/prioritized_dispatcher.h"
+#include "net/base/request_priority.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace net {
+
+namespace {
+
+// We rely on the priority enum values being sequential having starting at 0,
+// and increasing for lower priorities.
+COMPILE_ASSERT(HIGHEST == 0u &&
+               LOWEST > HIGHEST &&
+               IDLE > LOWEST &&
+               NUM_PRIORITIES > IDLE,
+               priority_indexes_incompatible);
+
+class PrioritizedDispatcherTest : public testing::Test {
+ public:
+  typedef PrioritizedDispatcher::Priority Priority;
+  // A job that appends |data| to |log_| when started and '.' when finished.
+  // This is intended to confirm the execution order of a sequence of jobs added
+  // to the dispatcher.
+  class TestJob : public PrioritizedDispatcher::Job {
+   public:
+    TestJob(PrioritizedDispatcherTest* test, char data, Priority priority)
+        : test_(test), data_(data), priority_(priority), running_(false) {}
+
+    // MSVS does not accept EXPECT_EQ(this, ...) so wrap it up.
+    PrioritizedDispatcher::Job* this_job() {
+      return this;
+    }
+
+    void Add() {
+      EXPECT_TRUE(handle_.is_null());
+      EXPECT_FALSE(running_);
+      size_t num_queued = dispatch().num_queued_jobs();
+      size_t num_running = dispatch().num_running_jobs();
+
+      handle_ = dispatch().Add(this, priority_);
+
+      if (handle_.is_null()) {
+        EXPECT_EQ(num_queued, dispatch().num_queued_jobs());
+        EXPECT_TRUE(running_);
+        EXPECT_EQ(num_running + 1, dispatch().num_running_jobs());
+      } else {
+        EXPECT_FALSE(running_);
+        EXPECT_EQ(priority_, handle_.priority());
+        EXPECT_EQ(this_job(), handle_.value());
+        EXPECT_EQ(num_running, dispatch().num_running_jobs());
+      }
+    }
+
+    void ChangePriority(Priority priority) {
+      EXPECT_FALSE(running_);
+      ASSERT_FALSE(handle_.is_null());
+      size_t num_queued = dispatch().num_queued_jobs();
+      size_t num_running = dispatch().num_running_jobs();
+
+      handle_ = dispatch().ChangePriority(handle_, priority);
+
+      if (handle_.is_null()) {
+        EXPECT_TRUE(running_);
+        EXPECT_EQ(num_queued - 1, dispatch().num_queued_jobs());
+        EXPECT_EQ(num_running + 1, dispatch().num_running_jobs());
+      } else {
+        EXPECT_FALSE(running_);
+        EXPECT_EQ(priority, handle_.priority());
+        EXPECT_EQ(this_job(), handle_.value());
+        EXPECT_EQ(num_queued, dispatch().num_queued_jobs());
+        EXPECT_EQ(num_running, dispatch().num_running_jobs());
+      }
+    }
+
+    void Cancel() {
+      EXPECT_FALSE(running_);
+      ASSERT_FALSE(handle_.is_null());
+      size_t num_queued = dispatch().num_queued_jobs();
+
+      dispatch().Cancel(handle_);
+
+      EXPECT_EQ(num_queued - 1, dispatch().num_queued_jobs());
+      handle_ = PrioritizedDispatcher::Handle();
+    }
+
+    void Finish() {
+      EXPECT_TRUE(running_);
+      running_ = false;
+      test_->log_.append(1u, '.');
+
+      dispatch().OnJobFinished();
+    }
+
+    // PriorityDispatch::Job interface
+    virtual void Start() OVERRIDE {
+      EXPECT_FALSE(running_);
+      handle_ = PrioritizedDispatcher::Handle();
+      running_ = true;
+      test_->log_.append(1u, data_);
+    }
+
+   private:
+    PrioritizedDispatcher& dispatch() { return *(test_->dispatch_); }
+
+    PrioritizedDispatcherTest* test_;
+
+    char data_;
+    Priority priority_;
+
+    PrioritizedDispatcher::Handle handle_;
+    bool running_;
+  };
+
+ protected:
+  void Prepare(const PrioritizedDispatcher::Limits& limits) {
+    dispatch_.reset(new PrioritizedDispatcher(limits));
+  }
+
+  TestJob* AddJob(char data, Priority priority) {
+    TestJob* job = new TestJob(this, data, priority);
+    jobs_.push_back(job);
+    job->Add();
+    return job;
+  }
+
+  void Expect(std::string log) {
+    EXPECT_EQ(0u, dispatch_->num_queued_jobs());
+    EXPECT_EQ(0u, dispatch_->num_running_jobs());
+    EXPECT_EQ(log, log_);
+    log_.clear();
+  }
+
+  std::string log_;
+  scoped_ptr<PrioritizedDispatcher> dispatch_;
+  ScopedVector<TestJob> jobs_;
+};
+
+TEST_F(PrioritizedDispatcherTest, AddAFIFO) {
+  // Allow only one running job.
+  PrioritizedDispatcher::Limits limits(NUM_PRIORITIES, 1);
+  Prepare(limits);
+
+  TestJob* job_a = AddJob('a', IDLE);
+  TestJob* job_b = AddJob('b', IDLE);
+  TestJob* job_c = AddJob('c', IDLE);
+  TestJob* job_d = AddJob('d', IDLE);
+
+  job_a->Finish();
+  job_b->Finish();
+  job_c->Finish();
+  job_d->Finish();
+
+  Expect("a.b.c.d.");
+}
+
+TEST_F(PrioritizedDispatcherTest, AddPriority) {
+  PrioritizedDispatcher::Limits limits(NUM_PRIORITIES, 1);
+  Prepare(limits);
+
+  TestJob* job_a = AddJob('a', IDLE);
+  TestJob* job_b = AddJob('b', MEDIUM);
+  TestJob* job_c = AddJob('c', HIGHEST);
+  TestJob* job_d = AddJob('d', HIGHEST);
+  TestJob* job_e = AddJob('e', MEDIUM);
+
+  job_a->Finish();
+  job_c->Finish();
+  job_d->Finish();
+  job_b->Finish();
+  job_e->Finish();
+
+  Expect("a.c.d.b.e.");
+}
+
+TEST_F(PrioritizedDispatcherTest, EnforceLimits) {
+  // Reserve 2 for HIGHEST and 1 for LOW or higher.
+  // This leaves 2 for LOWEST or lower.
+  PrioritizedDispatcher::Limits limits(NUM_PRIORITIES, 5);
+  limits.reserved_slots[HIGHEST] = 2;
+  limits.reserved_slots[LOW] = 1;
+  Prepare(limits);
+
+  TestJob* job_a = AddJob('a', IDLE);    // Uses unreserved slot.
+  TestJob* job_b = AddJob('b', IDLE);    // Uses unreserved slot.
+  TestJob* job_c = AddJob('c', LOWEST);  // Must wait.
+  TestJob* job_d = AddJob('d', LOW);     // Uses reserved slot.
+  TestJob* job_e = AddJob('e', MEDIUM);  // Must wait.
+  TestJob* job_f = AddJob('f', HIGHEST); // Uses reserved slot.
+  TestJob* job_g = AddJob('g', HIGHEST); // Uses reserved slot.
+  TestJob* job_h = AddJob('h', HIGHEST); // Must wait.
+
+  EXPECT_EQ(5u, dispatch_->num_running_jobs());
+  EXPECT_EQ(3u, dispatch_->num_queued_jobs());
+
+  job_a->Finish();  // Releases h.
+  job_b->Finish();
+  job_d->Finish();
+  job_f->Finish();  // Releases e.
+  job_g->Finish();
+  job_h->Finish();  // Releases c.
+  job_e->Finish();
+  job_c->Finish();
+
+  Expect("abdfg.h...e..c..");
+}
+
+TEST_F(PrioritizedDispatcherTest, ChangePriority) {
+  PrioritizedDispatcher::Limits limits(NUM_PRIORITIES, 1);
+  Prepare(limits);
+
+  TestJob* job_a = AddJob('a', IDLE);
+  TestJob* job_b = AddJob('b', MEDIUM);
+  TestJob* job_c = AddJob('c', HIGHEST);
+  TestJob* job_d = AddJob('d', HIGHEST);
+
+  job_b->ChangePriority(HIGHEST);
+  job_c->ChangePriority(MEDIUM);
+
+  job_a->Finish();
+  job_d->Finish();
+  job_b->Finish();
+  job_c->Finish();
+
+  Expect("a.d.b.c.");
+}
+
+TEST_F(PrioritizedDispatcherTest, Cancel) {
+  PrioritizedDispatcher::Limits limits(NUM_PRIORITIES, 1);
+  Prepare(limits);
+
+  TestJob* job_a = AddJob('a', IDLE);
+  TestJob* job_b = AddJob('b', IDLE);
+  TestJob* job_c = AddJob('c', IDLE);
+  TestJob* job_d = AddJob('d', IDLE);
+  TestJob* job_e = AddJob('e', IDLE);
+
+  job_b->Cancel();
+  job_d->Cancel();
+
+  job_a->Finish();
+  job_c->Finish();
+  job_e->Finish();
+
+  Expect("a.c.e.");
+}
+
+TEST_F(PrioritizedDispatcherTest, Evict) {
+  PrioritizedDispatcher::Limits limits(NUM_PRIORITIES, 1);
+  Prepare(limits);
+
+  TestJob* job_a = AddJob('a', IDLE);
+  TestJob* job_b = AddJob('b', LOW);
+  TestJob* job_c = AddJob('c', HIGHEST);
+  TestJob* job_d = AddJob('d', LOW);
+  TestJob* job_e = AddJob('e', HIGHEST);
+
+  EXPECT_EQ(job_b, dispatch_->EvictOldestLowest());
+  EXPECT_EQ(job_d, dispatch_->EvictOldestLowest());
+
+  job_a->Finish();
+  job_c->Finish();
+  job_e->Finish();
+
+  Expect("a.c.e.");
+}
+
+}  // namespace
+
+}  // namespace net
+
diff --git a/net/base/priority_queue.h b/net/base/priority_queue.h
new file mode 100644
index 0000000..037faed
--- /dev/null
+++ b/net/base/priority_queue.h
@@ -0,0 +1,214 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef NET_BASE_PRIORITY_QUEUE_H_
+#define NET_BASE_PRIORITY_QUEUE_H_
+#pragma once
+
+#include <list>
+#include <vector>
+
+#include "base/basictypes.h"
+#include "base/logging.h"
+#include "base/threading/non_thread_safe.h"
+#include "net/base/net_export.h"
+
+#if !defined(NDEBUG)
+#include "base/hash_tables.h"
+#endif
+
+namespace net {
+
+// A simple priority queue. The order of values is by priority and then FIFO.
+// Unlike the std::priority_queue, this implementation allows erasing elements
+// from the queue, and all operations are O(p) time for p priority levels.
+// The queue is agnostic to priority ordering (whether 0 precedes 1).
+// If the highest priority is 0, FirstMin() returns the first in order.
+//
+// In debug-mode, the internal queues store (id, value) pairs where id is used
+// to validate Pointers.
+//
+template<typename T>
+class PriorityQueue : public base::NonThreadSafe {
+ private:
+  // This section is up-front for Pointer only.
+#if !defined(NDEBUG)
+  typedef std::list<std::pair<size_t, T> > List;
+#else
+  typedef std::list<T> List;
+#endif
+
+ public:
+  typedef uint32 Priority;
+
+  // A pointer to a value stored in the queue. The pointer becomes invalid
+  // when the queue is destroyed or cleared, or the value is erased.
+  class Pointer {
+   public:
+    // Constructs a null pointer.
+    Pointer() : priority_(kNullPriority) {}
+
+    bool is_null() const { return priority_ == kNullPriority; }
+
+    Priority priority() const { return priority_; }
+
+#if !defined(NDEBUG)
+    const T& value() const { return iterator_->second; }
+#else
+    const T& value() const { return *iterator_; }
+#endif
+
+    // Comparing to Pointer from a different PriorityQueue is undefined.
+    bool Equals(const Pointer& other) const {
+      return (priority_ == other.priority_) && (iterator_ == other.iterator_);
+    }
+
+   private:
+    friend class PriorityQueue;
+
+    // Note that we need iterator not const_iterator to pass to List::erase.
+    // When C++0x comes, this could be changed to const_iterator and const could
+    // be added to First, Last, and OldestLowest.
+    typedef typename PriorityQueue::List::iterator ListIterator;
+
+    static const Priority kNullPriority = static_cast<Priority>(-1);
+
+    Pointer(Priority priority, const ListIterator& iterator)
+        : priority_(priority), iterator_(iterator) {
+#if !defined(NDEBUG)
+      id_ = iterator_->first;
+#endif
+    }
+
+    Priority priority_;
+    ListIterator iterator_;
+
+#if !defined(NDEBUG)
+    // Used by the queue to check if a Pointer is valid.
+    size_t id_;
+#endif
+  };
+
+  // Creates a new queue for |num_priorities|.
+  explicit PriorityQueue(Priority num_priorities)
+      : lists_(num_priorities), size_(0) {
+#if !defined(NDEBUG)
+    next_id_ = 0;
+#endif
+  }
+
+  // Adds |value| with |priority| to the queue. Returns a pointer to the
+  // created element.
+  Pointer Insert(const T& value, Priority priority) {
+    DCHECK(CalledOnValidThread());
+    DCHECK_LT(priority, lists_.size());
+    ++size_;
+    List& list = lists_[priority];
+#if !defined(NDEBUG)
+    size_t id = next_id_;
+    valid_ids_.insert(id);
+    ++next_id_;
+    return Pointer(priority, list.insert(list.end(),
+                                         std::make_pair(id, value)));
+#else
+    return Pointer(priority, list.insert(list.end(), value));
+#endif
+  }
+
+  // Removes the value pointed by |pointer| from the queue. All pointers to this
+  // value including |pointer| become invalid.
+  void Erase(const Pointer& pointer) {
+    DCHECK(CalledOnValidThread());
+    DCHECK_LT(pointer.priority_, lists_.size());
+    DCHECK_GT(size_, 0u);
+
+#if !defined(NDEBUG)
+    DCHECK_EQ(1u, valid_ids_.erase(pointer.id_));
+    DCHECK_EQ(pointer.iterator_->first, pointer.id_);
+#endif
+
+    --size_;
+    lists_[pointer.priority_].erase(pointer.iterator_);
+  }
+
+  // Returns a pointer to the first value of minimum priority or a null-pointer
+  // if empty.
+  Pointer FirstMin() {
+    DCHECK(CalledOnValidThread());
+    for (size_t i = 0; i < lists_.size(); ++i) {
+      if (!lists_[i].empty())
+        return Pointer(i, lists_[i].begin());
+    }
+    return Pointer();
+  }
+
+  // Returns a pointer to the last value of minimum priority or a null-pointer
+  // if empty.
+  Pointer LastMin() {
+    DCHECK(CalledOnValidThread());
+    for (size_t i = 0; i < lists_.size(); ++i) {
+      if (!lists_[i].empty())
+        return Pointer(i, --lists_[i].end());
+    }
+    return Pointer();
+  }
+
+  // Returns a pointer to the first value of maximum priority or a null-pointer
+  // if empty.
+  Pointer FirstMax() {
+    DCHECK(CalledOnValidThread());
+    for (size_t i = lists_.size(); i > 0; --i) {
+      size_t index = i - 1;
+      if (!lists_[index].empty())
+        return Pointer(index, lists_[index].begin());
+    }
+    return Pointer();
+  }
+
+  // Returns a pointer to the last value of maximum priority or a null-pointer
+  // if empty.
+  Pointer LastMax() {
+    DCHECK(CalledOnValidThread());
+    for (size_t i = lists_.size(); i > 0; --i) {
+      size_t index = i - 1;
+      if (!lists_[index].empty())
+        return Pointer(index, --lists_[index].end());
+    }
+    return Pointer();
+  }
+
+  // Empties the queue. All pointers become invalid.
+  void Clear() {
+    DCHECK(CalledOnValidThread());
+    for (size_t i = 0; i < lists_.size(); ++i) {
+      lists_[i].clear();
+    }
+#if !defined(NDEBUG)
+    valid_ids_.clear();
+#endif
+    size_ = 0u;
+  }
+
+  // Returns number of queued values.
+  size_t size() const {
+    DCHECK(CalledOnValidThread());
+    return size_;
+  }
+
+ private:
+  typedef std::vector<List> ListVector;
+
+#if !defined(NDEBUG)
+  size_t next_id_;
+  base::hash_set<size_t> valid_ids_;
+#endif
+
+  ListVector lists_;
+  size_t size_;
+};
+
+}  // namespace net
+
+#endif  // NET_BASE_PRIORITY_QUEUE_H_
+
diff --git a/net/base/priority_queue_unittest.cc b/net/base/priority_queue_unittest.cc
new file mode 100644
index 0000000..f0f65a6
--- /dev/null
+++ b/net/base/priority_queue_unittest.cc
@@ -0,0 +1,132 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "net/base/priority_queue.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace net {
+
+namespace {
+
+typedef PriorityQueue<int>::Priority Priority;
+const Priority kPriorities[] = { 2, 1, 2, 0, 4, 3, 1, 4, 0 };
+const Priority kNumPriorities = 5;  // max(kPriorities) + 1
+const size_t kNumElements = arraysize(kPriorities);
+const int kFirstMinOrder[kNumElements] = { 3, 8, 1, 6, 0, 2, 5, 4, 7 };
+const int kLastMaxOrder[kNumElements] = { 7, 4, 5, 2, 0, 6, 1, 8, 3 };
+const int kFirstMaxOrder[kNumElements] = { 4, 7, 5, 0, 2, 1, 6, 3, 8 };
+const int kLastMinOrder[kNumElements] = { 8, 3, 6, 1, 2, 0, 5, 7, 4 };
+
+void CheckEmpty(PriorityQueue<int>* queue) {
+  EXPECT_EQ(0u, queue->size());
+  EXPECT_TRUE(queue->FirstMin().is_null());
+  EXPECT_TRUE(queue->LastMin().is_null());
+  EXPECT_TRUE(queue->FirstMax().is_null());
+  EXPECT_TRUE(queue->LastMax().is_null());
+}
+
+TEST(PriorityQueueTest, AddAndClear) {
+  PriorityQueue<int> queue(kNumPriorities);
+  PriorityQueue<int>::Pointer pointers[kNumElements];
+
+  CheckEmpty(&queue);
+  for (size_t i = 0; i < kNumElements; ++i) {
+    EXPECT_EQ(i, queue.size());
+    pointers[i] = queue.Insert(static_cast<int>(i), kPriorities[i]);
+  }
+  EXPECT_EQ(kNumElements, queue.size());
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    EXPECT_EQ(kPriorities[i], pointers[i].priority());
+    EXPECT_EQ(static_cast<int>(i), pointers[i].value());
+  }
+
+  queue.Clear();
+  CheckEmpty(&queue);
+}
+
+TEST(PriorityQueueTest, FirstMinOrder) {
+  PriorityQueue<int> queue(kNumPriorities);
+  PriorityQueue<int>::Pointer pointers[kNumElements];
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    pointers[i] = queue.Insert(static_cast<int>(i), kPriorities[i]);
+  }
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    EXPECT_EQ(kNumElements - i, queue.size());
+    // Also check Equals.
+    EXPECT_TRUE(queue.FirstMin().Equals(pointers[kFirstMinOrder[i]]));
+    EXPECT_EQ(kFirstMinOrder[i], queue.FirstMin().value());
+    queue.Erase(queue.FirstMin());
+  }
+  CheckEmpty(&queue);
+}
+
+TEST(PriorityQueueTest, LastMinOrder) {
+  PriorityQueue<int> queue(kNumPriorities);
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    queue.Insert(static_cast<int>(i), kPriorities[i]);
+  }
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    EXPECT_EQ(kLastMinOrder[i], queue.LastMin().value());
+    queue.Erase(queue.LastMin());
+  }
+  CheckEmpty(&queue);
+}
+
+TEST(PriorityQueueTest, FirstMaxOrder) {
+  PriorityQueue<int> queue(kNumPriorities);
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    queue.Insert(static_cast<int>(i), kPriorities[i]);
+  }
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    EXPECT_EQ(kFirstMaxOrder[i], queue.FirstMax().value());
+    queue.Erase(queue.FirstMax());
+  }
+  CheckEmpty(&queue);
+}
+
+TEST(PriorityQueueTest, LastMaxOrder) {
+  PriorityQueue<int> queue(kNumPriorities);
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    queue.Insert(static_cast<int>(i), kPriorities[i]);
+  }
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    EXPECT_EQ(kLastMaxOrder[i], queue.LastMax().value());
+    queue.Erase(queue.LastMax());
+  }
+  CheckEmpty(&queue);
+}
+
+TEST(PriorityQueueTest, EraseFromMiddle) {
+  PriorityQueue<int> queue(kNumPriorities);
+  PriorityQueue<int>::Pointer pointers[kNumElements];
+
+  for (size_t i = 0; i < kNumElements; ++i) {
+    pointers[i] = queue.Insert(static_cast<int>(i), kPriorities[i]);
+  }
+
+  queue.Erase(pointers[2]);
+  queue.Erase(pointers[3]);
+
+  int expected_order[] = { 8, 1, 6, 0, 5, 4, 7 };
+
+  for (size_t i = 0; i < arraysize(expected_order); ++i) {
+    EXPECT_EQ(expected_order[i], queue.FirstMin().value());
+    queue.Erase(queue.FirstMin());
+  }
+  CheckEmpty(&queue);
+}
+
+}  // namespace
+
+}  // namespace net
+
diff --git a/net/net.gyp b/net/net.gyp
index 78e57a0..4a249cf 100644
--- a/net/net.gyp
+++ b/net/net.gyp
@@ -1,4 +1,4 @@
-# Copyright (c) 2011 The Chromium Authors. All rights reserved.
+# Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
@@ -199,6 +199,9 @@
         'base/platform_mime_util_linux.cc',
         'base/platform_mime_util_mac.cc',
         'base/platform_mime_util_win.cc',
+        'base/prioritized_dispatcher.cc',
+        'base/prioritized_dispatcher.h',
+        'base/priority_queue.h',
         'base/rand_callback.h',
         'base/registry_controlled_domain.cc',
         'base/registry_controlled_domain.h',
@@ -1033,6 +1036,8 @@
         'base/network_change_notifier_win_unittest.cc',
         'base/origin_bound_cert_service_unittest.cc',
         'base/pem_tokenizer_unittest.cc',
+        'base/prioritized_dispatcher_unittest.cc',
+        'base/priority_queue_unittest.cc',
         'base/registry_controlled_domain_unittest.cc',
         'base/run_all_unittests.cc',
         'base/sdch_filter_unittest.cc',