+++ /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 ASYNC_RESERVER_H
-#define ASYNC_RESERVER_H
-
-#include "common/Finisher.h"
-#include "common/Formatter.h"
-
-#define rdout(x) lgeneric_subdout(cct,reserver,x)
-
-/**
- * Manages a configurable number of asyncronous reservations.
- *
- * Memory usage is linear with the number of items queued and
- * linear with respect to the total number of priorities used
- * over all time.
- */
-template <typename T>
-class AsyncReserver {
- CephContext *cct;
- Finisher *f;
- unsigned max_allowed;
- unsigned min_priority;
- Mutex lock;
-
- struct Reservation {
- T item;
- unsigned prio = 0;
- Context *grant = 0;
- Context *preempt = 0;
- Reservation() {}
- Reservation(T i, unsigned pr, Context *g, Context *p = 0)
- : item(i), prio(pr), grant(g), preempt(p) {}
- void dump(Formatter *f) const {
- f->dump_stream("item") << item;
- f->dump_unsigned("prio", prio);
- f->dump_bool("can_preempt", !!preempt);
- }
- friend ostream& operator<<(ostream& out, const Reservation& r) {
- return out << r.item << "(prio " << r.prio << " grant " << r.grant
- << " preempt " << r.preempt << ")";
- }
- };
-
- map<unsigned, list<Reservation>> queues;
- map<T, pair<unsigned, typename list<Reservation>::iterator>> queue_pointers;
- map<T,Reservation> in_progress;
- set<pair<unsigned,T>> preempt_by_prio; ///< in_progress that can be preempted
-
- void preempt_one() {
- assert(!preempt_by_prio.empty());
- auto q = in_progress.find(preempt_by_prio.begin()->second);
- assert(q != in_progress.end());
- Reservation victim = q->second;
- rdout(10) << __func__ << " preempt " << victim << dendl;
- f->queue(victim.preempt);
- victim.preempt = nullptr;
- in_progress.erase(q);
- preempt_by_prio.erase(preempt_by_prio.begin());
- }
-
- void do_queues() {
- rdout(20) << __func__ << ":\n";
- JSONFormatter jf(true);
- jf.open_object_section("queue");
- _dump(&jf);
- jf.close_section();
- jf.flush(*_dout);
- *_dout << dendl;
-
- // in case min_priority was adjusted up or max_allowed was adjusted down
- while (!preempt_by_prio.empty() &&
- (in_progress.size() > max_allowed ||
- preempt_by_prio.begin()->first < min_priority)) {
- preempt_one();
- }
-
- while (!queues.empty()) {
- // choose highest priority queue
- auto it = queues.end();
- --it;
- assert(!it->second.empty());
- if (it->first < min_priority) {
- break;
- }
- if (in_progress.size() >= max_allowed &&
- !preempt_by_prio.empty() &&
- it->first > preempt_by_prio.begin()->first) {
- preempt_one();
- }
- if (in_progress.size() >= max_allowed) {
- break; // no room
- }
- // grant
- Reservation p = it->second.front();
- rdout(10) << __func__ << " grant " << p << dendl;
- queue_pointers.erase(p.item);
- it->second.pop_front();
- if (it->second.empty()) {
- queues.erase(it);
- }
- f->queue(p.grant);
- p.grant = nullptr;
- in_progress[p.item] = p;
- if (p.preempt) {
- preempt_by_prio.insert(make_pair(p.prio, p.item));
- }
- }
- }
-public:
- AsyncReserver(
- CephContext *cct,
- Finisher *f,
- unsigned max_allowed,
- unsigned min_priority = 0)
- : cct(cct),
- f(f),
- max_allowed(max_allowed),
- min_priority(min_priority),
- lock("AsyncReserver::lock") {}
-
- void set_max(unsigned max) {
- Mutex::Locker l(lock);
- max_allowed = max;
- do_queues();
- }
-
- void set_min_priority(unsigned min) {
- Mutex::Locker l(lock);
- min_priority = min;
- do_queues();
- }
-
- void dump(Formatter *f) {
- Mutex::Locker l(lock);
- _dump(f);
- }
- void _dump(Formatter *f) {
- f->dump_unsigned("max_allowed", max_allowed);
- f->dump_unsigned("min_priority", min_priority);
- f->open_array_section("queues");
- for (auto& p : queues) {
- f->open_object_section("queue");
- f->dump_unsigned("priority", p.first);
- f->open_array_section("items");
- for (auto& q : p.second) {
- f->dump_object("item", q);
- }
- f->close_section();
- f->close_section();
- }
- f->close_section();
- f->open_array_section("in_progress");
- for (auto& p : in_progress) {
- f->dump_object("item", p.second);
- }
- f->close_section();
- }
-
- /**
- * Requests a reservation
- *
- * Note, on_reserved may be called following cancel_reservation. Thus,
- * the callback must be safe in that case. Callback will be called
- * with no locks held. cancel_reservation must be called to release the
- * reservation slot.
- */
- void request_reservation(
- T item, ///< [in] reservation key
- Context *on_reserved, ///< [in] callback to be called on reservation
- unsigned prio, ///< [in] priority
- Context *on_preempt = 0 ///< [in] callback to be called if we are preempted (optional)
- ) {
- Mutex::Locker l(lock);
- Reservation r(item, prio, on_reserved, on_preempt);
- rdout(10) << __func__ << " queue " << r << dendl;
- assert(!queue_pointers.count(item) &&
- !in_progress.count(item));
- queues[prio].push_back(r);
- queue_pointers.insert(make_pair(item,
- make_pair(prio,--(queues[prio]).end())));
- do_queues();
- }
-
- /**
- * Cancels reservation
- *
- * Frees the reservation under key for use.
- * Note, after cancel_reservation, the reservation_callback may or
- * may not still be called.
- */
- void cancel_reservation(
- T item ///< [in] key for reservation to cancel
- ) {
- Mutex::Locker l(lock);
- auto i = queue_pointers.find(item);
- if (i != queue_pointers.end()) {
- unsigned prio = i->second.first;
- const Reservation& r = *i->second.second;
- rdout(10) << __func__ << " cancel " << r << " (was queued)" << dendl;
- delete r.grant;
- delete r.preempt;
- queues[prio].erase(i->second.second);
- if (queues[prio].empty()) {
- queues.erase(prio);
- }
- queue_pointers.erase(i);
- } else {
- auto p = in_progress.find(item);
- if (p != in_progress.end()) {
- rdout(10) << __func__ << " cancel " << p->second
- << " (was in progress)" << dendl;
- if (p->second.preempt) {
- preempt_by_prio.erase(make_pair(p->second.prio, p->second.item));
- delete p->second.preempt;
- }
- in_progress.erase(p);
- } else {
- rdout(10) << __func__ << " cancel " << item << " (not found)" << dendl;
- }
- }
- do_queues();
- }
-
- /**
- * Has reservations
- *
- * Return true if there are reservations in progress
- */
- bool has_reservation() {
- Mutex::Locker l(lock);
- return !in_progress.empty();
- }
- static const unsigned MAX_PRIORITY = (unsigned)-1;
-};
-
-#undef rdout
-#endif
Code Review / stor4nfv.git / blobdiff