--- /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 WP_QUEUE_H
+#define WP_QUEUE_H
+
+#include "OpQueue.h"
+
+#include <boost/intrusive/list.hpp>
+#include <boost/intrusive/rbtree.hpp>
+#include <boost/intrusive/avl_set.hpp>
+
+namespace bi = boost::intrusive;
+
+template <typename T, typename S>
+class MapKey
+{
+ public:
+ bool operator()(const S i, const T &k) const
+ {
+ return i < k.key;
+ }
+ bool operator()(const T &k, const S i) const
+ {
+ return k.key < i;
+ }
+};
+
+template <typename T>
+class DelItem
+{
+ public:
+ void operator()(T* delete_this)
+ { delete delete_this; }
+};
+
+template <typename T, typename K>
+class WeightedPriorityQueue : public OpQueue <T, K>
+{
+ private:
+ class ListPair : public bi::list_base_hook<>
+ {
+ public:
+ unsigned cost;
+ T item;
+ ListPair(unsigned c, T& i) :
+ cost(c),
+ item(i)
+ {}
+ };
+ class Klass : public bi::set_base_hook<>
+ {
+ typedef bi::list<ListPair> ListPairs;
+ typedef typename ListPairs::iterator Lit;
+ public:
+ K key; // klass
+ ListPairs lp;
+ Klass(K& k) :
+ key(k)
+ {}
+ friend bool operator< (const Klass &a, const Klass &b)
+ { return a.key < b.key; }
+ friend bool operator> (const Klass &a, const Klass &b)
+ { return a.key > b.key; }
+ friend bool operator== (const Klass &a, const Klass &b)
+ { return a.key == b.key; }
+ void insert(unsigned cost, T& item, bool front) {
+ if (front) {
+ lp.push_front(*new ListPair(cost, item));
+ } else {
+ lp.push_back(*new ListPair(cost, item));
+ }
+ }
+ //Get the cost of the next item to dequeue
+ unsigned get_cost() const {
+ assert(!empty());
+ return lp.begin()->cost;
+ }
+ T pop() {
+ assert(!lp.empty());
+ T ret = lp.begin()->item;
+ lp.erase_and_dispose(lp.begin(), DelItem<ListPair>());
+ return ret;
+ }
+ bool empty() const {
+ return lp.empty();
+ }
+ unsigned get_size() const {
+ return lp.size();
+ }
+ unsigned filter_class(std::list<T>* out) {
+ unsigned count = 0;
+ for (Lit i = --lp.end();; --i) {
+ if (out) {
+ out->push_front(i->item);
+ }
+ i = lp.erase_and_dispose(i, DelItem<ListPair>());
+ ++count;
+ if (i == lp.begin()) {
+ break;
+ }
+ }
+ return count;
+ }
+ };
+ class SubQueue : public bi::set_base_hook<>
+ {
+ typedef bi::rbtree<Klass> Klasses;
+ typedef typename Klasses::iterator Kit;
+ void check_end() {
+ if (next == klasses.end()) {
+ next = klasses.begin();
+ }
+ }
+ public:
+ unsigned key; // priority
+ Klasses klasses;
+ Kit next;
+ SubQueue(unsigned& p) :
+ key(p),
+ next(klasses.begin())
+ {}
+ friend bool operator< (const SubQueue &a, const SubQueue &b)
+ { return a.key < b.key; }
+ friend bool operator> (const SubQueue &a, const SubQueue &b)
+ { return a.key > b.key; }
+ friend bool operator== (const SubQueue &a, const SubQueue &b)
+ { return a.key == b.key; }
+ bool empty() const {
+ return klasses.empty();
+ }
+ void insert(K cl, unsigned cost, T& item, bool front = false) {
+ typename Klasses::insert_commit_data insert_data;
+ std::pair<Kit, bool> ret =
+ klasses.insert_unique_check(cl, MapKey<Klass, K>(), insert_data);
+ if (ret.second) {
+ ret.first = klasses.insert_unique_commit(*new Klass(cl), insert_data);
+ check_end();
+ }
+ ret.first->insert(cost, item, front);
+ }
+ unsigned get_cost() const {
+ assert(!empty());
+ return next->get_cost();
+ }
+ T pop() {
+ T ret = next->pop();
+ if (next->empty()) {
+ next = klasses.erase_and_dispose(next, DelItem<Klass>());
+ } else {
+ ++next;
+ }
+ check_end();
+ return ret;
+ }
+ unsigned filter_class(K& cl, std::list<T>* out) {
+ unsigned count = 0;
+ Kit i = klasses.find(cl, MapKey<Klass, K>());
+ if (i != klasses.end()) {
+ count = i->filter_class(out);
+ Kit tmp = klasses.erase_and_dispose(i, DelItem<Klass>());
+ if (next == i) {
+ next = tmp;
+ }
+ check_end();
+ }
+ return count;
+ }
+ void dump(ceph::Formatter *f) const {
+ f->dump_int("num_keys", next->get_size());
+ if (!empty()) {
+ f->dump_int("first_item_cost", next->get_cost());
+ }
+ }
+ };
+ class Queue {
+ typedef bi::rbtree<SubQueue> SubQueues;
+ typedef typename SubQueues::iterator Sit;
+ SubQueues queues;
+ unsigned total_prio;
+ unsigned max_cost;
+ public:
+ unsigned size;
+ Queue() :
+ total_prio(0),
+ max_cost(0),
+ size(0)
+ {}
+ bool empty() const {
+ return !size;
+ }
+ void insert(unsigned p, K cl, unsigned cost, T& item, bool front = false) {
+ typename SubQueues::insert_commit_data insert_data;
+ std::pair<typename SubQueues::iterator, bool> ret =
+ queues.insert_unique_check(p, MapKey<SubQueue, unsigned>(), insert_data);
+ if (ret.second) {
+ ret.first = queues.insert_unique_commit(*new SubQueue(p), insert_data);
+ total_prio += p;
+ }
+ ret.first->insert(cl, cost, item, front);
+ if (cost > max_cost) {
+ max_cost = cost;
+ }
+ ++size;
+ }
+ T pop(bool strict = false) {
+ --size;
+ Sit i = --queues.end();
+ if (strict) {
+ T ret = i->pop();
+ if (i->empty()) {
+ queues.erase_and_dispose(i, DelItem<SubQueue>());
+ }
+ return ret;
+ }
+ if (queues.size() > 1) {
+ while (true) {
+ // Pick a new priority out of the total priority.
+ unsigned prio = rand() % total_prio + 1;
+ unsigned tp = total_prio - i->key;
+ // Find the priority coresponding to the picked number.
+ // Subtract high priorities to low priorities until the picked number
+ // is more than the total and try to dequeue that priority.
+ // Reverse the direction from previous implementation because there is a higher
+ // chance of dequeuing a high priority op so spend less time spinning.
+ while (prio <= tp) {
+ --i;
+ tp -= i->key;
+ }
+ // Flip a coin to see if this priority gets to run based on cost.
+ // The next op's cost is multiplied by .9 and subtracted from the
+ // max cost seen. Ops with lower costs will have a larger value
+ // and allow them to be selected easier than ops with high costs.
+ if (max_cost == 0 || rand() % max_cost <=
+ (max_cost - ((i->get_cost() * 9) / 10))) {
+ break;
+ }
+ i = --queues.end();
+ }
+ }
+ T ret = i->pop();
+ if (i->empty()) {
+ total_prio -= i->key;
+ queues.erase_and_dispose(i, DelItem<SubQueue>());
+ }
+ return ret;
+ }
+ void filter_class(K& cl, std::list<T>* out) {
+ for (Sit i = queues.begin(); i != queues.end();) {
+ size -= i->filter_class(cl, out);
+ if (i->empty()) {
+ total_prio -= i->key;
+ i = queues.erase_and_dispose(i, DelItem<SubQueue>());
+ } else {
+ ++i;
+ }
+ }
+ }
+ void dump(ceph::Formatter *f) const {
+ for (typename SubQueues::const_iterator i = queues.begin();
+ i != queues.end(); ++i) {
+ f->dump_int("total_priority", total_prio);
+ f->dump_int("max_cost", max_cost);
+ f->open_object_section("subqueue");
+ f->dump_int("priority", i->key);
+ i->dump(f);
+ f->close_section();
+ }
+ }
+ };
+
+ Queue strict;
+ Queue normal;
+ public:
+ WeightedPriorityQueue(unsigned max_per, unsigned min_c) :
+ strict(),
+ normal()
+ {
+ std::srand(time(0));
+ }
+ unsigned length() const final {
+ return strict.size + normal.size;
+ }
+ void remove_by_class(K cl, std::list<T>* removed = 0) final {
+ strict.filter_class(cl, removed);
+ normal.filter_class(cl, removed);
+ }
+ bool empty() const final {
+ return !(strict.size + normal.size);
+ }
+ void enqueue_strict(K cl, unsigned p, T item) final {
+ strict.insert(p, cl, 0, item);
+ }
+ void enqueue_strict_front(K cl, unsigned p, T item) final {
+ strict.insert(p, cl, 0, item, true);
+ }
+ void enqueue(K cl, unsigned p, unsigned cost, T item) final {
+ normal.insert(p, cl, cost, item);
+ }
+ void enqueue_front(K cl, unsigned p, unsigned cost, T item) final {
+ normal.insert(p, cl, cost, item, true);
+ }
+ T dequeue() override {
+ assert(strict.size + normal.size > 0);
+ if (!strict.empty()) {
+ return strict.pop(true);
+ }
+ return normal.pop();
+ }
+ void dump(ceph::Formatter *f) const override {
+ f->open_array_section("high_queues");
+ strict.dump(f);
+ f->close_section();
+ f->open_array_section("queues");
+ normal.dump(f);
+ f->close_section();
+ }
+};
+
+#endif
Code Review / stor4nfv.git / blobdiff