Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 349 insertions, 0 deletions

diff --git a/src/common/WeightedPriorityQueue.h b/src/common/WeightedPriorityQueue.h
new file mode 100644
index 00000000..a05174e8
--- /dev/null
+++ b/src/common/WeightedPriorityQueue.h
@@ -0,0 +1,349 @@
+// -*- 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>
+
+#include "include/ceph_assert.h"
+
+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(std::move(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) {
+        }
+        ~Klass() {
+          lp.clear_and_dispose(DelItem<ListPair>());
+        }
+      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, std::move(item)));
+        } else {
+          lp.push_back(*new ListPair(cost, std::move(item)));
+        }
+      }
+      //Get the cost of the next item to dequeue
+      unsigned get_cost() const {
+        ceph_assert(!empty());
+        return lp.begin()->cost;
+      }
+      T pop() {
+	ceph_assert(!lp.empty());
+	T ret = std::move(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();
+      }
+      void filter_class(std::list<T>* out) {
+        for (Lit i = --lp.end();; --i) {
+          if (out) {
+            out->push_front(std::move(i->item));
+          }
+          i = lp.erase_and_dispose(i, DelItem<ListPair>());
+          if (i == lp.begin()) {
+            break;
+          }
+        }
+      }
+    };
+    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()) {
+	}
+	~SubQueue() {
+	  klasses.clear_and_dispose(DelItem<Klass>());
+	}
+      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, std::move(item), front);
+      }
+      unsigned get_cost() const {
+        ceph_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;
+      }
+      void filter_class(K& cl, std::list<T>* out) {
+        Kit i = klasses.find(cl, MapKey<Klass, K>());
+        if (i != klasses.end()) {
+          i->filter_class(out);
+	  Kit tmp = klasses.erase_and_dispose(i, DelItem<Klass>());
+	  if (next == i) {
+            next = tmp;
+          }
+          check_end();
+        }
+      }
+      // this is intended for unit tests and should be never used on hot paths
+      unsigned get_size_slow() const {
+	unsigned count = 0;
+	for (const auto& klass : klasses) {
+	  count += klass.get_size();
+	}
+	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:
+	Queue() :
+	  total_prio(0),
+	  max_cost(0) {
+	}
+	~Queue() {
+	  queues.clear_and_dispose(DelItem<SubQueue>());
+	}
+	bool empty() const {
+	  return queues.empty();
+	}
+	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, std::move(item), front);
+	  if (cost > max_cost) {
+	    max_cost = cost;
+	  }
+	}
+	T pop(bool strict = false) {
+	  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 corresponding 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();) {
+	    i->filter_class(cl, out);
+	    if (i->empty()) {
+	      total_prio -= i->key;
+	      i = queues.erase_and_dispose(i, DelItem<SubQueue>());
+	    } else {
+	      ++i;
+	    }
+	  }
+	}
+	// this is intended for unit tests and should be never used on hot paths
+	unsigned get_size_slow() const {
+	  unsigned count = 0;
+	  for (const auto& queue : queues) {
+	    count += queue.get_size_slow();
+	  }
+	  return count;
+	}
+	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));
+      }
+    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.empty() && normal.empty();
+    }
+    void enqueue_strict(K cl, unsigned p, T&& item) final {
+      strict.insert(p, cl, 0, std::move(item));
+    }
+    void enqueue_strict_front(K cl, unsigned p, T&& item) final {
+      strict.insert(p, cl, 0, std::move(item), true);
+    }
+    void enqueue(K cl, unsigned p, unsigned cost, T&& item) final {
+      normal.insert(p, cl, cost, std::move(item));
+    }
+    void enqueue_front(K cl, unsigned p, unsigned cost, T&& item) final {
+      normal.insert(p, cl, cost, std::move(item), true);
+    }
+    T dequeue() override {
+      ceph_assert(!empty());
+      if (!strict.empty()) {
+	return strict.pop(true);
+      }
+      return normal.pop();
+    }
+    unsigned get_size_slow() {
+      return strict.get_size_slow() + normal.get_size_slow();
+    }
+    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