src/ceph/src/common/mClockPriorityQueue.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2016 Red Hat Inc.
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #pragma once
  16
  17
  18 #include <functional>
  19 #include <map>
  20 #include <list>
  21 #include <cmath>
  22
  23 #include "common/Formatter.h"
  24 #include "common/OpQueue.h"
  25
  26 #include "dmclock/src/dmclock_server.h"
  27
  28 // the following is done to unclobber _ASSERT_H so it returns to the
  29 // way ceph likes it
  30 #include "include/assert.h"
  31
  32
  33 namespace ceph {
  34
  35   namespace dmc = crimson::dmclock;
  36
  37   template <typename T, typename K>
  38   class mClockQueue : public OpQueue <T, K> {
  39
  40     using priority_t = unsigned;
  41     using cost_t = unsigned;
  42
  43     typedef std::list<std::pair<cost_t, T> > ListPairs;
  44
  45     static unsigned filter_list_pairs(ListPairs *l,
  46                                       std::function<bool (const T&)> f,
  47                                       std::list<T>* out = nullptr) {
  48       unsigned ret = 0;
  49       for (typename ListPairs::iterator i = l->end();
  50            i != l->begin();
  51            /* no inc */
  52         ) {
  53         auto next = i;
  54         --next;
  55         if (f(next->second)) {
  56           ++ret;
  57           if (out) out->push_back(next->second);
  58           l->erase(next);
  59         } else {
  60           i = next;
  61         }
  62       }
  63       return ret;
  64     }
  65
  66     struct SubQueue {
  67     private:
  68       typedef std::map<K, ListPairs> Classes;
  69       // client-class to ordered queue
  70       Classes q;
  71
  72       unsigned tokens, max_tokens;
  73       int64_t size;
  74
  75       typename Classes::iterator cur;
  76
  77     public:
  78
  79       SubQueue(const SubQueue &other)
  80         : q(other.q),
  81           tokens(other.tokens),
  82           max_tokens(other.max_tokens),
  83           size(other.size),
  84           cur(q.begin()) {}
  85
  86       SubQueue()
  87         : tokens(0),
  88           max_tokens(0),
  89           size(0), cur(q.begin()) {}
  90
  91       void set_max_tokens(unsigned mt) {
  92         max_tokens = mt;
  93       }
  94
  95       unsigned get_max_tokens() const {
  96         return max_tokens;
  97       }
  98
  99       unsigned num_tokens() const {
 100         return tokens;
 101       }
 102
 103       void put_tokens(unsigned t) {
 104         tokens += t;
 105         if (tokens > max_tokens) {
 106           tokens = max_tokens;
 107         }
 108       }
 109
 110       void take_tokens(unsigned t) {
 111         if (tokens > t) {
 112           tokens -= t;
 113         } else {
 114           tokens = 0;
 115         }
 116       }
 117
 118       void enqueue(K cl, cost_t cost, T item) {
 119         q[cl].push_back(std::make_pair(cost, item));
 120         if (cur == q.end())
 121           cur = q.begin();
 122         size++;
 123       }
 124
 125       void enqueue_front(K cl, cost_t cost, T item) {
 126         q[cl].push_front(std::make_pair(cost, item));
 127         if (cur == q.end())
 128           cur = q.begin();
 129         size++;
 130       }
 131
 132       std::pair<cost_t, T> front() const {
 133         assert(!(q.empty()));
 134         assert(cur != q.end());
 135         return cur->second.front();
 136       }
 137
 138       void pop_front() {
 139         assert(!(q.empty()));
 140         assert(cur != q.end());
 141         cur->second.pop_front();
 142         if (cur->second.empty()) {
 143           auto i = cur;
 144           ++cur;
 145           q.erase(i);
 146         } else {
 147           ++cur;
 148         }
 149         if (cur == q.end()) {
 150           cur = q.begin();
 151         }
 152         size--;
 153       }
 154
 155       unsigned length() const {
 156         assert(size >= 0);
 157         return (unsigned)size;
 158       }
 159
 160       bool empty() const {
 161         return q.empty();
 162       }
 163
 164       void remove_by_filter(std::function<bool (const T&)> f) {
 165         for (typename Classes::iterator i = q.begin();
 166              i != q.end();
 167              /* no-inc */) {
 168           size -= filter_list_pairs(&(i->second), f);
 169           if (i->second.empty()) {
 170             if (cur == i) {
 171               ++cur;
 172             }
 173             i = q.erase(i);
 174           } else {
 175             ++i;
 176           }
 177         }
 178         if (cur == q.end()) cur = q.begin();
 179       }
 180
 181       void remove_by_class(K k, std::list<T> *out) {
 182         typename Classes::iterator i = q.find(k);
 183         if (i == q.end()) {
 184           return;
 185         }
 186         size -= i->second.size();
 187         if (i == cur) {
 188           ++cur;
 189         }
 190         if (out) {
 191           for (auto j = i->second.rbegin(); j != i->second.rend(); ++j) {
 192             out->push_front(j->second);
 193           }
 194         }
 195         q.erase(i);
 196         if (cur == q.end()) cur = q.begin();
 197       }
 198
 199       void dump(ceph::Formatter *f) const {
 200         f->dump_int("size", size);
 201         f->dump_int("num_keys", q.size());
 202       }
 203     };
 204
 205     using SubQueues = std::map<priority_t, SubQueue>;
 206
 207     SubQueues high_queue;
 208
 209     dmc::PullPriorityQueue<K,T> queue;
 210
 211     // when enqueue_front is called, rather than try to re-calc tags
 212     // to put in mClock priority queue, we'll just keep a separate
 213     // list from which we dequeue items first, and only when it's
 214     // empty do we use queue.
 215     std::list<std::pair<K,T>> queue_front;
 216
 217   public:
 218
 219     mClockQueue(
 220       const typename dmc::PullPriorityQueue<K,T>::ClientInfoFunc& info_func) :
 221       queue(info_func, true)
 222     {
 223       // empty
 224     }
 225
 226     unsigned length() const override final {
 227       unsigned total = 0;
 228       total += queue_front.size();
 229       total += queue.request_count();
 230       for (auto i = high_queue.cbegin(); i != high_queue.cend(); ++i) {
 231         assert(i->second.length());
 232         total += i->second.length();
 233       }
 234       return total;
 235     }
 236
 237     // be sure to do things in reverse priority order and push_front
 238     // to the list so items end up on list in front-to-back priority
 239     // order
 240     void remove_by_filter(std::function<bool (const T&)> filter_accum) {
 241       queue.remove_by_req_filter(filter_accum, true);
 242
 243       for (auto i = queue_front.rbegin(); i != queue_front.rend(); /* no-inc */) {
 244         if (filter_accum(i->second)) {
 245           i = decltype(i){ queue_front.erase(std::next(i).base()) };
 246         } else {
 247           ++i;
 248         }
 249       }
 250
 251       for (typename SubQueues::iterator i = high_queue.begin();
 252            i != high_queue.end();
 253            /* no-inc */ ) {
 254         i->second.remove_by_filter(filter_accum);
 255         if (i->second.empty()) {
 256           i = high_queue.erase(i);
 257         } else {
 258           ++i;
 259         }
 260       }
 261     }
 262
 263     void remove_by_class(K k, std::list<T> *out = nullptr) override final {
 264       if (out) {
 265         queue.remove_by_client(k,
 266                                true,
 267                                [&out] (const T& t) { out->push_front(t); });
 268       } else {
 269         queue.remove_by_client(k, true);
 270       }
 271
 272       for (auto i = queue_front.rbegin(); i != queue_front.rend(); /* no-inc */) {
 273         if (k == i->first) {
 274           if (nullptr != out) out->push_front(i->second);
 275           i = decltype(i){ queue_front.erase(std::next(i).base()) };
 276         } else {
 277           ++i;
 278         }
 279       }
 280
 281       for (auto i = high_queue.begin(); i != high_queue.end(); /* no-inc */) {
 282         i->second.remove_by_class(k, out);
 283         if (i->second.empty()) {
 284           i = high_queue.erase(i);
 285         } else {
 286           ++i;
 287         }
 288       }
 289     }
 290
 291     void enqueue_strict(K cl, unsigned priority, T item) override final {
 292       high_queue[priority].enqueue(cl, 0, item);
 293     }
 294
 295     void enqueue_strict_front(K cl, unsigned priority, T item) override final {
 296       high_queue[priority].enqueue_front(cl, 0, item);
 297     }
 298
 299     void enqueue(K cl, unsigned priority, unsigned cost, T item) override final {
 300       // priority is ignored
 301       queue.add_request(std::move(item), cl, cost);
 302     }
 303
 304     void enqueue_front(K cl,
 305                        unsigned priority,
 306                        unsigned cost,
 307                        T item) override final {
 308       queue_front.emplace_front(std::pair<K,T>(cl, item));
 309     }
 310
 311     bool empty() const override final {
 312       return queue.empty() && high_queue.empty() && queue_front.empty();
 313     }
 314
 315     T dequeue() override final {
 316       assert(!empty());
 317
 318       if (!(high_queue.empty())) {
 319         T ret = high_queue.rbegin()->second.front().second;
 320         high_queue.rbegin()->second.pop_front();
 321         if (high_queue.rbegin()->second.empty()) {
 322           high_queue.erase(high_queue.rbegin()->first);
 323         }
 324         return ret;
 325       }
 326
 327       if (!queue_front.empty()) {
 328         T ret = queue_front.front().second;
 329         queue_front.pop_front();
 330         return ret;
 331       }
 332
 333       auto pr = queue.pull_request();
 334       assert(pr.is_retn());
 335       auto& retn = pr.get_retn();
 336       return *(retn.request);
 337     }
 338
 339     void dump(ceph::Formatter *f) const override final {
 340       f->open_array_section("high_queues");
 341       for (typename SubQueues::const_iterator p = high_queue.begin();
 342            p != high_queue.end();
 343            ++p) {
 344         f->open_object_section("subqueue");
 345         f->dump_int("priority", p->first);
 346         p->second.dump(f);
 347         f->close_section();
 348       }
 349       f->close_section();
 350
 351       f->open_object_section("queue_front");
 352       f->dump_int("size", queue_front.size());
 353       f->close_section();
 354
 355       f->open_object_section("queue");
 356       f->dump_int("size", queue.request_count());
 357       f->close_section();
 358     } // dump
 359   };
 360
 361 } // namespace ceph