+++ /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 CEPH_MON_SESSION_H
-#define CEPH_MON_SESSION_H
-
-#include "include/xlist.h"
-#include "msg/msg_types.h"
-#include "mon/mon_types.h"
-
-#include "auth/AuthServiceHandler.h"
-#include "osd/OSDMap.h"
-
-#include "MonCap.h"
-
-struct MonSession;
-
-struct Subscription {
- MonSession *session;
- string type;
- xlist<Subscription*>::item type_item;
- version_t next;
- bool onetime;
- bool incremental_onetime; // has CEPH_FEATURE_INCSUBOSDMAP
-
- Subscription(MonSession *s, const string& t) : session(s), type(t), type_item(this),
- next(0), onetime(false), incremental_onetime(false) {}
-};
-
-struct MonSession : public RefCountedObject {
- ConnectionRef con;
- int con_type = 0;
- uint64_t con_features = 0; // zero if AnonConnection
- entity_inst_t inst;
- utime_t session_timeout;
- utime_t time_established;
- bool closed;
- xlist<MonSession*>::item item;
- set<uint64_t> routed_request_tids;
- MonCap caps;
- uint64_t auid;
- uint64_t global_id;
-
- map<string, Subscription*> sub_map;
- epoch_t osd_epoch; // the osdmap epoch sent to the mon client
-
- AuthServiceHandler *auth_handler;
- EntityName entity_name;
-
- ConnectionRef proxy_con;
- uint64_t proxy_tid;
-
- MonSession(const entity_inst_t& i, Connection *c) :
- RefCountedObject(g_ceph_context),
- con(c),
- con_type(c->get_peer_type()),
- con_features(0),
- inst(i), closed(false), item(this),
- auid(0),
- global_id(0),
- osd_epoch(0),
- auth_handler(NULL),
- proxy_con(NULL), proxy_tid(0) {
- time_established = ceph_clock_now();
- if (c->get_messenger()) {
- // only fill in features if this is a non-anonymous connection
- con_features = c->get_features();
- }
- }
- ~MonSession() override {
- //generic_dout(0) << "~MonSession " << this << dendl;
- // we should have been removed before we get destructed; see MonSessionMap::remove_session()
- assert(!item.is_on_list());
- assert(sub_map.empty());
- delete auth_handler;
- }
-
- bool is_capable(string service, int mask) {
- map<string,string> args;
- return caps.is_capable(
- g_ceph_context,
- CEPH_ENTITY_TYPE_MON,
- entity_name,
- service, "", args,
- mask & MON_CAP_R, mask & MON_CAP_W, mask & MON_CAP_X);
- }
-};
-
-
-struct MonSessionMap {
- xlist<MonSession*> sessions;
- map<string, xlist<Subscription*>* > subs;
- multimap<int, MonSession*> by_osd;
- FeatureMap feature_map; // type -> features -> count
-
- MonSessionMap() {}
- ~MonSessionMap() {
- while (!subs.empty()) {
- assert(subs.begin()->second->empty());
- delete subs.begin()->second;
- subs.erase(subs.begin());
- }
- }
-
- unsigned get_size() const {
- return sessions.size();
- }
-
- void remove_session(MonSession *s) {
- assert(!s->closed);
- for (map<string,Subscription*>::iterator p = s->sub_map.begin(); p != s->sub_map.end(); ++p) {
- p->second->type_item.remove_myself();
- delete p->second;
- }
- s->sub_map.clear();
- s->item.remove_myself();
- if (s->inst.name.is_osd()) {
- for (multimap<int,MonSession*>::iterator p = by_osd.find(s->inst.name.num());
- p->first == s->inst.name.num();
- ++p)
- if (p->second == s) {
- by_osd.erase(p);
- break;
- }
- }
- if (s->con_features) {
- feature_map.rm(s->con_type, s->con_features);
- }
- s->closed = true;
- s->put();
- }
-
- MonSession *new_session(const entity_inst_t& i, Connection *c) {
- MonSession *s = new MonSession(i, c);
- assert(s);
- sessions.push_back(&s->item);
- if (i.name.is_osd())
- by_osd.insert(pair<int,MonSession*>(i.name.num(), s));
- if (s->con_features) {
- feature_map.add(s->con_type, s->con_features);
- }
- s->get(); // caller gets a ref
- return s;
- }
-
- MonSession *get_random_osd_session(OSDMap *osdmap) {
- // ok, this isn't actually random, but close enough.
- if (by_osd.empty())
- return 0;
- int n = by_osd.rbegin()->first + 1;
- int r = rand() % n;
-
- multimap<int,MonSession*>::iterator p = by_osd.lower_bound(r);
- if (p == by_osd.end())
- --p;
-
- if (!osdmap) {
- return p->second;
- }
-
- MonSession *s = NULL;
-
- multimap<int,MonSession*>::iterator b = p, f = p;
- bool backward = true, forward = true;
- while (backward || forward) {
- if (backward) {
- if (osdmap->is_up(b->first) &&
- osdmap->get_addr(b->first) == b->second->con->get_peer_addr()) {
- s = b->second;
- break;
- }
- if (b != by_osd.begin())
- --b;
- else
- backward = false;
- }
-
- forward = (f != by_osd.end());
- if (forward) {
- if (osdmap->is_up(f->first)) {
- s = f->second;
- break;
- }
- ++f;
- }
- }
-
- return s;
- }
-
- void add_update_sub(MonSession *s, const string& what, version_t start, bool onetime, bool incremental_onetime) {
- Subscription *sub = 0;
- if (s->sub_map.count(what)) {
- sub = s->sub_map[what];
- } else {
- sub = new Subscription(s, what);
- s->sub_map[what] = sub;
-
- if (!subs.count(what))
- subs[what] = new xlist<Subscription*>;
- subs[what]->push_back(&sub->type_item);
- }
- sub->next = start;
- sub->onetime = onetime;
- sub->incremental_onetime = onetime && incremental_onetime;
- }
-
- void remove_sub(Subscription *sub) {
- sub->session->sub_map.erase(sub->type);
- sub->type_item.remove_myself();
- delete sub;
- }
-};
-
-inline ostream& operator<<(ostream& out, const MonSession& s)
-{
- out << "MonSession(" << s.inst << " is "
- << (s.closed ? "closed" : "open");
- out << " " << s.caps << ")";
- return out;
-}
-
-#endif
Code Review / stor4nfv.git / blobdiff