ceph/src/mds/SessionMap.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) 2004-2006 Sage Weil <sage@newdream.net>
   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 #ifndef CEPH_MDS_SESSIONMAP_H
  16 #define CEPH_MDS_SESSIONMAP_H
  17
  18 #include <set>
  19
  20 #include "include/unordered_map.h"
  21
  22 #include "include/Context.h"
  23 #include "include/xlist.h"
  24 #include "include/elist.h"
  25 #include "include/interval_set.h"
  26 #include "mdstypes.h"
  27 #include "mds/MDSAuthCaps.h"
  28 #include "common/perf_counters.h"
  29 #include "common/DecayCounter.h"
  30
  31 #include "CInode.h"
  32 #include "Capability.h"
  33 #include "MDSContext.h"
  34 #include "msg/Message.h"
  35
  36 struct MDRequestImpl;
  37
  38 enum {
  39   l_mdssm_first = 5500,
  40   l_mdssm_session_count,
  41   l_mdssm_session_add,
  42   l_mdssm_session_remove,
  43   l_mdssm_session_open,
  44   l_mdssm_session_stale,
  45   l_mdssm_total_load,
  46   l_mdssm_avg_load,
  47   l_mdssm_avg_session_uptime,
  48   l_mdssm_last,
  49 };
  50
  51 class CInode;
  52
  53 /*
  54  * session
  55  */
  56
  57 class Session : public RefCountedObject {
  58   // -- state etc --
  59 public:
  60   /*
  61
  62         <deleted> <-- closed <------------+
  63              ^         |                  |
  64              |         v                  |
  65           killing <-- opening <----+      |
  66              ^         |           |      |
  67              |         v           |      |
  68            stale <--> open --> closing ---+
  69
  70     + additional dimension of 'importing' (with counter)
  71
  72   */
  73
  74   using clock = ceph::coarse_mono_clock;
  75   using time = ceph::coarse_mono_time;
  76
  77   enum {
  78     STATE_CLOSED = 0,
  79     STATE_OPENING = 1,   // journaling open
  80     STATE_OPEN = 2,
  81     STATE_CLOSING = 3,   // journaling close
  82     STATE_STALE = 4,
  83     STATE_KILLING = 5
  84   };
  85
  86   Session() = delete;
  87   Session(ConnectionRef con) :
  88     item_session_list(this),
  89     requests(member_offset(MDRequestImpl, item_session_request)),
  90     recall_caps(g_conf().get_val<double>("mds_recall_warning_decay_rate")),
  91     release_caps(g_conf().get_val<double>("mds_recall_warning_decay_rate")),
  92     recall_caps_throttle(g_conf().get_val<double>("mds_recall_max_decay_rate")),
  93     recall_caps_throttle2o(0.5),
  94     session_cache_liveness(g_conf().get_val<double>("mds_session_cache_liveness_decay_rate")),
  95     birth_time(clock::now())
  96   {
  97     set_connection(std::move(con));
  98   }
  99   ~Session() override {
 100     if (state == STATE_CLOSED) {
 101       item_session_list.remove_myself();
 102     } else {
 103       ceph_assert(!item_session_list.is_on_list());
 104     }
 105     preopen_out_queue.clear();
 106   }
 107
 108   static std::string_view get_state_name(int s) {
 109     switch (s) {
 110     case STATE_CLOSED: return "closed";
 111     case STATE_OPENING: return "opening";
 112     case STATE_OPEN: return "open";
 113     case STATE_CLOSING: return "closing";
 114     case STATE_STALE: return "stale";
 115     case STATE_KILLING: return "killing";
 116     default: return "???";
 117     }
 118   }
 119
 120   void dump(Formatter *f) const;
 121   void push_pv(version_t pv)
 122   {
 123     ceph_assert(projected.empty() || projected.back() != pv);
 124     projected.push_back(pv);
 125   }
 126
 127   void pop_pv(version_t v)
 128   {
 129     ceph_assert(!projected.empty());
 130     ceph_assert(projected.front() == v);
 131     projected.pop_front();
 132   }
 133
 134   int get_state() const { return state; }
 135   void set_state(int new_state)
 136   {
 137     if (state != new_state) {
 138       state = new_state;
 139       state_seq++;
 140     }
 141   }
 142
 143   void set_reconnecting(bool s) { reconnecting = s; }
 144
 145   void decode(bufferlist::const_iterator &p);
 146   template<typename T>
 147   void set_client_metadata(T&& meta)
 148   {
 149     info.client_metadata = std::forward<T>(meta);
 150     _update_human_name();
 151   }
 152
 153   const std::string& get_human_name() const {return human_name;}
 154
 155   size_t get_request_count() const;
 156
 157   void notify_cap_release(size_t n_caps);
 158   uint64_t notify_recall_sent(size_t new_limit);
 159   auto get_recall_caps_throttle() const {
 160     return recall_caps_throttle.get();
 161   }
 162   auto get_recall_caps_throttle2o() const {
 163     return recall_caps_throttle2o.get();
 164   }
 165   auto get_recall_caps() const {
 166     return recall_caps.get();
 167   }
 168   auto get_release_caps() const {
 169     return release_caps.get();
 170   }
 171   auto get_session_cache_liveness() const {
 172     return session_cache_liveness.get();
 173   }
 174
 175   inodeno_t take_ino(inodeno_t ino = 0) {
 176     if (ino) {
 177       if (!info.prealloc_inos.contains(ino))
 178         return 0;
 179       info.prealloc_inos.erase(ino);
 180       if (delegated_inos.contains(ino))
 181         delegated_inos.erase(ino);
 182     } else {
 183       /* Grab first prealloc_ino that isn't delegated */
 184       for (const auto& [start, len] : info.prealloc_inos) {
 185         for (auto i = start ; i < start + len ; i += 1) {
 186           inodeno_t dstart, dlen;
 187           if (!delegated_inos.contains(i, &dstart, &dlen)) {
 188             ino = i;
 189             info.prealloc_inos.erase(ino);
 190             break;
 191           }
 192           /* skip to end of delegated interval */
 193           i = dstart + dlen - 1;
 194         }
 195         if (ino)
 196           break;
 197       }
 198     }
 199     if (ino)
 200       info.used_inos.insert(ino, 1);
 201     return ino;
 202   }
 203   void delegate_inos(int want, interval_set<inodeno_t>& newinos) {
 204     want -= (int)delegated_inos.size();
 205     if (want <= 0)
 206       return;
 207
 208     for (const auto& [start, len] : info.prealloc_inos) {
 209       for (auto i = start ; i < start + len ; i += 1) {
 210         inodeno_t dstart, dlen;
 211         if (!delegated_inos.contains(i, &dstart, &dlen)) {
 212           delegated_inos.insert(i);
 213           newinos.insert(i);
 214           if (--want == 0)
 215              return;
 216         } else {
 217           /* skip to end of delegated interval */
 218           i = dstart + dlen - 1;
 219         }
 220       }
 221     }
 222   }
 223
 224   // sans any delegated ones
 225   int get_num_prealloc_inos() const {
 226     return info.prealloc_inos.size() - delegated_inos.size();
 227   }
 228
 229   int get_num_projected_prealloc_inos() const {
 230     return get_num_prealloc_inos() + pending_prealloc_inos.size();
 231   }
 232
 233   client_t get_client() const {
 234     return info.get_client();
 235   }
 236
 237   std::string_view get_state_name() const { return get_state_name(state); }
 238   uint64_t get_state_seq() const { return state_seq; }
 239   bool is_closed() const { return state == STATE_CLOSED; }
 240   bool is_opening() const { return state == STATE_OPENING; }
 241   bool is_open() const { return state == STATE_OPEN; }
 242   bool is_closing() const { return state == STATE_CLOSING; }
 243   bool is_stale() const { return state == STATE_STALE; }
 244   bool is_killing() const { return state == STATE_KILLING; }
 245
 246   void inc_importing() {
 247     ++importing_count;
 248   }
 249   void dec_importing() {
 250     ceph_assert(importing_count > 0);
 251     --importing_count;
 252   }
 253   bool is_importing() const { return importing_count > 0; }
 254
 255   void set_load_avg_decay_rate(double rate) {
 256     ceph_assert(is_open() || is_stale());
 257     load_avg = DecayCounter(rate);
 258   }
 259   uint64_t get_load_avg() const {
 260     return (uint64_t)load_avg.get();
 261   }
 262   void hit_session() {
 263     load_avg.adjust();
 264   }
 265
 266   double get_session_uptime() const {
 267     chrono::duration<double> uptime = clock::now() - birth_time;
 268     return uptime.count();
 269   }
 270
 271   time get_birth_time() const {
 272     return birth_time;
 273   }
 274
 275   void inc_cap_gen() { ++cap_gen; }
 276   uint32_t get_cap_gen() const { return cap_gen; }
 277
 278   version_t inc_push_seq() { return ++cap_push_seq; }
 279   version_t get_push_seq() const { return cap_push_seq; }
 280
 281   version_t wait_for_flush(MDSContext* c) {
 282     waitfor_flush[get_push_seq()].push_back(c);
 283     return get_push_seq();
 284   }
 285   void finish_flush(version_t seq, MDSContext::vec& ls) {
 286     while (!waitfor_flush.empty()) {
 287       auto it = waitfor_flush.begin();
 288       if (it->first > seq)
 289         break;
 290       auto& v = it->second;
 291       ls.insert(ls.end(), v.begin(), v.end());
 292       waitfor_flush.erase(it);
 293     }
 294   }
 295
 296   void touch_cap(Capability *cap) {
 297     session_cache_liveness.hit(1.0);
 298     caps.push_front(&cap->item_session_caps);
 299   }
 300
 301   void touch_cap_bottom(Capability *cap) {
 302     session_cache_liveness.hit(1.0);
 303     caps.push_back(&cap->item_session_caps);
 304   }
 305
 306   void touch_lease(ClientLease *r) {
 307     session_cache_liveness.hit(1.0);
 308     leases.push_back(&r->item_session_lease);
 309   }
 310
 311   bool is_any_flush_waiter() {
 312     return !waitfor_flush.empty();
 313   }
 314
 315   void add_completed_request(ceph_tid_t t, inodeno_t created) {
 316     info.completed_requests[t] = created;
 317     completed_requests_dirty = true;
 318   }
 319   bool trim_completed_requests(ceph_tid_t mintid) {
 320     // trim
 321     bool erased_any = false;
 322     while (!info.completed_requests.empty() &&
 323            (mintid == 0 || info.completed_requests.begin()->first < mintid)) {
 324       info.completed_requests.erase(info.completed_requests.begin());
 325       erased_any = true;
 326     }
 327
 328     if (erased_any) {
 329       completed_requests_dirty = true;
 330     }
 331     return erased_any;
 332   }
 333   bool have_completed_request(ceph_tid_t tid, inodeno_t *pcreated) const {
 334     map<ceph_tid_t,inodeno_t>::const_iterator p = info.completed_requests.find(tid);
 335     if (p == info.completed_requests.end())
 336       return false;
 337     if (pcreated)
 338       *pcreated = p->second;
 339     return true;
 340   }
 341
 342   void add_completed_flush(ceph_tid_t tid) {
 343     info.completed_flushes.insert(tid);
 344   }
 345   bool trim_completed_flushes(ceph_tid_t mintid) {
 346     bool erased_any = false;
 347     while (!info.completed_flushes.empty() &&
 348         (mintid == 0 || *info.completed_flushes.begin() < mintid)) {
 349       info.completed_flushes.erase(info.completed_flushes.begin());
 350       erased_any = true;
 351     }
 352     if (erased_any) {
 353       completed_requests_dirty = true;
 354     }
 355     return erased_any;
 356   }
 357   bool have_completed_flush(ceph_tid_t tid) const {
 358     return info.completed_flushes.count(tid);
 359   }
 360
 361   unsigned get_num_completed_flushes() const { return info.completed_flushes.size(); }
 362   unsigned get_num_trim_flushes_warnings() const {
 363     return num_trim_flushes_warnings;
 364   }
 365   void inc_num_trim_flushes_warnings() { ++num_trim_flushes_warnings; }
 366   void reset_num_trim_flushes_warnings() { num_trim_flushes_warnings = 0; }
 367
 368   unsigned get_num_completed_requests() const { return info.completed_requests.size(); }
 369   unsigned get_num_trim_requests_warnings() const {
 370     return num_trim_requests_warnings;
 371   }
 372   void inc_num_trim_requests_warnings() { ++num_trim_requests_warnings; }
 373   void reset_num_trim_requests_warnings() { num_trim_requests_warnings = 0; }
 374
 375   bool has_dirty_completed_requests() const
 376   {
 377     return completed_requests_dirty;
 378   }
 379
 380   void clear_dirty_completed_requests()
 381   {
 382     completed_requests_dirty = false;
 383   }
 384
 385   int check_access(CInode *in, unsigned mask, int caller_uid, int caller_gid,
 386                    const vector<uint64_t> *gid_list, int new_uid, int new_gid);
 387
 388   void set_connection(ConnectionRef con) {
 389     connection = std::move(con);
 390     auto& c = connection;
 391     if (c) {
 392       info.auth_name = c->get_peer_entity_name();
 393       info.inst.addr = c->get_peer_socket_addr();
 394       info.inst.name = entity_name_t(c->get_peer_type(), c->get_peer_global_id());
 395     }
 396   }
 397   const ConnectionRef& get_connection() const {
 398     return connection;
 399   }
 400
 401   void clear() {
 402     pending_prealloc_inos.clear();
 403     delegated_inos.clear();
 404     info.clear_meta();
 405
 406     cap_push_seq = 0;
 407     last_cap_renew = clock::zero();
 408   }
 409
 410   Session *reclaiming_from = nullptr;
 411   session_info_t info;                         ///< durable bits
 412   MDSAuthCaps auth_caps;
 413
 414   xlist<Session*>::item item_session_list;
 415
 416   list<ref_t<Message>> preopen_out_queue;  ///< messages for client, queued before they connect
 417
 418   /* This is mutable to allow get_request_count to be const. elist does not
 419    * support const iterators yet.
 420    */
 421   mutable elist<MDRequestImpl*> requests;
 422
 423   interval_set<inodeno_t> pending_prealloc_inos; // journaling prealloc, will be added to prealloc_inos
 424   interval_set<inodeno_t> delegated_inos; // hand these out to client
 425
 426   xlist<Capability*> caps;     // inodes with caps; front=most recently used
 427   xlist<ClientLease*> leases;  // metadata leases to clients
 428   time last_cap_renew = clock::zero();
 429   time last_seen = clock::zero();
 430
 431   // -- leases --
 432   uint32_t lease_seq = 0;
 433
 434 protected:
 435   ConnectionRef connection;
 436
 437 private:
 438   friend class SessionMap;
 439
 440   // Human (friendly) name is soft state generated from client metadata
 441   void _update_human_name();
 442
 443   int state = STATE_CLOSED;
 444   bool reconnecting = false;
 445   uint64_t state_seq = 0;
 446   int importing_count = 0;
 447
 448   std::string human_name;
 449
 450   // Versions in this session was projected: used to verify
 451   // that appropriate mark_dirty calls follow.
 452   std::deque<version_t> projected;
 453
 454   // request load average for this session
 455   DecayCounter load_avg;
 456
 457   // Ephemeral state for tracking progress of capability recalls
 458   // caps being recalled recently by this session; used for Beacon warnings
 459   DecayCounter recall_caps;  // caps that have been released
 460   DecayCounter release_caps;
 461   // throttle on caps recalled
 462   DecayCounter recall_caps_throttle;
 463   // second order throttle that prevents recalling too quickly
 464   DecayCounter recall_caps_throttle2o;
 465   // New limit in SESSION_RECALL
 466   uint32_t recall_limit = 0;
 467
 468   // session caps liveness
 469   DecayCounter session_cache_liveness;
 470
 471   // session start time -- used to track average session time
 472   // note that this is initialized in the constructor rather
 473   // than at the time of adding a session to the sessionmap
 474   // as journal replay of sessionmap will not call add_session().
 475   time birth_time;
 476
 477   // -- caps --
 478   uint32_t cap_gen = 0;
 479   version_t cap_push_seq = 0;        // cap push seq #
 480   map<version_t, MDSContext::vec > waitfor_flush; // flush session messages
 481
 482   // Has completed_requests been modified since the last time we
 483   // wrote this session out?
 484   bool completed_requests_dirty = false;
 485
 486   unsigned num_trim_flushes_warnings = 0;
 487   unsigned num_trim_requests_warnings = 0;
 488 };
 489
 490 class SessionFilter
 491 {
 492 public:
 493   SessionFilter() : reconnecting(false, false) {}
 494
 495   bool match(
 496       const Session &session,
 497       std::function<bool(client_t)> is_reconnecting) const;
 498   int parse(const std::vector<std::string> &args, std::stringstream *ss);
 499   void set_reconnecting(bool v)
 500   {
 501     reconnecting.first = true;
 502     reconnecting.second = v;
 503   }
 504
 505   std::map<std::string, std::string> metadata;
 506   std::string auth_name;
 507   std::string state;
 508   int64_t id = 0;
 509 protected:
 510   // First is whether to filter, second is filter value
 511   std::pair<bool, bool> reconnecting;
 512 };
 513
 514 /*
 515  * session map
 516  */
 517
 518 class MDSRank;
 519
 520 /**
 521  * Encapsulate the serialized state associated with SessionMap.  Allows
 522  * encode/decode outside of live MDS instance.
 523  */
 524 class SessionMapStore {
 525 public:
 526   using clock = Session::clock;
 527   using time = Session::time;
 528
 529   SessionMapStore(): total_load_avg(decay_rate) {}
 530   virtual ~SessionMapStore() {};
 531
 532   version_t get_version() const {return version;}
 533
 534   virtual void encode_header(bufferlist *header_bl);
 535   virtual void decode_header(bufferlist &header_bl);
 536   virtual void decode_values(std::map<std::string, bufferlist> &session_vals);
 537   virtual void decode_legacy(bufferlist::const_iterator& blp);
 538   void dump(Formatter *f) const;
 539
 540   void set_rank(mds_rank_t r)
 541   {
 542     rank = r;
 543   }
 544
 545   Session* get_or_add_session(const entity_inst_t& i) {
 546     Session *s;
 547     auto session_map_entry = session_map.find(i.name);
 548     if (session_map_entry != session_map.end()) {
 549       s = session_map_entry->second;
 550     } else {
 551       s = session_map[i.name] = new Session(ConnectionRef());
 552       s->info.inst = i;
 553       s->last_cap_renew = Session::clock::now();
 554       if (logger) {
 555         logger->set(l_mdssm_session_count, session_map.size());
 556         logger->inc(l_mdssm_session_add);
 557       }
 558     }
 559
 560     return s;
 561   }
 562
 563   static void generate_test_instances(std::list<SessionMapStore*>& ls);
 564
 565   void reset_state()
 566   {
 567     session_map.clear();
 568   }
 569
 570   mds_rank_t rank = MDS_RANK_NONE;
 571
 572 protected:
 573   version_t version = 0;
 574   ceph::unordered_map<entity_name_t, Session*> session_map;
 575   PerfCounters *logger =nullptr;
 576
 577   // total request load avg
 578   double decay_rate = g_conf().get_val<double>("mds_request_load_average_decay_rate");
 579   DecayCounter total_load_avg;
 580 };
 581
 582 class SessionMap : public SessionMapStore {
 583 public:
 584   SessionMap() = delete;
 585   explicit SessionMap(MDSRank *m) : mds(m) {}
 586
 587   ~SessionMap() override
 588   {
 589     for (auto p : by_state)
 590       delete p.second;
 591
 592     if (logger) {
 593       g_ceph_context->get_perfcounters_collection()->remove(logger);
 594     }
 595
 596     delete logger;
 597   }
 598
 599   uint64_t set_state(Session *session, int state);
 600   void update_average_session_age();
 601
 602   void register_perfcounters();
 603
 604   void set_version(const version_t v)
 605   {
 606     version = projected = v;
 607   }
 608
 609   void set_projected(const version_t v)
 610   {
 611     projected = v;
 612   }
 613
 614   version_t get_projected() const
 615   {
 616     return projected;
 617   }
 618
 619   version_t get_committed() const
 620   {
 621     return committed;
 622   }
 623
 624   version_t get_committing() const
 625   {
 626     return committing;
 627   }
 628
 629   // sessions
 630   void decode_legacy(bufferlist::const_iterator& blp) override;
 631   bool empty() const { return session_map.empty(); }
 632   const auto& get_sessions() const {
 633     return session_map;
 634   }
 635
 636   bool is_any_state(int state) const {
 637     auto it = by_state.find(state);
 638     if (it == by_state.end() || it->second->empty())
 639       return false;
 640     return true;
 641   }
 642
 643   bool have_unclosed_sessions() const {
 644     return
 645       is_any_state(Session::STATE_OPENING) ||
 646       is_any_state(Session::STATE_OPEN) ||
 647       is_any_state(Session::STATE_CLOSING) ||
 648       is_any_state(Session::STATE_STALE) ||
 649       is_any_state(Session::STATE_KILLING);
 650   }
 651   bool have_session(entity_name_t w) const {
 652     return session_map.count(w);
 653   }
 654   Session* get_session(entity_name_t w) {
 655     auto session_map_entry = session_map.find(w);
 656     return (session_map_entry != session_map.end() ?
 657             session_map_entry-> second : nullptr);
 658   }
 659   const Session* get_session(entity_name_t w) const {
 660     ceph::unordered_map<entity_name_t, Session*>::const_iterator p = session_map.find(w);
 661     if (p == session_map.end()) {
 662       return NULL;
 663     } else {
 664       return p->second;
 665     }
 666   }
 667
 668   void add_session(Session *s);
 669   void remove_session(Session *s);
 670   void touch_session(Session *session);
 671
 672   Session *get_oldest_session(int state) {
 673     auto by_state_entry = by_state.find(state);
 674     if (by_state_entry == by_state.end() || by_state_entry->second->empty())
 675       return 0;
 676     return by_state_entry->second->front();
 677   }
 678
 679   void dump();
 680
 681   template<typename F>
 682   void get_client_sessions(F&& f) const {
 683     for (const auto& p : session_map) {
 684       auto& session = p.second;
 685       if (session->info.inst.name.is_client())
 686         f(session);
 687     }
 688   }
 689   template<typename C>
 690   void get_client_session_set(C& c) const {
 691     auto f = [&c](auto& s) {
 692       c.insert(s);
 693     };
 694     get_client_sessions(f);
 695   }
 696
 697   // helpers
 698   entity_inst_t& get_inst(entity_name_t w) {
 699     ceph_assert(session_map.count(w));
 700     return session_map[w]->info.inst;
 701   }
 702   version_t get_push_seq(client_t client) {
 703     return get_session(entity_name_t::CLIENT(client.v))->get_push_seq();
 704   }
 705   bool have_completed_request(metareqid_t rid) {
 706     Session *session = get_session(rid.name);
 707     return session && session->have_completed_request(rid.tid, NULL);
 708   }
 709   void trim_completed_requests(entity_name_t c, ceph_tid_t tid) {
 710     Session *session = get_session(c);
 711     ceph_assert(session);
 712     session->trim_completed_requests(tid);
 713   }
 714
 715   void wipe();
 716   void wipe_ino_prealloc();
 717
 718   object_t get_object_name() const;
 719
 720   void load(MDSContext *onload);
 721   void _load_finish(
 722       int operation_r,
 723       int header_r,
 724       int values_r,
 725       bool first,
 726       bufferlist &header_bl,
 727       std::map<std::string, bufferlist> &session_vals,
 728       bool more_session_vals);
 729
 730   void load_legacy();
 731   void _load_legacy_finish(int r, bufferlist &bl);
 732
 733   void save(MDSContext *onsave, version_t needv=0);
 734   void _save_finish(version_t v);
 735
 736   /**
 737    * Advance the version, and mark this session
 738    * as dirty within the new version.
 739    *
 740    * Dirty means journalled but needing writeback
 741    * to the backing store.  Must have called
 742    * mark_projected previously for this session.
 743    */
 744   void mark_dirty(Session *session, bool may_save=true);
 745
 746   /**
 747    * Advance the projected version, and mark this
 748    * session as projected within the new version
 749    *
 750    * Projected means the session is updated in memory
 751    * but we're waiting for the journal write of the update
 752    * to finish.  Must subsequently call mark_dirty
 753    * for sessions in the same global order as calls
 754    * to mark_projected.
 755    */
 756   version_t mark_projected(Session *session);
 757
 758   /**
 759    * During replay, advance versions to account
 760    * for a session modification, and mark the
 761    * session dirty.
 762    */
 763   void replay_dirty_session(Session *session);
 764
 765   /**
 766    * During replay, if a session no longer present
 767    * would have consumed a version, advance `version`
 768    * and `projected` to account for that.
 769    */
 770   void replay_advance_version();
 771
 772   /**
 773    * During replay, open sessions, advance versions and
 774    * mark these sessions as dirty.
 775    */
 776   void replay_open_sessions(version_t event_cmapv,
 777                             map<client_t,entity_inst_t>& client_map,
 778                             map<client_t,client_metadata_t>& client_metadata_map);
 779
 780   /**
 781    * For these session IDs, if a session exists with this ID, and it has
 782    * dirty completed_requests, then persist it immediately
 783    * (ahead of usual project/dirty versioned writes
 784    *  of the map).
 785    */
 786   void save_if_dirty(const std::set<entity_name_t> &tgt_sessions,
 787                      MDSGatherBuilder *gather_bld);
 788
 789   void hit_session(Session *session);
 790   void handle_conf_change(const std::set <std::string> &changed);
 791
 792   MDSRank *mds;
 793   map<int,xlist<Session*>* > by_state;
 794   map<version_t, MDSContext::vec > commit_waiters;
 795
 796   // -- loading, saving --
 797   inodeno_t ino;
 798   MDSContext::vec waiting_for_load;
 799
 800 protected:
 801   void _mark_dirty(Session *session, bool may_save);
 802
 803   version_t projected = 0, committing = 0, committed = 0;
 804   std::set<entity_name_t> dirty_sessions;
 805   std::set<entity_name_t> null_sessions;
 806   bool loaded_legacy = false;
 807
 808 private:
 809   uint64_t get_session_count_in_state(int state) {
 810     return !is_any_state(state) ? 0 : by_state[state]->size();
 811   }
 812
 813   void update_average_birth_time(const Session &s, bool added=true) {
 814     uint32_t sessions = session_map.size();
 815     time birth_time = s.get_birth_time();
 816
 817     if (sessions == 1) {
 818       avg_birth_time = added ? birth_time : clock::zero();
 819       return;
 820     }
 821
 822     if (added) {
 823       avg_birth_time = clock::time_point(
 824         ((avg_birth_time - clock::zero()) / sessions) * (sessions - 1) +
 825         (birth_time - clock::zero()) / sessions);
 826     } else {
 827       avg_birth_time = clock::time_point(
 828         ((avg_birth_time - clock::zero()) / (sessions - 1)) * sessions -
 829         (birth_time - clock::zero()) / (sessions - 1));
 830     }
 831   }
 832
 833   time avg_birth_time = clock::zero();
 834 };
 835
 836 std::ostream& operator<<(std::ostream &out, const Session &s);
 837 #endif