summaryrefslogtreecommitdiffstats
path: root/src/mds/OpenFileTable.cc
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-27 18:24:20 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-27 18:24:20 +0000
commit483eb2f56657e8e7f419ab1a4fab8dce9ade8609 (patch)
treee5d88d25d870d5dedacb6bbdbe2a966086a0a5cf /src/mds/OpenFileTable.cc
parentInitial commit. (diff)
downloadceph-483eb2f56657e8e7f419ab1a4fab8dce9ade8609.tar.xz
ceph-483eb2f56657e8e7f419ab1a4fab8dce9ade8609.zip
Adding upstream version 14.2.21.upstream/14.2.21upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/mds/OpenFileTable.cc')
-rw-r--r--src/mds/OpenFileTable.cc1189
1 files changed, 1189 insertions, 0 deletions
diff --git a/src/mds/OpenFileTable.cc b/src/mds/OpenFileTable.cc
new file mode 100644
index 00000000..5e0d2ba5
--- /dev/null
+++ b/src/mds/OpenFileTable.cc
@@ -0,0 +1,1189 @@
+// -*- 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) 2018 Red Hat
+ *
+ * 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.
+ *
+ */
+
+#include "acconfig.h"
+#include "mds/CInode.h"
+#include "mds/CDir.h"
+#include "mds/MDSRank.h"
+#include "mds/MDCache.h"
+#include "osdc/Objecter.h"
+#include "OpenFileTable.h"
+
+#include "common/config.h"
+#include "common/errno.h"
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_mds
+#undef dout_prefix
+#define dout_prefix _prefix(_dout, mds)
+static ostream& _prefix(std::ostream *_dout, MDSRank *mds) {
+ return *_dout << "mds." << mds->get_nodeid() << ".openfiles ";
+}
+
+void OpenFileTable::get_ref(CInode *in)
+{
+ do {
+ auto p = anchor_map.find(in->ino());
+ if (p != anchor_map.end()) {
+ ceph_assert(in->state_test(CInode::STATE_TRACKEDBYOFT));
+ ceph_assert(p->second.nref > 0);
+ p->second.nref++;
+ break;
+ }
+
+ CDentry *dn = in->get_parent_dn();
+ CInode *pin = dn ? dn->get_dir()->get_inode() : nullptr;
+
+ auto ret = anchor_map.emplace(std::piecewise_construct, std::forward_as_tuple(in->ino()),
+ std::forward_as_tuple(in->ino(), (pin ? pin->ino() : inodeno_t(0)),
+ (dn ? dn->get_name() : string()), in->d_type(), 1));
+ ceph_assert(ret.second == true);
+ in->state_set(CInode::STATE_TRACKEDBYOFT);
+
+ auto ret1 = dirty_items.emplace(in->ino(), (int)DIRTY_NEW);
+ if (!ret1.second) {
+ int omap_idx = ret1.first->second;
+ ceph_assert(omap_idx >= 0);
+ ret.first->second.omap_idx = omap_idx;
+ }
+
+ in = pin;
+ } while (in);
+}
+
+void OpenFileTable::put_ref(CInode *in)
+{
+ do {
+ ceph_assert(in->state_test(CInode::STATE_TRACKEDBYOFT));
+ auto p = anchor_map.find(in->ino());
+ ceph_assert(p != anchor_map.end());
+ ceph_assert(p->second.nref > 0);
+
+ if (p->second.nref > 1) {
+ p->second.nref--;
+ break;
+ }
+
+ CDentry *dn = in->get_parent_dn();
+ CInode *pin = dn ? dn->get_dir()->get_inode() : nullptr;
+ if (dn) {
+ ceph_assert(p->second.dirino == pin->ino());
+ ceph_assert(p->second.d_name == dn->get_name());
+ } else {
+ ceph_assert(p->second.dirino == inodeno_t(0));
+ ceph_assert(p->second.d_name == "");
+ }
+
+ int omap_idx = p->second.omap_idx;
+ anchor_map.erase(p);
+ in->state_clear(CInode::STATE_TRACKEDBYOFT);
+
+ auto ret = dirty_items.emplace(in->ino(), omap_idx);
+ if (!ret.second) {
+ if (ret.first->second == DIRTY_NEW) {
+ ceph_assert(omap_idx < 0);
+ dirty_items.erase(ret.first);
+ } else {
+ ceph_assert(omap_idx >= 0);
+ ret.first->second = omap_idx;
+ }
+ }
+
+ in = pin;
+ } while (in);
+}
+
+void OpenFileTable::add_inode(CInode *in)
+{
+ dout(10) << __func__ << " " << *in << dendl;
+ if (!in->is_dir()) {
+ auto p = anchor_map.find(in->ino());
+ ceph_assert(p == anchor_map.end());
+ }
+ get_ref(in);
+}
+
+void OpenFileTable::remove_inode(CInode *in)
+{
+ dout(10) << __func__ << " " << *in << dendl;
+ if (!in->is_dir()) {
+ auto p = anchor_map.find(in->ino());
+ ceph_assert(p != anchor_map.end());
+ ceph_assert(p->second.nref == 1);
+ }
+ put_ref(in);
+}
+
+void OpenFileTable::add_dirfrag(CDir *dir)
+{
+ dout(10) << __func__ << " " << *dir << dendl;
+ ceph_assert(!dir->state_test(CDir::STATE_TRACKEDBYOFT));
+ dir->state_set(CDir::STATE_TRACKEDBYOFT);
+ auto ret = dirfrags.insert(dir->dirfrag());
+ ceph_assert(ret.second);
+ get_ref(dir->get_inode());
+ dirty_items.emplace(dir->ino(), (int)DIRTY_UNDEF);
+}
+
+void OpenFileTable::remove_dirfrag(CDir *dir)
+{
+ dout(10) << __func__ << " " << *dir << dendl;
+ ceph_assert(dir->state_test(CDir::STATE_TRACKEDBYOFT));
+ dir->state_clear(CDir::STATE_TRACKEDBYOFT);
+ auto p = dirfrags.find(dir->dirfrag());
+ ceph_assert(p != dirfrags.end());
+ dirfrags.erase(p);
+ dirty_items.emplace(dir->ino(), (int)DIRTY_UNDEF);
+ put_ref(dir->get_inode());
+}
+
+void OpenFileTable::notify_link(CInode *in)
+{
+ dout(10) << __func__ << " " << *in << dendl;
+ auto p = anchor_map.find(in->ino());
+ ceph_assert(p != anchor_map.end());
+ ceph_assert(p->second.nref > 0);
+ ceph_assert(p->second.dirino == inodeno_t(0));
+ ceph_assert(p->second.d_name == "");
+
+ CDentry *dn = in->get_parent_dn();
+ CInode *pin = dn->get_dir()->get_inode();
+
+ p->second.dirino = pin->ino();
+ p->second.d_name = dn->get_name();
+ dirty_items.emplace(in->ino(), (int)DIRTY_UNDEF);
+
+ get_ref(pin);
+}
+
+void OpenFileTable::notify_unlink(CInode *in)
+{
+ dout(10) << __func__ << " " << *in << dendl;
+ auto p = anchor_map.find(in->ino());
+ ceph_assert(p != anchor_map.end());
+ ceph_assert(p->second.nref > 0);
+
+ CDentry *dn = in->get_parent_dn();
+ CInode *pin = dn->get_dir()->get_inode();
+ ceph_assert(p->second.dirino == pin->ino());
+ ceph_assert(p->second.d_name == dn->get_name());
+
+ p->second.dirino = inodeno_t(0);
+ p->second.d_name = "";
+ dirty_items.emplace(in->ino(), (int)DIRTY_UNDEF);
+
+ put_ref(pin);
+}
+
+object_t OpenFileTable::get_object_name(unsigned idx) const
+{
+ char s[30];
+ snprintf(s, sizeof(s), "mds%d_openfiles.%x", int(mds->get_nodeid()), idx);
+ return object_t(s);
+}
+
+void OpenFileTable::_encode_header(bufferlist &bl, int j_state)
+{
+ std::string_view magic = CEPH_FS_ONDISK_MAGIC;
+ encode(magic, bl);
+ ENCODE_START(1, 1, bl);
+ encode(omap_version, bl);
+ encode(omap_num_objs, bl);
+ encode((__u8)j_state, bl);
+ ENCODE_FINISH(bl);
+}
+
+class C_IO_OFT_Save : public MDSIOContextBase {
+protected:
+ OpenFileTable *oft;
+ uint64_t log_seq;
+ MDSContext *fin;
+ MDSRank *get_mds() override { return oft->mds; }
+public:
+ C_IO_OFT_Save(OpenFileTable *t, uint64_t s, MDSContext *c) :
+ oft(t), log_seq(s), fin(c) {}
+ void finish(int r) {
+ oft->_commit_finish(r, log_seq, fin);
+ }
+ void print(ostream& out) const override {
+ out << "openfiles_save";
+ }
+};
+
+void OpenFileTable::_commit_finish(int r, uint64_t log_seq, MDSContext *fin)
+{
+ dout(10) << __func__ << " log_seq " << log_seq << dendl;
+ if (r < 0) {
+ mds->handle_write_error(r);
+ return;
+ }
+
+ ceph_assert(log_seq <= committing_log_seq);
+ ceph_assert(log_seq >= committed_log_seq);
+ committed_log_seq = log_seq;
+ num_pending_commit--;
+
+ if (fin)
+ fin->complete(r);
+}
+
+class C_IO_OFT_Journal : public MDSIOContextBase {
+protected:
+ OpenFileTable *oft;
+ uint64_t log_seq;
+ MDSContext *fin;
+ std::map<unsigned, std::vector<ObjectOperation> > ops_map;
+ MDSRank *get_mds() override { return oft->mds; }
+public:
+ C_IO_OFT_Journal(OpenFileTable *t, uint64_t s, MDSContext *c,
+ std::map<unsigned, std::vector<ObjectOperation> >& ops) :
+ oft(t), log_seq(s), fin(c) {
+ ops_map.swap(ops);
+ }
+ void finish(int r) {
+ oft->_journal_finish(r, log_seq, fin, ops_map);
+ }
+ void print(ostream& out) const override {
+ out << "openfiles_journal";
+ }
+};
+
+void OpenFileTable::_journal_finish(int r, uint64_t log_seq, MDSContext *c,
+ std::map<unsigned, std::vector<ObjectOperation> >& ops_map)
+{
+ dout(10) << __func__ << " log_seq " << log_seq << dendl;
+ if (r < 0) {
+ mds->handle_write_error(r);
+ return;
+ }
+
+ C_GatherBuilder gather(g_ceph_context,
+ new C_OnFinisher(new C_IO_OFT_Save(this, log_seq, c),
+ mds->finisher));
+ SnapContext snapc;
+ object_locator_t oloc(mds->mdsmap->get_metadata_pool());
+ for (auto& it : ops_map) {
+ object_t oid = get_object_name(it.first);
+ for (auto& op : it.second) {
+ mds->objecter->mutate(oid, oloc, op, snapc, ceph::real_clock::now(),
+ 0, gather.new_sub());
+ }
+ }
+ gather.activate();
+
+ journal_state = JOURNAL_NONE;
+ return;
+}
+
+void OpenFileTable::commit(MDSContext *c, uint64_t log_seq, int op_prio)
+{
+ dout(10) << __func__ << " log_seq " << log_seq << dendl;
+
+ ceph_assert(num_pending_commit == 0);
+ num_pending_commit++;
+ ceph_assert(log_seq >= committing_log_seq);
+ committing_log_seq = log_seq;
+
+ omap_version++;
+
+ C_GatherBuilder gather(g_ceph_context);
+
+ SnapContext snapc;
+ object_locator_t oloc(mds->mdsmap->get_metadata_pool());
+
+ const unsigned max_write_size = mds->mdcache->max_dir_commit_size;
+
+ struct omap_update_ctl {
+ unsigned write_size = 0;
+ unsigned journal_idx = 0;
+ bool clear = false;
+ std::map<string, bufferlist> to_update, journaled_update;
+ std::set<string> to_remove, journaled_remove;
+ };
+ std::vector<omap_update_ctl> omap_updates(omap_num_objs);
+
+ using ceph::encode;
+ auto journal_func = [&](unsigned idx) {
+ auto& ctl = omap_updates.at(idx);
+
+ ObjectOperation op;
+ op.priority = op_prio;
+
+ if (ctl.clear) {
+ ctl.clear = false;
+ op.omap_clear();
+ op.set_last_op_flags(CEPH_OSD_OP_FLAG_FAILOK);
+ }
+
+ if (ctl.journal_idx == 0) {
+ if (journal_state == JOURNAL_NONE)
+ journal_state = JOURNAL_START;
+ else
+ ceph_assert(journal_state == JOURNAL_START);
+
+ bufferlist header;
+ _encode_header(header, journal_state);
+ op.omap_set_header(header);
+ }
+
+ bufferlist bl;
+ encode(omap_version, bl);
+ encode(ctl.to_update, bl);
+ encode(ctl.to_remove, bl);
+
+ char key[32];
+ snprintf(key, sizeof(key), "_journal.%x", ctl.journal_idx++);
+ std::map<string, bufferlist> tmp_map;
+ tmp_map[key].swap(bl);
+ op.omap_set(tmp_map);
+
+ object_t oid = get_object_name(idx);
+ mds->objecter->mutate(oid, oloc, op, snapc, ceph::real_clock::now(), 0,
+ gather.new_sub());
+
+#ifdef HAVE_STDLIB_MAP_SPLICING
+ ctl.journaled_update.merge(ctl.to_update);
+ ctl.journaled_remove.merge(ctl.to_remove);
+#else
+ ctl.journaled_update.insert(make_move_iterator(begin(ctl.to_update)),
+ make_move_iterator(end(ctl.to_update)));
+ ctl.journaled_remove.insert(make_move_iterator(begin(ctl.to_remove)),
+ make_move_iterator(end(ctl.to_remove)));
+#endif
+ ctl.to_update.clear();
+ ctl.to_remove.clear();
+ };
+
+ std::map<unsigned, std::vector<ObjectOperation> > ops_map;
+
+ auto create_op_func = [&](unsigned idx, bool update_header) {
+ auto& ctl = omap_updates.at(idx);
+
+ auto& op_vec = ops_map[idx];
+ op_vec.resize(op_vec.size() + 1);
+ ObjectOperation& op = op_vec.back();
+ op.priority = op_prio;
+
+ if (ctl.clear) {
+ ctl.clear = false;
+ op.omap_clear();
+ op.set_last_op_flags(CEPH_OSD_OP_FLAG_FAILOK);
+ }
+
+ if (update_header) {
+ bufferlist header;
+ _encode_header(header, journal_state);
+ op.omap_set_header(header);
+ }
+
+ if (!ctl.to_update.empty()) {
+ op.omap_set(ctl.to_update);
+ ctl.to_update.clear();
+ }
+ if (!ctl.to_remove.empty()) {
+ op.omap_rm_keys(ctl.to_remove);
+ ctl.to_remove.clear();
+ }
+ };
+
+ auto submit_ops_func = [&]() {
+ gather.set_finisher(new C_OnFinisher(new C_IO_OFT_Save(this, log_seq, c),
+ mds->finisher));
+ for (auto& it : ops_map) {
+ object_t oid = get_object_name(it.first);
+ for (auto& op : it.second) {
+ mds->objecter->mutate(oid, oloc, op, snapc, ceph::real_clock::now(),
+ 0, gather.new_sub());
+ }
+ }
+ gather.activate();
+ };
+
+ bool first_commit = !loaded_anchor_map.empty();
+
+ unsigned first_free_idx = 0;
+ unsigned old_num_objs = omap_num_objs;
+ if (omap_num_objs == 0) {
+ omap_num_objs = 1;
+ omap_num_items.resize(omap_num_objs);
+ omap_updates.resize(omap_num_objs);
+ omap_updates.back().clear = true;
+ }
+
+ for (auto& it : dirty_items) {
+ frag_vec_t frags;
+ auto p = anchor_map.find(it.first);
+ if (p != anchor_map.end()) {
+ for (auto q = dirfrags.lower_bound(dirfrag_t(it.first, 0));
+ q != dirfrags.end() && q->ino == it.first;
+ ++q)
+ frags.push_back(q->frag);
+ }
+
+ if (first_commit) {
+ auto q = loaded_anchor_map.find(it.first);
+ if (q != loaded_anchor_map.end()) {
+ ceph_assert(p != anchor_map.end());
+ p->second.omap_idx = q->second.omap_idx;
+ bool same = p->second == q->second;
+ if (same) {
+ auto r = loaded_dirfrags.lower_bound(dirfrag_t(it.first, 0));
+ for (const auto& fg : frags) {
+ if (r == loaded_dirfrags.end() || !(*r == dirfrag_t(it.first, fg))) {
+ same = false;
+ break;
+ }
+ ++r;
+ }
+ if (same && r != loaded_dirfrags.end() && r->ino == it.first)
+ same = false;
+ }
+ loaded_anchor_map.erase(q);
+ if (same)
+ continue;
+ }
+ }
+
+ char key[32];
+ int len = snprintf(key, sizeof(key), "%llx", (unsigned long long)it.first.val);
+
+ int omap_idx;
+ if (p != anchor_map.end()) {
+ omap_idx = p->second.omap_idx;
+ if (omap_idx < 0) {
+ ceph_assert(it.second == DIRTY_NEW);
+ // find omap object to store the key
+ for (unsigned i = first_free_idx; i < omap_num_objs; i++) {
+ if (omap_num_items[i] < MAX_ITEMS_PER_OBJ) {
+ omap_idx = i;
+ break;
+ }
+ }
+ if (omap_idx < 0) {
+ ++omap_num_objs;
+ ceph_assert(omap_num_objs <= MAX_OBJECTS);
+ omap_num_items.resize(omap_num_objs);
+ omap_updates.resize(omap_num_objs);
+ omap_updates.back().clear = true;
+ omap_idx = omap_num_objs - 1;
+ }
+ first_free_idx = omap_idx;
+
+ p->second.omap_idx = omap_idx;
+ ++omap_num_items[omap_idx];
+ }
+ } else {
+ omap_idx = it.second;
+ unsigned& count = omap_num_items.at(omap_idx);
+ ceph_assert(count > 0);
+ --count;
+ if ((unsigned)omap_idx < first_free_idx && count < MAX_ITEMS_PER_OBJ)
+ first_free_idx = omap_idx;
+ }
+ auto& ctl = omap_updates.at(omap_idx);
+
+ if (p != anchor_map.end()) {
+ bufferlist bl;
+ encode(p->second, bl);
+ encode(frags, bl);
+
+ ctl.write_size += bl.length() + len + 2 * sizeof(__u32);
+ ctl.to_update[key].swap(bl);
+ } else {
+ ctl.write_size += len + sizeof(__u32);
+ ctl.to_remove.emplace(key);
+ }
+
+ if (ctl.write_size >= max_write_size) {
+ journal_func(omap_idx);
+ ctl.write_size = 0;
+ }
+ }
+
+ dirty_items.clear();
+
+ if (first_commit) {
+ for (auto& it : loaded_anchor_map) {
+ char key[32];
+ int len = snprintf(key, sizeof(key), "%llx", (unsigned long long)it.first.val);
+
+ int omap_idx = it.second.omap_idx;
+ unsigned& count = omap_num_items.at(omap_idx);
+ ceph_assert(count > 0);
+ --count;
+
+ auto& ctl = omap_updates.at(omap_idx);
+ ctl.write_size += len + sizeof(__u32);
+ ctl.to_remove.emplace(key);
+
+ if (ctl.write_size >= max_write_size) {
+ journal_func(omap_idx);
+ ctl.write_size = 0;
+ }
+ }
+ loaded_anchor_map.clear();
+ loaded_dirfrags.clear();
+ }
+
+ {
+ size_t total_items = 0;
+ unsigned used_objs = 1;
+ std::list<unsigned> objs_to_write;
+ bool journaled = false;
+ for (unsigned i = 0; i < omap_num_objs; i++) {
+ total_items += omap_num_items[i];
+ if (omap_updates[i].journal_idx)
+ journaled = true;
+ else if (omap_updates[i].write_size)
+ objs_to_write.push_back(i);
+
+ if (omap_num_items[i] > 0)
+ used_objs = i + 1;
+ }
+ ceph_assert(total_items == anchor_map.size());
+ // adjust omap object count
+ if (used_objs < omap_num_objs) {
+ omap_num_objs = used_objs;
+ omap_num_items.resize(omap_num_objs);
+ }
+ // skip journal if only one osd request is required and object count
+ // does not change.
+ if (!journaled && old_num_objs == omap_num_objs &&
+ objs_to_write.size() <= 1) {
+ ceph_assert(journal_state == JOURNAL_NONE);
+ ceph_assert(!gather.has_subs());
+
+ unsigned omap_idx = objs_to_write.empty() ? 0 : objs_to_write.front();
+ create_op_func(omap_idx, true);
+ submit_ops_func();
+ return;
+ }
+ }
+
+ for (unsigned omap_idx = 0; omap_idx < omap_updates.size(); omap_idx++) {
+ auto& ctl = omap_updates[omap_idx];
+ if (ctl.write_size > 0) {
+ journal_func(omap_idx);
+ ctl.write_size = 0;
+ }
+ }
+
+ if (journal_state == JOURNAL_START) {
+ ceph_assert(gather.has_subs());
+ journal_state = JOURNAL_FINISH;
+ } else {
+ // only object count changes
+ ceph_assert(journal_state == JOURNAL_NONE);
+ ceph_assert(!gather.has_subs());
+ }
+
+ for (unsigned omap_idx = 0; omap_idx < omap_updates.size(); omap_idx++) {
+ auto& ctl = omap_updates[omap_idx];
+ ceph_assert(ctl.to_update.empty() && ctl.to_remove.empty());
+ if (ctl.journal_idx == 0)
+ ceph_assert(ctl.journaled_update.empty() && ctl.journaled_remove.empty());
+
+ bool first = true;
+ for (auto& it : ctl.journaled_update) {
+ ctl.write_size += it.first.length() + it.second.length() + 2 * sizeof(__u32);
+ ctl.to_update[it.first].swap(it.second);
+ if (ctl.write_size >= max_write_size) {
+ create_op_func(omap_idx, first);
+ ctl.write_size = 0;
+ first = false;
+ }
+ }
+
+ for (auto& key : ctl.journaled_remove) {
+ ctl.write_size += key.length() + sizeof(__u32);
+ ctl.to_remove.emplace(key);
+ if (ctl.write_size >= max_write_size) {
+ create_op_func(omap_idx, first);
+ ctl.write_size = 0;
+ first = false;
+ }
+ }
+
+ for (unsigned i = 0; i < ctl.journal_idx; ++i) {
+ char key[32];
+ snprintf(key, sizeof(key), "_journal.%x", i);
+ ctl.to_remove.emplace(key);
+ }
+
+ // update first object's omap header if object count changes
+ if (ctl.clear ||
+ ctl.journal_idx > 0 ||
+ (omap_idx == 0 && old_num_objs != omap_num_objs))
+ create_op_func(omap_idx, first);
+ }
+
+ ceph_assert(!ops_map.empty());
+ if (journal_state == JOURNAL_FINISH) {
+ gather.set_finisher(new C_OnFinisher(new C_IO_OFT_Journal(this, log_seq, c, ops_map),
+ mds->finisher));
+ gather.activate();
+ } else {
+ submit_ops_func();
+ }
+}
+
+class C_IO_OFT_Load : public MDSIOContextBase {
+protected:
+ OpenFileTable *oft;
+ MDSRank *get_mds() override { return oft->mds; }
+
+public:
+ int header_r = 0; //< Return value from OMAP header read
+ int values_r = 0; //< Return value from OMAP value read
+ bufferlist header_bl;
+ std::map<std::string, bufferlist> values;
+ unsigned index;
+ bool first;
+ bool more = false;
+
+ C_IO_OFT_Load(OpenFileTable *t, unsigned i, bool f) :
+ oft(t), index(i), first(f) {}
+ void finish(int r) override {
+ oft->_load_finish(r, header_r, values_r, index, first, more, header_bl, values);
+ }
+ void print(ostream& out) const override {
+ out << "openfiles_load";
+ }
+};
+
+class C_IO_OFT_Recover : public MDSIOContextBase {
+protected:
+ OpenFileTable *oft;
+ MDSRank *get_mds() override { return oft->mds; }
+public:
+ C_IO_OFT_Recover(OpenFileTable *t) : oft(t) {}
+ void finish(int r) override {
+ oft->_recover_finish(r);
+ }
+ void print(ostream& out) const override {
+ out << "openfiles_recover";
+ }
+};
+
+void OpenFileTable::_recover_finish(int r)
+{
+ if (r < 0) {
+ derr << __func__ << " got " << cpp_strerror(r) << dendl;
+ _reset_states();
+ } else {
+ dout(10) << __func__ << ": load complete" << dendl;
+ }
+
+ journal_state = JOURNAL_NONE;
+ load_done = true;
+ finish_contexts(g_ceph_context, waiting_for_load);
+ waiting_for_load.clear();
+}
+
+void OpenFileTable::_load_finish(int op_r, int header_r, int values_r,
+ unsigned idx, bool first, bool more,
+ bufferlist &header_bl,
+ std::map<std::string, bufferlist> &values)
+{
+ using ceph::decode;
+ int err = -EINVAL;
+
+ auto decode_func = [this](unsigned idx, inodeno_t ino, bufferlist &bl) {
+ auto p = bl.cbegin();
+
+ size_t count = loaded_anchor_map.size();
+ auto it = loaded_anchor_map.emplace_hint(loaded_anchor_map.end(),
+ std::piecewise_construct,
+ std::make_tuple(ino),
+ std::make_tuple());
+ RecoveredAnchor& anchor = it->second;
+ decode(anchor, p);
+ ceph_assert(ino == anchor.ino);
+ anchor.omap_idx = idx;
+ anchor.auth = MDS_RANK_NONE;
+
+ frag_vec_t frags;
+ decode(frags, p);
+ for (const auto& fg : frags)
+ loaded_dirfrags.insert(loaded_dirfrags.end(), dirfrag_t(anchor.ino, fg));
+
+ if (loaded_anchor_map.size() > count)
+ ++omap_num_items[idx];
+ };
+
+ if (op_r < 0) {
+ derr << __func__ << " got " << cpp_strerror(op_r) << dendl;
+ err = op_r;
+ goto out;
+ }
+
+ try {
+ if (first) {
+ auto p = header_bl.cbegin();
+
+ string magic;
+ version_t version;
+ unsigned num_objs;
+ __u8 jstate;
+
+ if (header_bl.length() == 13) {
+ // obsolete format.
+ decode(version, p);
+ decode(num_objs, p);
+ decode(jstate, p);
+ } else {
+ decode(magic, p);
+ if (magic != CEPH_FS_ONDISK_MAGIC) {
+ std::ostringstream oss;
+ oss << "invalid magic '" << magic << "'";
+ throw buffer::malformed_input(oss.str());
+ }
+
+ DECODE_START(1, p);
+ decode(version, p);
+ decode(num_objs, p);
+ decode(jstate, p);
+ DECODE_FINISH(p);
+ }
+
+ if (num_objs > MAX_OBJECTS) {
+ std::ostringstream oss;
+ oss << "invalid object count '" << num_objs << "'";
+ throw buffer::malformed_input(oss.str());
+ }
+ if (jstate > JOURNAL_FINISH) {
+ std::ostringstream oss;
+ oss << "invalid journal state '" << jstate << "'";
+ throw buffer::malformed_input(oss.str());
+ }
+
+ if (version > omap_version) {
+ omap_version = version;
+ omap_num_objs = num_objs;
+ omap_num_items.resize(omap_num_objs);
+ journal_state = jstate;
+ } else if (version == omap_version) {
+ ceph_assert(omap_num_objs == num_objs);
+ if (jstate > journal_state)
+ journal_state = jstate;
+ }
+ }
+
+ for (auto& it : values) {
+ if (it.first.compare(0, 9, "_journal.") == 0) {
+ if (idx >= loaded_journals.size())
+ loaded_journals.resize(idx + 1);
+
+ if (journal_state == JOURNAL_FINISH) {
+ loaded_journals[idx][it.first].swap(it.second);
+ } else { // incomplete journal
+ loaded_journals[idx][it.first].length();
+ }
+ continue;
+ }
+
+ inodeno_t ino;
+ sscanf(it.first.c_str(), "%llx", (unsigned long long*)&ino.val);
+ decode_func(idx, ino, it.second);
+ }
+ } catch (buffer::error &e) {
+ derr << __func__ << ": corrupted header/values: " << e.what() << dendl;
+ goto out;
+ }
+
+ if (more || idx + 1 < omap_num_objs) {
+ // Issue another read if we're not at the end of the omap
+ std::string last_key;
+ if (more)
+ last_key = values.rbegin()->first;
+ else
+ idx++;
+ dout(10) << __func__ << ": continue to load from '" << last_key << "'" << dendl;
+ object_t oid = get_object_name(idx);
+ object_locator_t oloc(mds->mdsmap->get_metadata_pool());
+ C_IO_OFT_Load *c = new C_IO_OFT_Load(this, idx, !more);
+ ObjectOperation op;
+ if (!more)
+ op.omap_get_header(&c->header_bl, &c->header_r);
+ op.omap_get_vals(last_key, "", uint64_t(-1),
+ &c->values, &c->more, &c->values_r);
+ mds->objecter->read(oid, oloc, op, CEPH_NOSNAP, nullptr, 0,
+ new C_OnFinisher(c, mds->finisher));
+ return;
+ }
+
+ // replay journal
+ if (loaded_journals.size() > 0) {
+ dout(10) << __func__ << ": recover journal" << dendl;
+
+ C_GatherBuilder gather(g_ceph_context,
+ new C_OnFinisher(new C_IO_OFT_Recover(this),
+ mds->finisher));
+ object_locator_t oloc(mds->mdsmap->get_metadata_pool());
+ SnapContext snapc;
+
+ for (unsigned omap_idx = 0; omap_idx < loaded_journals.size(); omap_idx++) {
+ auto& loaded_journal = loaded_journals[omap_idx];
+
+ std::vector<ObjectOperation> op_vec;
+ try {
+ for (auto& it : loaded_journal) {
+ if (journal_state != JOURNAL_FINISH)
+ continue;
+ auto p = it.second.cbegin();
+ version_t version;
+ std::map<string, bufferlist> to_update;
+ std::set<string> to_remove;
+ decode(version, p);
+ if (version != omap_version)
+ continue;
+ decode(to_update, p);
+ decode(to_remove, p);
+ it.second.clear();
+
+ for (auto& q : to_update) {
+ inodeno_t ino;
+ sscanf(q.first.c_str(), "%llx", (unsigned long long*)&ino.val);
+ decode_func(omap_idx, ino, q.second);
+ }
+ for (auto& q : to_remove) {
+ inodeno_t ino;
+ sscanf(q.c_str(), "%llx",(unsigned long long*)&ino.val);
+ ceph_assert(ino.val > 0);
+ if (loaded_anchor_map.erase(ino)) {
+ unsigned& count = omap_num_items[omap_idx];
+ ceph_assert(count > 0);
+ --count;
+ }
+ auto r = loaded_dirfrags.lower_bound(dirfrag_t(ino, 0));
+ while (r != loaded_dirfrags.end() && r->ino == ino)
+ loaded_dirfrags.erase(r++);
+ }
+
+ op_vec.resize(op_vec.size() + 1);
+ ObjectOperation& op = op_vec.back();
+ op.priority = CEPH_MSG_PRIO_HIGH;
+ if (!to_update.empty())
+ op.omap_set(to_update);
+ if (!to_remove.empty())
+ op.omap_rm_keys(to_remove);
+ }
+ } catch (buffer::error &e) {
+ derr << __func__ << ": corrupted journal: " << e.what() << dendl;
+ goto out;
+ }
+
+ op_vec.resize(op_vec.size() + 1);
+ ObjectOperation& op = op_vec.back();
+ {
+ bufferlist header;
+ if (journal_state == JOURNAL_FINISH)
+ _encode_header(header, JOURNAL_FINISH);
+ else
+ _encode_header(header, JOURNAL_NONE);
+ op.omap_set_header(header);
+ }
+ {
+ // remove journal
+ std::set<string> to_remove;
+ for (auto &it : loaded_journal)
+ to_remove.emplace(it.first);
+ op.omap_rm_keys(to_remove);
+ }
+ loaded_journal.clear();
+
+ object_t oid = get_object_name(omap_idx);
+ for (auto& op : op_vec) {
+ mds->objecter->mutate(oid, oloc, op, snapc, ceph::real_clock::now(),
+ 0, gather.new_sub());
+ }
+ }
+ gather.activate();
+ return;
+ }
+
+ journal_state = JOURNAL_NONE;
+ err = 0;
+ dout(10) << __func__ << ": load complete" << dendl;
+out:
+
+ if (err < 0)
+ _reset_states();
+
+ load_done = true;
+ finish_contexts(g_ceph_context, waiting_for_load);
+ waiting_for_load.clear();
+}
+
+void OpenFileTable::load(MDSContext *onload)
+{
+ dout(10) << __func__ << dendl;
+ ceph_assert(!load_done);
+ if (onload)
+ waiting_for_load.push_back(onload);
+
+ C_IO_OFT_Load *c = new C_IO_OFT_Load(this, 0, true);
+ object_t oid = get_object_name(0);
+ object_locator_t oloc(mds->mdsmap->get_metadata_pool());
+
+ ObjectOperation op;
+ op.omap_get_header(&c->header_bl, &c->header_r);
+ op.omap_get_vals("", "", uint64_t(-1),
+ &c->values, &c->more, &c->values_r);
+
+ mds->objecter->read(oid, oloc, op, CEPH_NOSNAP, nullptr, 0,
+ new C_OnFinisher(c, mds->finisher));
+}
+
+bool OpenFileTable::get_ancestors(inodeno_t ino, vector<inode_backpointer_t>& ancestors,
+ mds_rank_t& auth_hint)
+{
+ auto p = loaded_anchor_map.find(ino);
+ if (p == loaded_anchor_map.end())
+ return false;
+
+ inodeno_t dirino = p->second.dirino;
+ if (dirino == inodeno_t(0))
+ return false;
+
+ bool first = true;
+ ancestors.clear();
+ while (true) {
+ ancestors.push_back(inode_backpointer_t(dirino, p->second.d_name, 0));
+
+ p = loaded_anchor_map.find(dirino);
+ if (p == loaded_anchor_map.end())
+ break;
+
+ if (first)
+ auth_hint = p->second.auth;
+
+ dirino = p->second.dirino;
+ if (dirino == inodeno_t(0))
+ break;
+
+ first = false;
+ }
+ return true;
+}
+
+class C_OFT_OpenInoFinish: public MDSContext {
+ OpenFileTable *oft;
+ inodeno_t ino;
+ MDSRank *get_mds() override { return oft->mds; }
+public:
+ C_OFT_OpenInoFinish(OpenFileTable *t, inodeno_t i) : oft(t), ino(i) {}
+ void finish(int r) override {
+ oft->_open_ino_finish(ino, r);
+ }
+};
+
+void OpenFileTable::_open_ino_finish(inodeno_t ino, int r)
+{
+ if (prefetch_state == DIR_INODES && r >= 0 && ino != inodeno_t(0)) {
+ auto p = loaded_anchor_map.find(ino);
+ ceph_assert(p != loaded_anchor_map.end());
+ p->second.auth = mds_rank_t(r);
+ }
+
+ if (r != mds->get_nodeid())
+ mds->mdcache->rejoin_prefetch_ino_finish(ino, r);
+
+ num_opening_inodes--;
+ if (num_opening_inodes == 0) {
+ if (prefetch_state == DIR_INODES) {
+ prefetch_state = DIRFRAGS;
+ _prefetch_dirfrags();
+ } else if (prefetch_state == FILE_INODES) {
+ prefetch_state = DONE;
+ logseg_destroyed_inos.clear();
+ destroyed_inos_set.clear();
+ finish_contexts(g_ceph_context, waiting_for_prefetch);
+ waiting_for_prefetch.clear();
+ } else {
+ ceph_abort();
+ }
+ }
+}
+
+void OpenFileTable::_prefetch_dirfrags()
+{
+ dout(10) << __func__ << dendl;
+ ceph_assert(prefetch_state == DIRFRAGS);
+
+ MDCache *mdcache = mds->mdcache;
+ list<CDir*> fetch_queue;
+
+ CInode *last_in = nullptr;
+ for (auto df : loaded_dirfrags) {
+ CInode *diri;
+ if (last_in && last_in->ino() == df.ino) {
+ diri = last_in;
+ } else {
+ diri = mdcache->get_inode(df.ino);
+ if (!diri)
+ continue;
+ last_in = diri;
+ }
+ if (diri->state_test(CInode::STATE_REJOINUNDEF))
+ continue;
+
+ CDir *dir = diri->get_dirfrag(df.frag);
+ if (dir) {
+ if (dir->is_auth() && !dir->is_complete())
+ fetch_queue.push_back(dir);
+ } else {
+ frag_vec_t leaves;
+ diri->dirfragtree.get_leaves_under(df.frag, leaves);
+ for (const auto& leaf : leaves) {
+ if (diri->is_auth()) {
+ dir = diri->get_or_open_dirfrag(mdcache, leaf);
+ } else {
+ dir = diri->get_dirfrag(leaf);
+ }
+ if (dir && dir->is_auth() && !dir->is_complete())
+ fetch_queue.push_back(dir);
+ }
+ }
+ }
+
+ MDSGatherBuilder gather(g_ceph_context);
+ int num_opening_dirfrags = 0;
+ for (auto dir : fetch_queue) {
+ if (dir->state_test(CDir::STATE_REJOINUNDEF))
+ ceph_assert(dir->get_inode()->dirfragtree.is_leaf(dir->get_frag()));
+ dir->fetch(gather.new_sub());
+
+ if (!(++num_opening_dirfrags % 1000))
+ mds->heartbeat_reset();
+ }
+
+ auto finish_func = [this](int r) {
+ prefetch_state = FILE_INODES;
+ _prefetch_inodes();
+ };
+ if (gather.has_subs()) {
+ gather.set_finisher(
+ new MDSInternalContextWrapper(mds,
+ new FunctionContext(finish_func)));
+ gather.activate();
+ } else {
+ finish_func(0);
+ }
+}
+
+void OpenFileTable::_prefetch_inodes()
+{
+ dout(10) << __func__ << " state " << prefetch_state << dendl;
+ ceph_assert(!num_opening_inodes);
+ num_opening_inodes = 1;
+
+ int64_t pool;
+ if (prefetch_state == DIR_INODES)
+ pool = mds->mdsmap->get_metadata_pool();
+ else if (prefetch_state == FILE_INODES)
+ pool = mds->mdsmap->get_first_data_pool();
+ else
+ ceph_abort();
+
+ MDCache *mdcache = mds->mdcache;
+
+ if (destroyed_inos_set.empty()) {
+ for (auto& it : logseg_destroyed_inos)
+ destroyed_inos_set.insert(it.second.begin(), it.second.end());
+ }
+
+ for (auto& it : loaded_anchor_map) {
+ if (destroyed_inos_set.count(it.first))
+ continue;
+ if (it.second.d_type == DT_DIR) {
+ if (prefetch_state != DIR_INODES)
+ continue;
+ if (MDS_INO_IS_MDSDIR(it.first)) {
+ it.second.auth = MDS_INO_MDSDIR_OWNER(it.first);
+ continue;
+ }
+ if (MDS_INO_IS_STRAY(it.first)) {
+ it.second.auth = MDS_INO_STRAY_OWNER(it.first);
+ continue;
+ }
+ } else {
+ if (prefetch_state != FILE_INODES)
+ continue;
+ // load all file inodes for MDCache::identify_files_to_recover()
+ }
+ CInode *in = mdcache->get_inode(it.first);
+ if (in)
+ continue;
+
+ num_opening_inodes++;
+ mdcache->open_ino(it.first, pool, new C_OFT_OpenInoFinish(this, it.first), false);
+
+ if (!(num_opening_inodes % 1000))
+ mds->heartbeat_reset();
+ }
+
+ _open_ino_finish(inodeno_t(0), 0);
+}
+
+bool OpenFileTable::prefetch_inodes()
+{
+ dout(10) << __func__ << dendl;
+ ceph_assert(!prefetch_state);
+ prefetch_state = DIR_INODES;
+
+ if (!load_done) {
+ wait_for_load(
+ new MDSInternalContextWrapper(mds,
+ new FunctionContext([this](int r) {
+ _prefetch_inodes();
+ })
+ )
+ );
+ return true;
+ }
+
+ _prefetch_inodes();
+ return !is_prefetched();
+}
+
+bool OpenFileTable::should_log_open(CInode *in)
+{
+ if (in->state_test(CInode::STATE_TRACKEDBYOFT)) {
+ // inode just journaled
+ if (in->last_journaled >= committing_log_seq)
+ return false;
+ // item not dirty. it means the item has already been saved
+ auto p = dirty_items.find(in->ino());
+ if (p == dirty_items.end())
+ return false;
+ }
+ return true;
+}
+
+void OpenFileTable::note_destroyed_inos(uint64_t seq, const vector<inodeno_t>& inos)
+{
+ auto& vec = logseg_destroyed_inos[seq];
+ vec.insert(vec.end(), inos.begin(), inos.end());
+}
+
+void OpenFileTable::trim_destroyed_inos(uint64_t seq)
+{
+ auto p = logseg_destroyed_inos.begin();
+ while (p != logseg_destroyed_inos.end()) {
+ if (p->first >= seq)
+ break;
+ logseg_destroyed_inos.erase(p++);
+ }
+}