// -*- 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 * * 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_MDS_SESSIONMAP_H #define CEPH_MDS_SESSIONMAP_H #include using std::set; #include "include/unordered_map.h" #include "include/Context.h" #include "include/xlist.h" #include "include/elist.h" #include "include/interval_set.h" #include "mdstypes.h" #include "mds/MDSAuthCaps.h" #include "common/perf_counters.h" class CInode; struct MDRequestImpl; #include "CInode.h" #include "Capability.h" #include "msg/Message.h" enum { l_mdssm_first = 5500, l_mdssm_session_count, l_mdssm_session_add, l_mdssm_session_remove, l_mdssm_last, }; /* * session */ class Session : public RefCountedObject { // -- state etc -- public: /* <-- closed <------------+ ^ | | | v | killing <-- opening <----+ | ^ | | | | v | | stale <--> open --> closing ---+ + additional dimension of 'importing' (with counter) */ enum { STATE_CLOSED = 0, STATE_OPENING = 1, // journaling open STATE_OPEN = 2, STATE_CLOSING = 3, // journaling close STATE_STALE = 4, STATE_KILLING = 5 }; const char *get_state_name(int s) const { switch (s) { case STATE_CLOSED: return "closed"; case STATE_OPENING: return "opening"; case STATE_OPEN: return "open"; case STATE_CLOSING: return "closing"; case STATE_STALE: return "stale"; case STATE_KILLING: return "killing"; default: return "???"; } } private: int state; uint64_t state_seq; int importing_count; friend class SessionMap; // Human (friendly) name is soft state generated from client metadata void _update_human_name(); std::string human_name; // Versions in this session was projected: used to verify // that appropriate mark_dirty calls follow. std::deque projected; public: void push_pv(version_t pv) { assert(projected.empty() || projected.back() != pv); projected.push_back(pv); } void pop_pv(version_t v) { assert(!projected.empty()); assert(projected.front() == v); projected.pop_front(); } int get_state() const { return state; } void set_state(int new_state) { if (state != new_state) { state = new_state; state_seq++; } } void decode(bufferlist::iterator &p); void set_client_metadata(std::map const &meta); std::string get_human_name() const {return human_name;} // Ephemeral state for tracking progress of capability recalls utime_t recalled_at; // When was I asked to SESSION_RECALL? utime_t last_recall_sent; uint32_t recall_count; // How many caps was I asked to SESSION_RECALL? uint32_t recall_release_count; // How many caps have I actually revoked? session_info_t info; ///< durable bits MDSAuthCaps auth_caps; ConnectionRef connection; xlist::item item_session_list; list preopen_out_queue; ///< messages for client, queued before they connect elist requests; size_t get_request_count(); interval_set pending_prealloc_inos; // journaling prealloc, will be added to prealloc_inos void notify_cap_release(size_t n_caps); void notify_recall_sent(const size_t new_limit); void clear_recalled_at(); inodeno_t next_ino() const { if (info.prealloc_inos.empty()) return 0; return info.prealloc_inos.range_start(); } inodeno_t take_ino(inodeno_t ino = 0) { assert(!info.prealloc_inos.empty()); if (ino) { if (info.prealloc_inos.contains(ino)) info.prealloc_inos.erase(ino); else ino = 0; } if (!ino) { ino = info.prealloc_inos.range_start(); info.prealloc_inos.erase(ino); } info.used_inos.insert(ino, 1); return ino; } int get_num_projected_prealloc_inos() const { return info.prealloc_inos.size() + pending_prealloc_inos.size(); } client_t get_client() const { return info.get_client(); } const char *get_state_name() const { return get_state_name(state); } uint64_t get_state_seq() const { return state_seq; } bool is_closed() const { return state == STATE_CLOSED; } bool is_opening() const { return state == STATE_OPENING; } bool is_open() const { return state == STATE_OPEN; } bool is_closing() const { return state == STATE_CLOSING; } bool is_stale() const { return state == STATE_STALE; } bool is_killing() const { return state == STATE_KILLING; } void inc_importing() { ++importing_count; } void dec_importing() { assert(importing_count > 0); --importing_count; } bool is_importing() const { return importing_count > 0; } // -- caps -- private: version_t cap_push_seq; // cap push seq # map > waitfor_flush; // flush session messages public: xlist caps; // inodes with caps; front=most recently used xlist leases; // metadata leases to clients utime_t last_cap_renew; public: version_t inc_push_seq() { return ++cap_push_seq; } version_t get_push_seq() const { return cap_push_seq; } version_t wait_for_flush(MDSInternalContextBase* c) { waitfor_flush[get_push_seq()].push_back(c); return get_push_seq(); } void finish_flush(version_t seq, list& ls) { while (!waitfor_flush.empty()) { if (waitfor_flush.begin()->first > seq) break; ls.splice(ls.end(), waitfor_flush.begin()->second); waitfor_flush.erase(waitfor_flush.begin()); } } void add_cap(Capability *cap) { caps.push_back(&cap->item_session_caps); } void touch_lease(ClientLease *r) { leases.push_back(&r->item_session_lease); } // -- leases -- uint32_t lease_seq; // -- completed requests -- private: // Has completed_requests been modified since the last time we // wrote this session out? bool completed_requests_dirty; unsigned num_trim_flushes_warnings; unsigned num_trim_requests_warnings; public: void add_completed_request(ceph_tid_t t, inodeno_t created) { info.completed_requests[t] = created; completed_requests_dirty = true; } bool trim_completed_requests(ceph_tid_t mintid) { // trim bool erased_any = false; while (!info.completed_requests.empty() && (mintid == 0 || info.completed_requests.begin()->first < mintid)) { info.completed_requests.erase(info.completed_requests.begin()); erased_any = true; } if (erased_any) { completed_requests_dirty = true; } return erased_any; } bool have_completed_request(ceph_tid_t tid, inodeno_t *pcreated) const { map::const_iterator p = info.completed_requests.find(tid); if (p == info.completed_requests.end()) return false; if (pcreated) *pcreated = p->second; return true; } void add_completed_flush(ceph_tid_t tid) { info.completed_flushes.insert(tid); } bool trim_completed_flushes(ceph_tid_t mintid) { bool erased_any = false; while (!info.completed_flushes.empty() && (mintid == 0 || *info.completed_flushes.begin() < mintid)) { info.completed_flushes.erase(info.completed_flushes.begin()); erased_any = true; } if (erased_any) { completed_requests_dirty = true; } return erased_any; } bool have_completed_flush(ceph_tid_t tid) const { return info.completed_flushes.count(tid); } unsigned get_num_completed_flushes() const { return info.completed_flushes.size(); } unsigned get_num_trim_flushes_warnings() const { return num_trim_flushes_warnings; } void inc_num_trim_flushes_warnings() { ++num_trim_flushes_warnings; } void reset_num_trim_flushes_warnings() { num_trim_flushes_warnings = 0; } unsigned get_num_completed_requests() const { return info.completed_requests.size(); } unsigned get_num_trim_requests_warnings() const { return num_trim_requests_warnings; } void inc_num_trim_requests_warnings() { ++num_trim_requests_warnings; } void reset_num_trim_requests_warnings() { num_trim_requests_warnings = 0; } bool has_dirty_completed_requests() const { return completed_requests_dirty; } void clear_dirty_completed_requests() { completed_requests_dirty = false; } int check_access(CInode *in, unsigned mask, int caller_uid, int caller_gid, const vector *gid_list, int new_uid, int new_gid); Session() : state(STATE_CLOSED), state_seq(0), importing_count(0), recall_count(0), recall_release_count(0), auth_caps(g_ceph_context), connection(NULL), item_session_list(this), requests(0), // member_offset passed to front() manually cap_push_seq(0), lease_seq(0), completed_requests_dirty(false), num_trim_flushes_warnings(0), num_trim_requests_warnings(0) { } ~Session() override { assert(!item_session_list.is_on_list()); while (!preopen_out_queue.empty()) { preopen_out_queue.front()->put(); preopen_out_queue.pop_front(); } } void clear() { pending_prealloc_inos.clear(); info.clear_meta(); cap_push_seq = 0; last_cap_renew = utime_t(); } }; class SessionFilter { protected: // First is whether to filter, second is filter value std::pair reconnecting; public: std::map metadata; std::string auth_name; std::string state; int64_t id; SessionFilter() : reconnecting(false, false), id(0) {} bool match( const Session &session, std::function is_reconnecting) const; int parse(const std::vector &args, std::stringstream *ss); void set_reconnecting(bool v) { reconnecting.first = true; reconnecting.second = v; } }; /* * session map */ class MDSRank; /** * Encapsulate the serialized state associated with SessionMap. Allows * encode/decode outside of live MDS instance. */ class SessionMapStore { protected: version_t version; ceph::unordered_map session_map; PerfCounters *logger; public: mds_rank_t rank; version_t get_version() const {return version;} virtual void encode_header(bufferlist *header_bl); virtual void decode_header(bufferlist &header_bl); virtual void decode_values(std::map &session_vals); virtual void decode_legacy(bufferlist::iterator& blp); void dump(Formatter *f) const; void set_rank(mds_rank_t r) { rank = r; } Session* get_or_add_session(const entity_inst_t& i) { Session *s; auto session_map_entry = session_map.find(i.name); if (session_map_entry != session_map.end()) { s = session_map_entry->second; } else { s = session_map[i.name] = new Session; s->info.inst = i; s->last_cap_renew = ceph_clock_now(); if (logger) { logger->set(l_mdssm_session_count, session_map.size()); logger->inc(l_mdssm_session_add); } } return s; } static void generate_test_instances(list& ls); void reset_state() { session_map.clear(); } SessionMapStore() : version(0), logger(nullptr), rank(MDS_RANK_NONE) {} virtual ~SessionMapStore() {}; }; class SessionMap : public SessionMapStore { public: MDSRank *mds; protected: version_t projected, committing, committed; public: map* > by_state; uint64_t set_state(Session *session, int state); map > commit_waiters; explicit SessionMap(MDSRank *m) : mds(m), projected(0), committing(0), committed(0), loaded_legacy(false) { } ~SessionMap() override { for (auto p : by_state) delete p.second; if (logger) { g_ceph_context->get_perfcounters_collection()->remove(logger); } delete logger; } void register_perfcounters(); void set_version(const version_t v) { version = projected = v; } void set_projected(const version_t v) { projected = v; } version_t get_projected() const { return projected; } version_t get_committed() const { return committed; } version_t get_committing() const { return committing; } // sessions void decode_legacy(bufferlist::iterator& blp) override; bool empty() const { return session_map.empty(); } const ceph::unordered_map &get_sessions() const { return session_map; } bool is_any_state(int state) const { map* >::const_iterator p = by_state.find(state); if (p == by_state.end() || p->second->empty()) return false; return true; } bool have_unclosed_sessions() const { return is_any_state(Session::STATE_OPENING) || is_any_state(Session::STATE_OPEN) || is_any_state(Session::STATE_CLOSING) || is_any_state(Session::STATE_STALE) || is_any_state(Session::STATE_KILLING); } bool have_session(entity_name_t w) const { return session_map.count(w); } Session* get_session(entity_name_t w) { auto session_map_entry = session_map.find(w); return (session_map_entry != session_map.end() ? session_map_entry-> second : nullptr); } const Session* get_session(entity_name_t w) const { ceph::unordered_map::const_iterator p = session_map.find(w); if (p == session_map.end()) { return NULL; } else { return p->second; } } void add_session(Session *s); void remove_session(Session *s); void touch_session(Session *session); Session *get_oldest_session(int state) { auto by_state_entry = by_state.find(state); if (by_state_entry == by_state.end() || by_state_entry->second->empty()) return 0; return by_state_entry->second->front(); } void dump(); void get_client_set(set& s) { for (ceph::unordered_map::iterator p = session_map.begin(); p != session_map.end(); ++p) if (p->second->info.inst.name.is_client()) s.insert(p->second->info.inst.name.num()); } void get_client_session_set(set& s) const { for (ceph::unordered_map::const_iterator p = session_map.begin(); p != session_map.end(); ++p) if (p->second->info.inst.name.is_client()) s.insert(p->second); } void open_sessions(map& client_map) { for (map::iterator p = client_map.begin(); p != client_map.end(); ++p) { Session *s = get_or_add_session(p->second); set_state(s, Session::STATE_OPEN); version++; } } // helpers entity_inst_t& get_inst(entity_name_t w) { assert(session_map.count(w)); return session_map[w]->info.inst; } version_t inc_push_seq(client_t client) { return get_session(entity_name_t::CLIENT(client.v))->inc_push_seq(); } version_t get_push_seq(client_t client) { return get_session(entity_name_t::CLIENT(client.v))->get_push_seq(); } bool have_completed_request(metareqid_t rid) { Session *session = get_session(rid.name); return session && session->have_completed_request(rid.tid, NULL); } void trim_completed_requests(entity_name_t c, ceph_tid_t tid) { Session *session = get_session(c); assert(session); session->trim_completed_requests(tid); } void wipe(); void wipe_ino_prealloc(); // -- loading, saving -- inodeno_t ino; list waiting_for_load; object_t get_object_name() const; void load(MDSInternalContextBase *onload); void _load_finish( int operation_r, int header_r, int values_r, bool first, bufferlist &header_bl, std::map &session_vals, bool more_session_vals); void load_legacy(); void _load_legacy_finish(int r, bufferlist &bl); void save(MDSInternalContextBase *onsave, version_t needv=0); void _save_finish(version_t v); protected: std::set dirty_sessions; std::set null_sessions; bool loaded_legacy; void _mark_dirty(Session *session); public: /** * Advance the version, and mark this session * as dirty within the new version. * * Dirty means journalled but needing writeback * to the backing store. Must have called * mark_projected previously for this session. */ void mark_dirty(Session *session); /** * Advance the projected version, and mark this * session as projected within the new version * * Projected means the session is updated in memory * but we're waiting for the journal write of the update * to finish. Must subsequently call mark_dirty * for sessions in the same global order as calls * to mark_projected. */ version_t mark_projected(Session *session); /** * During replay, advance versions to account * for a session modification, and mark the * session dirty. */ void replay_dirty_session(Session *session); /** * During replay, if a session no longer present * would have consumed a version, advance `version` * and `projected` to account for that. */ void replay_advance_version(); /** * For these session IDs, if a session exists with this ID, and it has * dirty completed_requests, then persist it immediately * (ahead of usual project/dirty versioned writes * of the map). */ void save_if_dirty(const std::set &tgt_sessions, MDSGatherBuilder *gather_bld); }; std::ostream& operator<<(std::ostream &out, const Session &s); #endif