+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-/*
- * Ceph - scalable distributed file system
- *
- * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
- *
- * This is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License version 2.1, as published by the Free Software
- * Foundation. See file COPYING.
- *
- */
-
-#ifndef PRIORITY_QUEUE_H
-#define PRIORITY_QUEUE_H
-
-#include "common/Formatter.h"
-#include "common/OpQueue.h"
-
-/**
- * Manages queue for normal and strict priority items
- *
- * On dequeue, the queue will select the lowest priority queue
- * such that the q has bucket > cost of front queue item.
- *
- * If there is no such queue, we choose the next queue item for
- * the highest priority queue.
- *
- * Before returning a dequeued item, we place into each bucket
- * cost * (priority/total_priority) tokens.
- *
- * enqueue_strict and enqueue_strict_front queue items into queues
- * which are serviced in strict priority order before items queued
- * with enqueue and enqueue_front
- *
- * Within a priority class, we schedule round robin based on the class
- * of type K used to enqueue items. e.g. you could use entity_inst_t
- * to provide fairness for different clients.
- */
-template <typename T, typename K>
-class PrioritizedQueue : public OpQueue <T, K> {
- int64_t total_priority;
- int64_t max_tokens_per_subqueue;
- int64_t min_cost;
-
- typedef std::list<std::pair<unsigned, T> > ListPairs;
-
- struct SubQueue {
- private:
- typedef std::map<K, ListPairs> Classes;
- Classes q;
- unsigned tokens, max_tokens;
- int64_t size;
- typename Classes::iterator cur;
- public:
- SubQueue(const SubQueue &other)
- : q(other.q),
- tokens(other.tokens),
- max_tokens(other.max_tokens),
- size(other.size),
- cur(q.begin()) {}
- SubQueue()
- : tokens(0),
- max_tokens(0),
- size(0), cur(q.begin()) {}
- void set_max_tokens(unsigned mt) {
- max_tokens = mt;
- }
- unsigned get_max_tokens() const {
- return max_tokens;
- }
- unsigned num_tokens() const {
- return tokens;
- }
- void put_tokens(unsigned t) {
- tokens += t;
- if (tokens > max_tokens) {
- tokens = max_tokens;
- }
- }
- void take_tokens(unsigned t) {
- if (tokens > t) {
- tokens -= t;
- } else {
- tokens = 0;
- }
- }
- void enqueue(K cl, unsigned cost, T item) {
- q[cl].push_back(std::make_pair(cost, item));
- if (cur == q.end())
- cur = q.begin();
- size++;
- }
- void enqueue_front(K cl, unsigned cost, T item) {
- q[cl].push_front(std::make_pair(cost, item));
- if (cur == q.end())
- cur = q.begin();
- size++;
- }
- std::pair<unsigned, T> front() const {
- assert(!(q.empty()));
- assert(cur != q.end());
- return cur->second.front();
- }
- void pop_front() {
- assert(!(q.empty()));
- assert(cur != q.end());
- cur->second.pop_front();
- if (cur->second.empty()) {
- q.erase(cur++);
- } else {
- ++cur;
- }
- if (cur == q.end()) {
- cur = q.begin();
- }
- size--;
- }
- unsigned length() const {
- assert(size >= 0);
- return (unsigned)size;
- }
- bool empty() const {
- return q.empty();
- }
- void remove_by_class(K k, std::list<T> *out) {
- typename Classes::iterator i = q.find(k);
- if (i == q.end()) {
- return;
- }
- size -= i->second.size();
- if (i == cur) {
- ++cur;
- }
- if (out) {
- for (typename ListPairs::reverse_iterator j =
- i->second.rbegin();
- j != i->second.rend();
- ++j) {
- out->push_front(j->second);
- }
- }
- q.erase(i);
- if (cur == q.end()) {
- cur = q.begin();
- }
- }
-
- void dump(ceph::Formatter *f) const {
- f->dump_int("tokens", tokens);
- f->dump_int("max_tokens", max_tokens);
- f->dump_int("size", size);
- f->dump_int("num_keys", q.size());
- if (!empty()) {
- f->dump_int("first_item_cost", front().first);
- }
- }
- };
-
- typedef std::map<unsigned, SubQueue> SubQueues;
- SubQueues high_queue;
- SubQueues queue;
-
- SubQueue *create_queue(unsigned priority) {
- typename SubQueues::iterator p = queue.find(priority);
- if (p != queue.end()) {
- return &p->second;
- }
- total_priority += priority;
- SubQueue *sq = &queue[priority];
- sq->set_max_tokens(max_tokens_per_subqueue);
- return sq;
- }
-
- void remove_queue(unsigned priority) {
- assert(queue.count(priority));
- queue.erase(priority);
- total_priority -= priority;
- assert(total_priority >= 0);
- }
-
- void distribute_tokens(unsigned cost) {
- if (total_priority == 0) {
- return;
- }
- for (typename SubQueues::iterator i = queue.begin();
- i != queue.end();
- ++i) {
- i->second.put_tokens(((i->first * cost) / total_priority) + 1);
- }
- }
-
-public:
- PrioritizedQueue(unsigned max_per, unsigned min_c)
- : total_priority(0),
- max_tokens_per_subqueue(max_per),
- min_cost(min_c)
- {}
-
- unsigned length() const final {
- unsigned total = 0;
- for (typename SubQueues::const_iterator i = queue.begin();
- i != queue.end();
- ++i) {
- assert(i->second.length());
- total += i->second.length();
- }
- for (typename SubQueues::const_iterator i = high_queue.begin();
- i != high_queue.end();
- ++i) {
- assert(i->second.length());
- total += i->second.length();
- }
- return total;
- }
-
- void remove_by_class(K k, std::list<T> *out = 0) final {
- for (typename SubQueues::iterator i = queue.begin();
- i != queue.end();
- ) {
- i->second.remove_by_class(k, out);
- if (i->second.empty()) {
- unsigned priority = i->first;
- ++i;
- remove_queue(priority);
- } else {
- ++i;
- }
- }
- for (typename SubQueues::iterator i = high_queue.begin();
- i != high_queue.end();
- ) {
- i->second.remove_by_class(k, out);
- if (i->second.empty()) {
- high_queue.erase(i++);
- } else {
- ++i;
- }
- }
- }
-
- void enqueue_strict(K cl, unsigned priority, T item) final {
- high_queue[priority].enqueue(cl, 0, item);
- }
-
- void enqueue_strict_front(K cl, unsigned priority, T item) final {
- high_queue[priority].enqueue_front(cl, 0, item);
- }
-
- void enqueue(K cl, unsigned priority, unsigned cost, T item) final {
- if (cost < min_cost)
- cost = min_cost;
- if (cost > max_tokens_per_subqueue)
- cost = max_tokens_per_subqueue;
- create_queue(priority)->enqueue(cl, cost, item);
- }
-
- void enqueue_front(K cl, unsigned priority, unsigned cost, T item) final {
- if (cost < min_cost)
- cost = min_cost;
- if (cost > max_tokens_per_subqueue)
- cost = max_tokens_per_subqueue;
- create_queue(priority)->enqueue_front(cl, cost, item);
- }
-
- bool empty() const final {
- assert(total_priority >= 0);
- assert((total_priority == 0) || !(queue.empty()));
- return queue.empty() && high_queue.empty();
- }
-
- T dequeue() final {
- assert(!empty());
-
- if (!(high_queue.empty())) {
- T ret = high_queue.rbegin()->second.front().second;
- high_queue.rbegin()->second.pop_front();
- if (high_queue.rbegin()->second.empty()) {
- high_queue.erase(high_queue.rbegin()->first);
- }
- return ret;
- }
-
- // if there are multiple buckets/subqueues with sufficient tokens,
- // we behave like a strict priority queue among all subqueues that
- // are eligible to run.
- for (typename SubQueues::iterator i = queue.begin();
- i != queue.end();
- ++i) {
- assert(!(i->second.empty()));
- if (i->second.front().first < i->second.num_tokens()) {
- T ret = i->second.front().second;
- unsigned cost = i->second.front().first;
- i->second.take_tokens(cost);
- i->second.pop_front();
- if (i->second.empty()) {
- remove_queue(i->first);
- }
- distribute_tokens(cost);
- return ret;
- }
- }
-
- // if no subqueues have sufficient tokens, we behave like a strict
- // priority queue.
- T ret = queue.rbegin()->second.front().second;
- unsigned cost = queue.rbegin()->second.front().first;
- queue.rbegin()->second.pop_front();
- if (queue.rbegin()->second.empty()) {
- remove_queue(queue.rbegin()->first);
- }
- distribute_tokens(cost);
- return ret;
- }
-
- void dump(ceph::Formatter *f) const final {
- f->dump_int("total_priority", total_priority);
- f->dump_int("max_tokens_per_subqueue", max_tokens_per_subqueue);
- f->dump_int("min_cost", min_cost);
- f->open_array_section("high_queues");
- for (typename SubQueues::const_iterator p = high_queue.begin();
- p != high_queue.end();
- ++p) {
- f->open_object_section("subqueue");
- f->dump_int("priority", p->first);
- p->second.dump(f);
- f->close_section();
- }
- f->close_section();
- f->open_array_section("queues");
- for (typename SubQueues::const_iterator p = queue.begin();
- p != queue.end();
- ++p) {
- f->open_object_section("subqueue");
- f->dump_int("priority", p->first);
- p->second.dump(f);
- f->close_section();
- }
- f->close_section();
- }
-};
-
-#endif
Code Review / stor4nfv.git / blobdiff