summaryrefslogtreecommitdiffstats
path: root/src/osdc/Objecter.cc
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/osdc/Objecter.cc5344
1 files changed, 5344 insertions, 0 deletions
diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
new file mode 100644
index 000000000..6fb200eb1
--- /dev/null
+++ b/src/osdc/Objecter.cc
@@ -0,0 +1,5344 @@
+// -*- 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.
+ *
+ */
+
+#include <algorithm>
+#include <cerrno>
+
+#include "Objecter.h"
+#include "osd/OSDMap.h"
+#include "osd/error_code.h"
+#include "Filer.h"
+
+#include "mon/MonClient.h"
+#include "mon/error_code.h"
+
+#include "msg/Messenger.h"
+#include "msg/Message.h"
+
+#include "messages/MPing.h"
+#include "messages/MOSDOp.h"
+#include "messages/MOSDOpReply.h"
+#include "messages/MOSDBackoff.h"
+#include "messages/MOSDMap.h"
+
+#include "messages/MPoolOp.h"
+#include "messages/MPoolOpReply.h"
+
+#include "messages/MGetPoolStats.h"
+#include "messages/MGetPoolStatsReply.h"
+#include "messages/MStatfs.h"
+#include "messages/MStatfsReply.h"
+
+#include "messages/MMonCommand.h"
+
+#include "messages/MCommand.h"
+#include "messages/MCommandReply.h"
+
+#include "messages/MWatchNotify.h"
+
+
+#include "common/Cond.h"
+#include "common/config.h"
+#include "common/perf_counters.h"
+#include "common/scrub_types.h"
+#include "include/str_list.h"
+#include "common/errno.h"
+#include "common/EventTrace.h"
+#include "common/async/waiter.h"
+#include "error_code.h"
+
+
+using std::list;
+using std::make_pair;
+using std::map;
+using std::ostream;
+using std::ostringstream;
+using std::pair;
+using std::set;
+using std::string;
+using std::stringstream;
+using std::vector;
+
+using ceph::decode;
+using ceph::encode;
+using ceph::Formatter;
+
+using std::defer_lock;
+
+using ceph::real_time;
+using ceph::real_clock;
+
+using ceph::mono_clock;
+using ceph::mono_time;
+
+using ceph::timespan;
+
+using ceph::shunique_lock;
+using ceph::acquire_shared;
+using ceph::acquire_unique;
+
+namespace bc = boost::container;
+namespace bs = boost::system;
+namespace ca = ceph::async;
+namespace cb = ceph::buffer;
+
+#define dout_subsys ceph_subsys_objecter
+#undef dout_prefix
+#define dout_prefix *_dout << messenger->get_myname() << ".objecter "
+
+
+enum {
+ l_osdc_first = 123200,
+ l_osdc_op_active,
+ l_osdc_op_laggy,
+ l_osdc_op_send,
+ l_osdc_op_send_bytes,
+ l_osdc_op_resend,
+ l_osdc_op_reply,
+ l_osdc_oplen_avg,
+
+ l_osdc_op,
+ l_osdc_op_r,
+ l_osdc_op_w,
+ l_osdc_op_rmw,
+ l_osdc_op_pg,
+
+ l_osdc_osdop_stat,
+ l_osdc_osdop_create,
+ l_osdc_osdop_read,
+ l_osdc_osdop_write,
+ l_osdc_osdop_writefull,
+ l_osdc_osdop_writesame,
+ l_osdc_osdop_append,
+ l_osdc_osdop_zero,
+ l_osdc_osdop_truncate,
+ l_osdc_osdop_delete,
+ l_osdc_osdop_mapext,
+ l_osdc_osdop_sparse_read,
+ l_osdc_osdop_clonerange,
+ l_osdc_osdop_getxattr,
+ l_osdc_osdop_setxattr,
+ l_osdc_osdop_cmpxattr,
+ l_osdc_osdop_rmxattr,
+ l_osdc_osdop_resetxattrs,
+ l_osdc_osdop_call,
+ l_osdc_osdop_watch,
+ l_osdc_osdop_notify,
+ l_osdc_osdop_src_cmpxattr,
+ l_osdc_osdop_pgls,
+ l_osdc_osdop_pgls_filter,
+ l_osdc_osdop_other,
+
+ l_osdc_linger_active,
+ l_osdc_linger_send,
+ l_osdc_linger_resend,
+ l_osdc_linger_ping,
+
+ l_osdc_poolop_active,
+ l_osdc_poolop_send,
+ l_osdc_poolop_resend,
+
+ l_osdc_poolstat_active,
+ l_osdc_poolstat_send,
+ l_osdc_poolstat_resend,
+
+ l_osdc_statfs_active,
+ l_osdc_statfs_send,
+ l_osdc_statfs_resend,
+
+ l_osdc_command_active,
+ l_osdc_command_send,
+ l_osdc_command_resend,
+
+ l_osdc_map_epoch,
+ l_osdc_map_full,
+ l_osdc_map_inc,
+
+ l_osdc_osd_sessions,
+ l_osdc_osd_session_open,
+ l_osdc_osd_session_close,
+ l_osdc_osd_laggy,
+
+ l_osdc_osdop_omap_wr,
+ l_osdc_osdop_omap_rd,
+ l_osdc_osdop_omap_del,
+
+ l_osdc_last,
+};
+
+namespace {
+inline bs::error_code osdcode(int r) {
+ return (r < 0) ? bs::error_code(-r, osd_category()) : bs::error_code();
+}
+}
+
+// config obs ----------------------------
+
+class Objecter::RequestStateHook : public AdminSocketHook {
+ Objecter *m_objecter;
+public:
+ explicit RequestStateHook(Objecter *objecter);
+ int call(std::string_view command, const cmdmap_t& cmdmap,
+ Formatter *f,
+ std::ostream& ss,
+ cb::list& out) override;
+};
+
+std::unique_lock<std::mutex> Objecter::OSDSession::get_lock(object_t& oid)
+{
+ if (oid.name.empty())
+ return {};
+
+ static constexpr uint32_t HASH_PRIME = 1021;
+ uint32_t h = ceph_str_hash_linux(oid.name.c_str(), oid.name.size())
+ % HASH_PRIME;
+
+ return {completion_locks[h % num_locks], std::defer_lock};
+}
+
+const char** Objecter::get_tracked_conf_keys() const
+{
+ static const char *config_keys[] = {
+ "crush_location",
+ "rados_mon_op_timeout",
+ "rados_osd_op_timeout",
+ NULL
+ };
+ return config_keys;
+}
+
+
+void Objecter::handle_conf_change(const ConfigProxy& conf,
+ const std::set <std::string> &changed)
+{
+ if (changed.count("crush_location")) {
+ update_crush_location();
+ }
+ if (changed.count("rados_mon_op_timeout")) {
+ mon_timeout = conf.get_val<std::chrono::seconds>("rados_mon_op_timeout");
+ }
+ if (changed.count("rados_osd_op_timeout")) {
+ osd_timeout = conf.get_val<std::chrono::seconds>("rados_osd_op_timeout");
+ }
+}
+
+void Objecter::update_crush_location()
+{
+ unique_lock wl(rwlock);
+ crush_location = cct->crush_location.get_location();
+}
+
+// messages ------------------------------
+
+/*
+ * initialize only internal data structures, don't initiate cluster interaction
+ */
+void Objecter::init()
+{
+ ceph_assert(!initialized);
+
+ if (!logger) {
+ PerfCountersBuilder pcb(cct, "objecter", l_osdc_first, l_osdc_last);
+
+ pcb.add_u64(l_osdc_op_active, "op_active", "Operations active", "actv",
+ PerfCountersBuilder::PRIO_CRITICAL);
+ pcb.add_u64(l_osdc_op_laggy, "op_laggy", "Laggy operations");
+ pcb.add_u64_counter(l_osdc_op_send, "op_send", "Sent operations");
+ pcb.add_u64_counter(l_osdc_op_send_bytes, "op_send_bytes", "Sent data", NULL, 0, unit_t(UNIT_BYTES));
+ pcb.add_u64_counter(l_osdc_op_resend, "op_resend", "Resent operations");
+ pcb.add_u64_counter(l_osdc_op_reply, "op_reply", "Operation reply");
+ pcb.add_u64_avg(l_osdc_oplen_avg, "oplen_avg", "Average length of operation vector");
+
+ pcb.add_u64_counter(l_osdc_op, "op", "Operations");
+ pcb.add_u64_counter(l_osdc_op_r, "op_r", "Read operations", "rd",
+ PerfCountersBuilder::PRIO_CRITICAL);
+ pcb.add_u64_counter(l_osdc_op_w, "op_w", "Write operations", "wr",
+ PerfCountersBuilder::PRIO_CRITICAL);
+ pcb.add_u64_counter(l_osdc_op_rmw, "op_rmw", "Read-modify-write operations",
+ "rdwr", PerfCountersBuilder::PRIO_INTERESTING);
+ pcb.add_u64_counter(l_osdc_op_pg, "op_pg", "PG operation");
+
+ pcb.add_u64_counter(l_osdc_osdop_stat, "osdop_stat", "Stat operations");
+ pcb.add_u64_counter(l_osdc_osdop_create, "osdop_create",
+ "Create object operations");
+ pcb.add_u64_counter(l_osdc_osdop_read, "osdop_read", "Read operations");
+ pcb.add_u64_counter(l_osdc_osdop_write, "osdop_write", "Write operations");
+ pcb.add_u64_counter(l_osdc_osdop_writefull, "osdop_writefull",
+ "Write full object operations");
+ pcb.add_u64_counter(l_osdc_osdop_writesame, "osdop_writesame",
+ "Write same operations");
+ pcb.add_u64_counter(l_osdc_osdop_append, "osdop_append",
+ "Append operation");
+ pcb.add_u64_counter(l_osdc_osdop_zero, "osdop_zero",
+ "Set object to zero operations");
+ pcb.add_u64_counter(l_osdc_osdop_truncate, "osdop_truncate",
+ "Truncate object operations");
+ pcb.add_u64_counter(l_osdc_osdop_delete, "osdop_delete",
+ "Delete object operations");
+ pcb.add_u64_counter(l_osdc_osdop_mapext, "osdop_mapext",
+ "Map extent operations");
+ pcb.add_u64_counter(l_osdc_osdop_sparse_read, "osdop_sparse_read",
+ "Sparse read operations");
+ pcb.add_u64_counter(l_osdc_osdop_clonerange, "osdop_clonerange",
+ "Clone range operations");
+ pcb.add_u64_counter(l_osdc_osdop_getxattr, "osdop_getxattr",
+ "Get xattr operations");
+ pcb.add_u64_counter(l_osdc_osdop_setxattr, "osdop_setxattr",
+ "Set xattr operations");
+ pcb.add_u64_counter(l_osdc_osdop_cmpxattr, "osdop_cmpxattr",
+ "Xattr comparison operations");
+ pcb.add_u64_counter(l_osdc_osdop_rmxattr, "osdop_rmxattr",
+ "Remove xattr operations");
+ pcb.add_u64_counter(l_osdc_osdop_resetxattrs, "osdop_resetxattrs",
+ "Reset xattr operations");
+ pcb.add_u64_counter(l_osdc_osdop_call, "osdop_call",
+ "Call (execute) operations");
+ pcb.add_u64_counter(l_osdc_osdop_watch, "osdop_watch",
+ "Watch by object operations");
+ pcb.add_u64_counter(l_osdc_osdop_notify, "osdop_notify",
+ "Notify about object operations");
+ pcb.add_u64_counter(l_osdc_osdop_src_cmpxattr, "osdop_src_cmpxattr",
+ "Extended attribute comparison in multi operations");
+ pcb.add_u64_counter(l_osdc_osdop_pgls, "osdop_pgls");
+ pcb.add_u64_counter(l_osdc_osdop_pgls_filter, "osdop_pgls_filter");
+ pcb.add_u64_counter(l_osdc_osdop_other, "osdop_other", "Other operations");
+
+ pcb.add_u64(l_osdc_linger_active, "linger_active",
+ "Active lingering operations");
+ pcb.add_u64_counter(l_osdc_linger_send, "linger_send",
+ "Sent lingering operations");
+ pcb.add_u64_counter(l_osdc_linger_resend, "linger_resend",
+ "Resent lingering operations");
+ pcb.add_u64_counter(l_osdc_linger_ping, "linger_ping",
+ "Sent pings to lingering operations");
+
+ pcb.add_u64(l_osdc_poolop_active, "poolop_active",
+ "Active pool operations");
+ pcb.add_u64_counter(l_osdc_poolop_send, "poolop_send",
+ "Sent pool operations");
+ pcb.add_u64_counter(l_osdc_poolop_resend, "poolop_resend",
+ "Resent pool operations");
+
+ pcb.add_u64(l_osdc_poolstat_active, "poolstat_active",
+ "Active get pool stat operations");
+ pcb.add_u64_counter(l_osdc_poolstat_send, "poolstat_send",
+ "Pool stat operations sent");
+ pcb.add_u64_counter(l_osdc_poolstat_resend, "poolstat_resend",
+ "Resent pool stats");
+
+ pcb.add_u64(l_osdc_statfs_active, "statfs_active", "Statfs operations");
+ pcb.add_u64_counter(l_osdc_statfs_send, "statfs_send", "Sent FS stats");
+ pcb.add_u64_counter(l_osdc_statfs_resend, "statfs_resend",
+ "Resent FS stats");
+
+ pcb.add_u64(l_osdc_command_active, "command_active", "Active commands");
+ pcb.add_u64_counter(l_osdc_command_send, "command_send",
+ "Sent commands");
+ pcb.add_u64_counter(l_osdc_command_resend, "command_resend",
+ "Resent commands");
+
+ pcb.add_u64(l_osdc_map_epoch, "map_epoch", "OSD map epoch");
+ pcb.add_u64_counter(l_osdc_map_full, "map_full",
+ "Full OSD maps received");
+ pcb.add_u64_counter(l_osdc_map_inc, "map_inc",
+ "Incremental OSD maps received");
+
+ pcb.add_u64(l_osdc_osd_sessions, "osd_sessions",
+ "Open sessions"); // open sessions
+ pcb.add_u64_counter(l_osdc_osd_session_open, "osd_session_open",
+ "Sessions opened");
+ pcb.add_u64_counter(l_osdc_osd_session_close, "osd_session_close",
+ "Sessions closed");
+ pcb.add_u64(l_osdc_osd_laggy, "osd_laggy", "Laggy OSD sessions");
+
+ pcb.add_u64_counter(l_osdc_osdop_omap_wr, "omap_wr",
+ "OSD OMAP write operations");
+ pcb.add_u64_counter(l_osdc_osdop_omap_rd, "omap_rd",
+ "OSD OMAP read operations");
+ pcb.add_u64_counter(l_osdc_osdop_omap_del, "omap_del",
+ "OSD OMAP delete operations");
+
+ logger = pcb.create_perf_counters();
+ cct->get_perfcounters_collection()->add(logger);
+ }
+
+ m_request_state_hook = new RequestStateHook(this);
+ auto admin_socket = cct->get_admin_socket();
+ int ret = admin_socket->register_command("objecter_requests",
+ m_request_state_hook,
+ "show in-progress osd requests");
+
+ /* Don't warn on EEXIST, happens if multiple ceph clients
+ * are instantiated from one process */
+ if (ret < 0 && ret != -EEXIST) {
+ lderr(cct) << "error registering admin socket command: "
+ << cpp_strerror(ret) << dendl;
+ }
+
+ update_crush_location();
+
+ cct->_conf.add_observer(this);
+
+ initialized = true;
+}
+
+/*
+ * ok, cluster interaction can happen
+ */
+void Objecter::start(const OSDMap* o)
+{
+ shared_lock rl(rwlock);
+
+ start_tick();
+ if (o) {
+ osdmap->deepish_copy_from(*o);
+ prune_pg_mapping(osdmap->get_pools());
+ } else if (osdmap->get_epoch() == 0) {
+ _maybe_request_map();
+ }
+}
+
+void Objecter::shutdown()
+{
+ ceph_assert(initialized);
+
+ unique_lock wl(rwlock);
+
+ initialized = false;
+
+ wl.unlock();
+ cct->_conf.remove_observer(this);
+ wl.lock();
+
+ while (!osd_sessions.empty()) {
+ auto p = osd_sessions.begin();
+ close_session(p->second);
+ }
+
+ while(!check_latest_map_lingers.empty()) {
+ auto i = check_latest_map_lingers.begin();
+ i->second->put();
+ check_latest_map_lingers.erase(i->first);
+ }
+
+ while(!check_latest_map_ops.empty()) {
+ auto i = check_latest_map_ops.begin();
+ i->second->put();
+ check_latest_map_ops.erase(i->first);
+ }
+
+ while(!check_latest_map_commands.empty()) {
+ auto i = check_latest_map_commands.begin();
+ i->second->put();
+ check_latest_map_commands.erase(i->first);
+ }
+
+ while(!poolstat_ops.empty()) {
+ auto i = poolstat_ops.begin();
+ delete i->second;
+ poolstat_ops.erase(i->first);
+ }
+
+ while(!statfs_ops.empty()) {
+ auto i = statfs_ops.begin();
+ delete i->second;
+ statfs_ops.erase(i->first);
+ }
+
+ while(!pool_ops.empty()) {
+ auto i = pool_ops.begin();
+ delete i->second;
+ pool_ops.erase(i->first);
+ }
+
+ ldout(cct, 20) << __func__ << " clearing up homeless session..." << dendl;
+ while(!homeless_session->linger_ops.empty()) {
+ auto i = homeless_session->linger_ops.begin();
+ ldout(cct, 10) << " linger_op " << i->first << dendl;
+ LingerOp *lop = i->second;
+ {
+ std::unique_lock swl(homeless_session->lock);
+ _session_linger_op_remove(homeless_session, lop);
+ }
+ linger_ops.erase(lop->linger_id);
+ linger_ops_set.erase(lop);
+ lop->put();
+ }
+
+ while(!homeless_session->ops.empty()) {
+ auto i = homeless_session->ops.begin();
+ ldout(cct, 10) << " op " << i->first << dendl;
+ auto op = i->second;
+ {
+ std::unique_lock swl(homeless_session->lock);
+ _session_op_remove(homeless_session, op);
+ }
+ op->put();
+ }
+
+ while(!homeless_session->command_ops.empty()) {
+ auto i = homeless_session->command_ops.begin();
+ ldout(cct, 10) << " command_op " << i->first << dendl;
+ auto cop = i->second;
+ {
+ std::unique_lock swl(homeless_session->lock);
+ _session_command_op_remove(homeless_session, cop);
+ }
+ cop->put();
+ }
+
+ if (tick_event) {
+ if (timer.cancel_event(tick_event)) {
+ ldout(cct, 10) << " successfully canceled tick" << dendl;
+ }
+ tick_event = 0;
+ }
+
+ if (logger) {
+ cct->get_perfcounters_collection()->remove(logger);
+ delete logger;
+ logger = NULL;
+ }
+
+ // Let go of Objecter write lock so timer thread can shutdown
+ wl.unlock();
+
+ // Outside of lock to avoid cycle WRT calls to RequestStateHook
+ // This is safe because we guarantee no concurrent calls to
+ // shutdown() with the ::initialized check at start.
+ if (m_request_state_hook) {
+ auto admin_socket = cct->get_admin_socket();
+ admin_socket->unregister_commands(m_request_state_hook);
+ delete m_request_state_hook;
+ m_request_state_hook = NULL;
+ }
+}
+
+void Objecter::_send_linger(LingerOp *info,
+ ceph::shunique_lock<ceph::shared_mutex>& sul)
+{
+ ceph_assert(sul.owns_lock() && sul.mutex() == &rwlock);
+
+ fu2::unique_function<Op::OpSig> oncommit;
+ osdc_opvec opv;
+ std::shared_lock watchl(info->watch_lock);
+ cb::list *poutbl = nullptr;
+ if (info->registered && info->is_watch) {
+ ldout(cct, 15) << "send_linger " << info->linger_id << " reconnect"
+ << dendl;
+ opv.push_back(OSDOp());
+ opv.back().op.op = CEPH_OSD_OP_WATCH;
+ opv.back().op.watch.cookie = info->get_cookie();
+ opv.back().op.watch.op = CEPH_OSD_WATCH_OP_RECONNECT;
+ opv.back().op.watch.gen = ++info->register_gen;
+ oncommit = CB_Linger_Reconnect(this, info);
+ } else {
+ ldout(cct, 15) << "send_linger " << info->linger_id << " register"
+ << dendl;
+ opv = info->ops;
+ // TODO Augment ca::Completion with an equivalent of
+ // target so we can handle these cases better.
+ auto c = std::make_unique<CB_Linger_Commit>(this, info);
+ if (!info->is_watch) {
+ info->notify_id = 0;
+ poutbl = &c->outbl;
+ }
+ oncommit = [c = std::move(c)](bs::error_code ec) mutable {
+ std::move(*c)(ec);
+ };
+ }
+ watchl.unlock();
+ auto o = new Op(info->target.base_oid, info->target.base_oloc,
+ std::move(opv), info->target.flags | CEPH_OSD_FLAG_READ,
+ std::move(oncommit), info->pobjver);
+ o->outbl = poutbl;
+ o->snapid = info->snap;
+ o->snapc = info->snapc;
+ o->mtime = info->mtime;
+
+ o->target = info->target;
+ o->tid = ++last_tid;
+
+ // do not resend this; we will send a new op to reregister
+ o->should_resend = false;
+ o->ctx_budgeted = true;
+
+ if (info->register_tid) {
+ // repeat send. cancel old registration op, if any.
+ std::unique_lock sl(info->session->lock);
+ if (info->session->ops.count(info->register_tid)) {
+ auto o = info->session->ops[info->register_tid];
+ _op_cancel_map_check(o);
+ _cancel_linger_op(o);
+ }
+ sl.unlock();
+ }
+
+ _op_submit_with_budget(o, sul, &info->register_tid, &info->ctx_budget);
+
+ logger->inc(l_osdc_linger_send);
+}
+
+void Objecter::_linger_commit(LingerOp *info, bs::error_code ec,
+ cb::list& outbl)
+{
+ std::unique_lock wl(info->watch_lock);
+ ldout(cct, 10) << "_linger_commit " << info->linger_id << dendl;
+ if (info->on_reg_commit) {
+ info->on_reg_commit->defer(std::move(info->on_reg_commit),
+ ec, cb::list{});
+ info->on_reg_commit.reset();
+ }
+ if (ec && info->on_notify_finish) {
+ info->on_notify_finish->defer(std::move(info->on_notify_finish),
+ ec, cb::list{});
+ info->on_notify_finish.reset();
+ }
+
+ // only tell the user the first time we do this
+ info->registered = true;
+ info->pobjver = NULL;
+
+ if (!info->is_watch) {
+ // make note of the notify_id
+ auto p = outbl.cbegin();
+ try {
+ decode(info->notify_id, p);
+ ldout(cct, 10) << "_linger_commit notify_id=" << info->notify_id
+ << dendl;
+ }
+ catch (cb::error& e) {
+ }
+ }
+}
+
+class CB_DoWatchError {
+ Objecter *objecter;
+ boost::intrusive_ptr<Objecter::LingerOp> info;
+ bs::error_code ec;
+public:
+ CB_DoWatchError(Objecter *o, Objecter::LingerOp *i,
+ bs::error_code ec)
+ : objecter(o), info(i), ec(ec) {
+ info->_queued_async();
+ }
+ void operator()() {
+ std::unique_lock wl(objecter->rwlock);
+ bool canceled = info->canceled;
+ wl.unlock();
+
+ if (!canceled) {
+ info->handle(ec, 0, info->get_cookie(), 0, {});
+ }
+
+ info->finished_async();
+ }
+};
+
+bs::error_code Objecter::_normalize_watch_error(bs::error_code ec)
+{
+ // translate ENOENT -> ENOTCONN so that a delete->disconnection
+ // notification and a failure to reconnect because we raced with
+ // the delete appear the same to the user.
+ if (ec == bs::errc::no_such_file_or_directory)
+ ec = bs::error_code(ENOTCONN, osd_category());
+ return ec;
+}
+
+void Objecter::_linger_reconnect(LingerOp *info, bs::error_code ec)
+{
+ ldout(cct, 10) << __func__ << " " << info->linger_id << " = " << ec
+ << " (last_error " << info->last_error << ")" << dendl;
+ std::unique_lock wl(info->watch_lock);
+ if (ec) {
+ if (!info->last_error) {
+ ec = _normalize_watch_error(ec);
+ if (info->handle) {
+ boost::asio::defer(finish_strand, CB_DoWatchError(this, info, ec));
+ }
+ }
+ }
+
+ info->last_error = ec;
+}
+
+void Objecter::_send_linger_ping(LingerOp *info)
+{
+ // rwlock is locked unique
+ // info->session->lock is locked
+
+ if (cct->_conf->objecter_inject_no_watch_ping) {
+ ldout(cct, 10) << __func__ << " " << info->linger_id << " SKIPPING"
+ << dendl;
+ return;
+ }
+ if (osdmap->test_flag(CEPH_OSDMAP_PAUSERD)) {
+ ldout(cct, 10) << __func__ << " PAUSERD" << dendl;
+ return;
+ }
+
+ ceph::coarse_mono_time now = ceph::coarse_mono_clock::now();
+ ldout(cct, 10) << __func__ << " " << info->linger_id << " now " << now
+ << dendl;
+
+ osdc_opvec opv(1);
+ opv[0].op.op = CEPH_OSD_OP_WATCH;
+ opv[0].op.watch.cookie = info->get_cookie();
+ opv[0].op.watch.op = CEPH_OSD_WATCH_OP_PING;
+ opv[0].op.watch.gen = info->register_gen;
+
+ Op *o = new Op(info->target.base_oid, info->target.base_oloc,
+ std::move(opv), info->target.flags | CEPH_OSD_FLAG_READ,
+ CB_Linger_Ping(this, info, now),
+ nullptr, nullptr);
+ o->target = info->target;
+ o->should_resend = false;
+ _send_op_account(o);
+ o->tid = ++last_tid;
+ _session_op_assign(info->session, o);
+ _send_op(o);
+ info->ping_tid = o->tid;
+
+ logger->inc(l_osdc_linger_ping);
+}
+
+void Objecter::_linger_ping(LingerOp *info, bs::error_code ec, ceph::coarse_mono_time sent,
+ uint32_t register_gen)
+{
+ std::unique_lock l(info->watch_lock);
+ ldout(cct, 10) << __func__ << " " << info->linger_id
+ << " sent " << sent << " gen " << register_gen << " = " << ec
+ << " (last_error " << info->last_error
+ << " register_gen " << info->register_gen << ")" << dendl;
+ if (info->register_gen == register_gen) {
+ if (!ec) {
+ info->watch_valid_thru = sent;
+ } else if (ec && !info->last_error) {
+ ec = _normalize_watch_error(ec);
+ info->last_error = ec;
+ if (info->handle) {
+ boost::asio::defer(finish_strand, CB_DoWatchError(this, info, ec));
+ }
+ }
+ } else {
+ ldout(cct, 20) << " ignoring old gen" << dendl;
+ }
+}
+
+tl::expected<ceph::timespan,
+ bs::error_code> Objecter::linger_check(LingerOp *info)
+{
+ std::shared_lock l(info->watch_lock);
+
+ ceph::coarse_mono_time stamp = info->watch_valid_thru;
+ if (!info->watch_pending_async.empty())
+ stamp = std::min(info->watch_valid_thru, info->watch_pending_async.front());
+ auto age = ceph::coarse_mono_clock::now() - stamp;
+
+ ldout(cct, 10) << __func__ << " " << info->linger_id
+ << " err " << info->last_error
+ << " age " << age << dendl;
+ if (info->last_error)
+ return tl::unexpected(info->last_error);
+ // return a safe upper bound (we are truncating to ms)
+ return age;
+}
+
+void Objecter::linger_cancel(LingerOp *info)
+{
+ unique_lock wl(rwlock);
+ _linger_cancel(info);
+ info->put();
+}
+
+void Objecter::_linger_cancel(LingerOp *info)
+{
+ // rwlock is locked unique
+ ldout(cct, 20) << __func__ << " linger_id=" << info->linger_id << dendl;
+ if (!info->canceled) {
+ OSDSession *s = info->session;
+ std::unique_lock sl(s->lock);
+ _session_linger_op_remove(s, info);
+ sl.unlock();
+
+ linger_ops.erase(info->linger_id);
+ linger_ops_set.erase(info);
+ ceph_assert(linger_ops.size() == linger_ops_set.size());
+
+ info->canceled = true;
+ info->put();
+
+ logger->dec(l_osdc_linger_active);
+ }
+}
+
+
+
+Objecter::LingerOp *Objecter::linger_register(const object_t& oid,
+ const object_locator_t& oloc,
+ int flags)
+{
+ unique_lock l(rwlock);
+ // Acquire linger ID
+ auto info = new LingerOp(this, ++max_linger_id);
+ info->target.base_oid = oid;
+ info->target.base_oloc = oloc;
+ if (info->target.base_oloc.key == oid)
+ info->target.base_oloc.key.clear();
+ info->target.flags = flags;
+ info->watch_valid_thru = ceph::coarse_mono_clock::now();
+ ldout(cct, 10) << __func__ << " info " << info
+ << " linger_id " << info->linger_id
+ << " cookie " << info->get_cookie()
+ << dendl;
+ linger_ops[info->linger_id] = info;
+ linger_ops_set.insert(info);
+ ceph_assert(linger_ops.size() == linger_ops_set.size());
+
+ info->get(); // for the caller
+ return info;
+}
+
+ceph_tid_t Objecter::linger_watch(LingerOp *info,
+ ObjectOperation& op,
+ const SnapContext& snapc,
+ real_time mtime,
+ cb::list& inbl,
+ decltype(info->on_reg_commit)&& oncommit,
+ version_t *objver)
+{
+ info->is_watch = true;
+ info->snapc = snapc;
+ info->mtime = mtime;
+ info->target.flags |= CEPH_OSD_FLAG_WRITE;
+ info->ops = op.ops;
+ info->inbl = inbl;
+ info->pobjver = objver;
+ info->on_reg_commit = std::move(oncommit);
+
+ info->ctx_budget = take_linger_budget(info);
+
+ shunique_lock sul(rwlock, ceph::acquire_unique);
+ _linger_submit(info, sul);
+ logger->inc(l_osdc_linger_active);
+
+ op.clear();
+ return info->linger_id;
+}
+
+ceph_tid_t Objecter::linger_notify(LingerOp *info,
+ ObjectOperation& op,
+ snapid_t snap, cb::list& inbl,
+ decltype(LingerOp::on_reg_commit)&& onfinish,
+ version_t *objver)
+{
+ info->snap = snap;
+ info->target.flags |= CEPH_OSD_FLAG_READ;
+ info->ops = op.ops;
+ info->inbl = inbl;
+ info->pobjver = objver;
+ info->on_reg_commit = std::move(onfinish);
+ info->ctx_budget = take_linger_budget(info);
+
+ shunique_lock sul(rwlock, ceph::acquire_unique);
+ _linger_submit(info, sul);
+ logger->inc(l_osdc_linger_active);
+
+ op.clear();
+ return info->linger_id;
+}
+
+void Objecter::_linger_submit(LingerOp *info,
+ ceph::shunique_lock<ceph::shared_mutex>& sul)
+{
+ ceph_assert(sul.owns_lock() && sul.mutex() == &rwlock);
+ ceph_assert(info->linger_id);
+ ceph_assert(info->ctx_budget != -1); // caller needs to have taken budget already!
+
+ // Populate Op::target
+ OSDSession *s = NULL;
+ _calc_target(&info->target, nullptr);
+
+ // Create LingerOp<->OSDSession relation
+ int r = _get_session(info->target.osd, &s, sul);
+ ceph_assert(r == 0);
+ unique_lock sl(s->lock);
+ _session_linger_op_assign(s, info);
+ sl.unlock();
+ put_session(s);
+
+ _send_linger(info, sul);
+}
+
+struct CB_DoWatchNotify {
+ Objecter *objecter;
+ boost::intrusive_ptr<Objecter::LingerOp> info;
+ boost::intrusive_ptr<MWatchNotify> msg;
+ CB_DoWatchNotify(Objecter *o, Objecter::LingerOp *i, MWatchNotify *m)
+ : objecter(o), info(i), msg(m) {
+ info->_queued_async();
+ }
+ void operator()() {
+ objecter->_do_watch_notify(std::move(info), std::move(msg));
+ }
+};
+
+void Objecter::handle_watch_notify(MWatchNotify *m)
+{
+ shared_lock l(rwlock);
+ if (!initialized) {
+ return;
+ }
+
+ LingerOp *info = reinterpret_cast<LingerOp*>(m->cookie);
+ if (linger_ops_set.count(info) == 0) {
+ ldout(cct, 7) << __func__ << " cookie " << m->cookie << " dne" << dendl;
+ return;
+ }
+ std::unique_lock wl(info->watch_lock);
+ if (m->opcode == CEPH_WATCH_EVENT_DISCONNECT) {
+ if (!info->last_error) {
+ info->last_error = bs::error_code(ENOTCONN, osd_category());
+ if (info->handle) {
+ boost::asio::defer(finish_strand, CB_DoWatchError(this, info,
+ info->last_error));
+ }
+ }
+ } else if (!info->is_watch) {
+ // we have CEPH_WATCH_EVENT_NOTIFY_COMPLETE; we can do this inline
+ // since we know the only user (librados) is safe to call in
+ // fast-dispatch context
+ if (info->notify_id &&
+ info->notify_id != m->notify_id) {
+ ldout(cct, 10) << __func__ << " reply notify " << m->notify_id
+ << " != " << info->notify_id << ", ignoring" << dendl;
+ } else if (info->on_notify_finish) {
+ info->on_notify_finish->defer(
+ std::move(info->on_notify_finish),
+ osdcode(m->return_code), std::move(m->get_data()));
+
+ // if we race with reconnect we might get a second notify; only
+ // notify the caller once!
+ info->on_notify_finish = nullptr;
+ }
+ } else {
+ boost::asio::defer(finish_strand, CB_DoWatchNotify(this, info, m));
+ }
+}
+
+void Objecter::_do_watch_notify(boost::intrusive_ptr<LingerOp> info,
+ boost::intrusive_ptr<MWatchNotify> m)
+{
+ ldout(cct, 10) << __func__ << " " << *m << dendl;
+
+ shared_lock l(rwlock);
+ ceph_assert(initialized);
+
+ if (info->canceled) {
+ l.unlock();
+ goto out;
+ }
+
+ // notify completion?
+ ceph_assert(info->is_watch);
+ ceph_assert(info->handle);
+ ceph_assert(m->opcode != CEPH_WATCH_EVENT_DISCONNECT);
+
+ l.unlock();
+
+ switch (m->opcode) {
+ case CEPH_WATCH_EVENT_NOTIFY:
+ info->handle({}, m->notify_id, m->cookie, m->notifier_gid, std::move(m->bl));
+ break;
+ }
+
+ out:
+ info->finished_async();
+}
+
+bool Objecter::ms_dispatch(Message *m)
+{
+ ldout(cct, 10) << __func__ << " " << cct << " " << *m << dendl;
+ switch (m->get_type()) {
+ // these we exlusively handle
+ case CEPH_MSG_OSD_OPREPLY:
+ handle_osd_op_reply(static_cast<MOSDOpReply*>(m));
+ return true;
+
+ case CEPH_MSG_OSD_BACKOFF:
+ handle_osd_backoff(static_cast<MOSDBackoff*>(m));
+ return true;
+
+ case CEPH_MSG_WATCH_NOTIFY:
+ handle_watch_notify(static_cast<MWatchNotify*>(m));
+ m->put();
+ return true;
+
+ case MSG_COMMAND_REPLY:
+ if (m->get_source().type() == CEPH_ENTITY_TYPE_OSD) {
+ handle_command_reply(static_cast<MCommandReply*>(m));
+ return true;
+ } else {
+ return false;
+ }
+
+ case MSG_GETPOOLSTATSREPLY:
+ handle_get_pool_stats_reply(static_cast<MGetPoolStatsReply*>(m));
+ return true;
+
+ case CEPH_MSG_POOLOP_REPLY:
+ handle_pool_op_reply(static_cast<MPoolOpReply*>(m));
+ return true;
+
+ case CEPH_MSG_STATFS_REPLY:
+ handle_fs_stats_reply(static_cast<MStatfsReply*>(m));
+ return true;
+
+ // these we give others a chance to inspect
+
+ // MDS, OSD
+ case CEPH_MSG_OSD_MAP:
+ handle_osd_map(static_cast<MOSDMap*>(m));
+ return false;
+ }
+ return false;
+}
+
+void Objecter::_scan_requests(
+ OSDSession *s,
+ bool skipped_map,
+ bool cluster_full,
+ map<int64_t, bool> *pool_full_map,
+ map<ceph_tid_t, Op*>& need_resend,
+ list<LingerOp*>& need_resend_linger,
+ map<ceph_tid_t, CommandOp*>& need_resend_command,
+ ceph::shunique_lock<ceph::shared_mutex>& sul)
+{
+ ceph_assert(sul.owns_lock() && sul.mutex() == &rwlock);
+
+ list<LingerOp*> unregister_lingers;
+
+ std::unique_lock sl(s->lock);
+
+ // check for changed linger mappings (_before_ regular ops)
+ auto lp = s->linger_ops.begin();
+ while (lp != s->linger_ops.end()) {
+ auto op = lp->second;
+ ceph_assert(op->session == s);
+ // check_linger_pool_dne() may touch linger_ops; prevent iterator
+ // invalidation
+ ++lp;
+ ldout(cct, 10) << " checking linger op " << op->linger_id << dendl;
+ bool unregister, force_resend_writes = cluster_full;
+ int r = _recalc_linger_op_target(op, sul);
+ if (pool_full_map)
+ force_resend_writes = force_resend_writes ||
+ (*pool_full_map)[op->target.base_oloc.pool];
+ switch (r) {
+ case RECALC_OP_TARGET_NO_ACTION:
+ if (!skipped_map && !force_resend_writes)
+ break;
+ // -- fall-thru --
+ case RECALC_OP_TARGET_NEED_RESEND:
+ need_resend_linger.push_back(op);
+ _linger_cancel_map_check(op);
+ break;
+ case RECALC_OP_TARGET_POOL_DNE:
+ _check_linger_pool_dne(op, &unregister);
+ if (unregister) {
+ ldout(cct, 10) << " need to unregister linger op "
+ << op->linger_id << dendl;
+ op->get();
+ unregister_lingers.push_back(op);
+ }
+ break;
+ }
+ }
+
+ // check for changed request mappings
+ auto p = s->ops.begin();
+ while (p != s->ops.end()) {
+ Op *op = p->second;
+ ++p; // check_op_pool_dne() may touch ops; prevent iterator invalidation
+ ldout(cct, 10) << " checking op " << op->tid << dendl;
+ _prune_snapc(osdmap->get_new_removed_snaps(), op);
+ bool force_resend_writes = cluster_full;
+ if (pool_full_map)
+ force_resend_writes = force_resend_writes ||
+ (*pool_full_map)[op->target.base_oloc.pool];
+ int r = _calc_target(&op->target,
+ op->session ? op->session->con.get() : nullptr);
+ switch (r) {
+ case RECALC_OP_TARGET_NO_ACTION:
+ if (!skipped_map && !(force_resend_writes && op->target.respects_full()))
+ break;
+ // -- fall-thru --
+ case RECALC_OP_TARGET_NEED_RESEND:
+ _session_op_remove(op->session, op);
+ need_resend[op->tid] = op;
+ _op_cancel_map_check(op);
+ break;
+ case RECALC_OP_TARGET_POOL_DNE:
+ _check_op_pool_dne(op, &sl);
+ break;
+ }
+ }
+
+ // commands
+ auto cp = s->command_ops.begin();
+ while (cp != s->command_ops.end()) {
+ auto c = cp->second;
+ ++cp;
+ ldout(cct, 10) << " checking command " << c->tid << dendl;
+ bool force_resend_writes = cluster_full;
+ if (pool_full_map)
+ force_resend_writes = force_resend_writes ||
+ (*pool_full_map)[c->target_pg.pool()];
+ int r = _calc_command_target(c, sul);
+ switch (r) {
+ case RECALC_OP_TARGET_NO_ACTION:
+ // resend if skipped map; otherwise do nothing.
+ if (!skipped_map && !force_resend_writes)
+ break;
+ // -- fall-thru --
+ case RECALC_OP_TARGET_NEED_RESEND:
+ need_resend_command[c->tid] = c;
+ _session_command_op_remove(c->session, c);
+ _command_cancel_map_check(c);
+ break;
+ case RECALC_OP_TARGET_POOL_DNE:
+ case RECALC_OP_TARGET_OSD_DNE:
+ case RECALC_OP_TARGET_OSD_DOWN:
+ _check_command_map_dne(c);
+ break;
+ }
+ }
+
+ sl.unlock();
+
+ for (auto iter = unregister_lingers.begin();
+ iter != unregister_lingers.end();
+ ++iter) {
+ _linger_cancel(*iter);
+ (*iter)->put();
+ }
+}
+
+void Objecter::handle_osd_map(MOSDMap *m)
+{
+ ceph::shunique_lock sul(rwlock, acquire_unique);
+ if (!initialized)
+ return;
+
+ ceph_assert(osdmap);
+
+ if (m->fsid != monc->get_fsid()) {
+ ldout(cct, 0) << "handle_osd_map fsid " << m->fsid
+ << " != " << monc->get_fsid() << dendl;
+ return;
+ }
+
+ bool was_pauserd = osdmap->test_flag(CEPH_OSDMAP_PAUSERD);
+ bool cluster_full = _osdmap_full_flag();
+ bool was_pausewr = osdmap->test_flag(CEPH_OSDMAP_PAUSEWR) || cluster_full ||
+ _osdmap_has_pool_full();
+ map<int64_t, bool> pool_full_map;
+ for (auto it = osdmap->get_pools().begin();
+ it != osdmap->get_pools().end(); ++it)
+ pool_full_map[it->first] = _osdmap_pool_full(it->second);
+
+
+ list<LingerOp*> need_resend_linger;
+ map<ceph_tid_t, Op*> need_resend;
+ map<ceph_tid_t, CommandOp*> need_resend_command;
+
+ if (m->get_last() <= osdmap->get_epoch()) {
+ ldout(cct, 3) << "handle_osd_map ignoring epochs ["
+ << m->get_first() << "," << m->get_last()
+ << "] <= " << osdmap->get_epoch() << dendl;
+ } else {
+ ldout(cct, 3) << "handle_osd_map got epochs ["
+ << m->get_first() << "," << m->get_last()
+ << "] > " << osdmap->get_epoch() << dendl;
+
+ if (osdmap->get_epoch()) {
+ bool skipped_map = false;
+ // we want incrementals
+ for (epoch_t e = osdmap->get_epoch() + 1;
+ e <= m->get_last();
+ e++) {
+
+ if (osdmap->get_epoch() == e-1 &&
+ m->incremental_maps.count(e)) {
+ ldout(cct, 3) << "handle_osd_map decoding incremental epoch " << e
+ << dendl;
+ OSDMap::Incremental inc(m->incremental_maps[e]);
+ osdmap->apply_incremental(inc);
+
+ emit_blocklist_events(inc);
+
+ logger->inc(l_osdc_map_inc);
+ }
+ else if (m->maps.count(e)) {
+ ldout(cct, 3) << "handle_osd_map decoding full epoch " << e << dendl;
+ auto new_osdmap = std::make_unique<OSDMap>();
+ new_osdmap->decode(m->maps[e]);
+
+ emit_blocklist_events(*osdmap, *new_osdmap);
+ osdmap = std::move(new_osdmap);
+
+ logger->inc(l_osdc_map_full);
+ }
+ else {
+ if (e >= m->get_oldest()) {
+ ldout(cct, 3) << "handle_osd_map requesting missing epoch "
+ << osdmap->get_epoch()+1 << dendl;
+ _maybe_request_map();
+ break;
+ }
+ ldout(cct, 3) << "handle_osd_map missing epoch "
+ << osdmap->get_epoch()+1
+ << ", jumping to " << m->get_oldest() << dendl;
+ e = m->get_oldest() - 1;
+ skipped_map = true;
+ continue;
+ }
+ logger->set(l_osdc_map_epoch, osdmap->get_epoch());
+
+ prune_pg_mapping(osdmap->get_pools());
+ cluster_full = cluster_full || _osdmap_full_flag();
+ update_pool_full_map(pool_full_map);
+
+ // check all outstanding requests on every epoch
+ for (auto& i : need_resend) {
+ _prune_snapc(osdmap->get_new_removed_snaps(), i.second);
+ }
+ _scan_requests(homeless_session, skipped_map, cluster_full,
+ &pool_full_map, need_resend,
+ need_resend_linger, need_resend_command, sul);
+ for (auto p = osd_sessions.begin();
+ p != osd_sessions.end(); ) {
+ auto s = p->second;
+ _scan_requests(s, skipped_map, cluster_full,
+ &pool_full_map, need_resend,
+ need_resend_linger, need_resend_command, sul);
+ ++p;
+ // osd down or addr change?
+ if (!osdmap->is_up(s->osd) ||
+ (s->con &&
+ s->con->get_peer_addrs() != osdmap->get_addrs(s->osd))) {
+ close_session(s);
+ }
+ }
+
+ ceph_assert(e == osdmap->get_epoch());
+ }
+
+ } else {
+ // first map. we want the full thing.
+ if (m->maps.count(m->get_last())) {
+ for (auto p = osd_sessions.begin();
+ p != osd_sessions.end(); ++p) {
+ OSDSession *s = p->second;
+ _scan_requests(s, false, false, NULL, need_resend,
+ need_resend_linger, need_resend_command, sul);
+ }
+ ldout(cct, 3) << "handle_osd_map decoding full epoch "
+ << m->get_last() << dendl;
+ osdmap->decode(m->maps[m->get_last()]);
+ prune_pg_mapping(osdmap->get_pools());
+
+ _scan_requests(homeless_session, false, false, NULL,
+ need_resend, need_resend_linger,
+ need_resend_command, sul);
+ } else {
+ ldout(cct, 3) << "handle_osd_map hmm, i want a full map, requesting"
+ << dendl;
+ monc->sub_want("osdmap", 0, CEPH_SUBSCRIBE_ONETIME);
+ monc->renew_subs();
+ }
+ }
+ }
+
+ // make sure need_resend targets reflect latest map
+ for (auto p = need_resend.begin(); p != need_resend.end(); ) {
+ Op *op = p->second;
+ if (op->target.epoch < osdmap->get_epoch()) {
+ ldout(cct, 10) << __func__ << " checking op " << p->first << dendl;
+ int r = _calc_target(&op->target, nullptr);
+ if (r == RECALC_OP_TARGET_POOL_DNE) {
+ p = need_resend.erase(p);
+ _check_op_pool_dne(op, nullptr);
+ } else {
+ ++p;
+ }
+ } else {
+ ++p;
+ }
+ }
+
+ bool pauserd = osdmap->test_flag(CEPH_OSDMAP_PAUSERD);
+ bool pausewr = osdmap->test_flag(CEPH_OSDMAP_PAUSEWR) || _osdmap_full_flag()
+ || _osdmap_has_pool_full();
+
+ // was/is paused?
+ if (was_pauserd || was_pausewr || pauserd || pausewr ||
+ osdmap->get_epoch() < epoch_barrier) {
+ _maybe_request_map();
+ }
+
+ // resend requests
+ for (auto p = need_resend.begin();
+ p != need_resend.end(); ++p) {
+ auto op = p->second;
+ auto s = op->session;
+ bool mapped_session = false;
+ if (!s) {
+ int r = _map_session(&op->target, &s, sul);
+ ceph_assert(r == 0);
+ mapped_session = true;
+ } else {
+ get_session(s);
+ }
+ std::unique_lock sl(s->lock);
+ if (mapped_session) {
+ _session_op_assign(s, op);
+ }
+ if (op->should_resend) {
+ if (!op->session->is_homeless() && !op->target.paused) {
+ logger->inc(l_osdc_op_resend);
+ _send_op(op);
+ }
+ } else {
+ _op_cancel_map_check(op);
+ _cancel_linger_op(op);
+ }
+ sl.unlock();
+ put_session(s);
+ }
+ for (auto p = need_resend_linger.begin();
+ p != need_resend_linger.end(); ++p) {
+ LingerOp *op = *p;
+ ceph_assert(op->session);
+ if (!op->session->is_homeless()) {
+ logger->inc(l_osdc_linger_resend);
+ _send_linger(op, sul);
+ }
+ }
+ for (auto p = need_resend_command.begin();
+ p != need_resend_command.end(); ++p) {
+ auto c = p->second;
+ if (c->target.osd >= 0) {
+ _assign_command_session(c, sul);
+ if (c->session && !c->session->is_homeless()) {
+ _send_command(c);
+ }
+ }
+ }
+
+ _dump_active();
+
+ // finish any Contexts that were waiting on a map update
+ auto p = waiting_for_map.begin();
+ while (p != waiting_for_map.end() &&
+ p->first <= osdmap->get_epoch()) {
+ //go through the list and call the onfinish methods
+ for (auto& [c, ec] : p->second) {
+ ca::post(std::move(c), ec);
+ }
+ waiting_for_map.erase(p++);
+ }
+
+ monc->sub_got("osdmap", osdmap->get_epoch());
+
+ if (!waiting_for_map.empty()) {
+ _maybe_request_map();
+ }
+}
+
+void Objecter::enable_blocklist_events()
+{
+ unique_lock wl(rwlock);
+
+ blocklist_events_enabled = true;
+}
+
+void Objecter::consume_blocklist_events(std::set<entity_addr_t> *events)
+{
+ unique_lock wl(rwlock);
+
+ if (events->empty()) {
+ events->swap(blocklist_events);
+ } else {
+ for (const auto &i : blocklist_events) {
+ events->insert(i);
+ }
+ blocklist_events.clear();
+ }
+}
+
+void Objecter::emit_blocklist_events(const OSDMap::Incremental &inc)
+{
+ if (!blocklist_events_enabled) {
+ return;
+ }
+
+ for (const auto &i : inc.new_blocklist) {
+ blocklist_events.insert(i.first);
+ }
+}
+
+void Objecter::emit_blocklist_events(const OSDMap &old_osd_map,
+ const OSDMap &new_osd_map)
+{
+ if (!blocklist_events_enabled) {
+ return;
+ }
+
+ std::set<entity_addr_t> old_set;
+ std::set<entity_addr_t> new_set;
+ std::set<entity_addr_t> old_range_set;
+ std::set<entity_addr_t> new_range_set;
+
+ old_osd_map.get_blocklist(&old_set, &old_range_set);
+ new_osd_map.get_blocklist(&new_set, &new_range_set);
+
+ std::set<entity_addr_t> delta_set;
+ std::set_difference(
+ new_set.begin(), new_set.end(), old_set.begin(), old_set.end(),
+ std::inserter(delta_set, delta_set.begin()));
+ std::set_difference(
+ new_range_set.begin(), new_range_set.end(),
+ old_range_set.begin(), old_range_set.end(),
+ std::inserter(delta_set, delta_set.begin()));
+ blocklist_events.insert(delta_set.begin(), delta_set.end());
+}
+
+// op pool check
+
+void Objecter::CB_Op_Map_Latest::operator()(bs::error_code e,
+ version_t latest, version_t)
+{
+ if (e == bs::errc::resource_unavailable_try_again ||
+ e == bs::errc::operation_canceled)
+ return;
+
+ lgeneric_subdout(objecter->cct, objecter, 10)
+ << "op_map_latest r=" << e << " tid=" << tid
+ << " latest " << latest << dendl;
+
+ unique_lock wl(objecter->rwlock);
+
+ auto iter = objecter->check_latest_map_ops.find(tid);
+ if (iter == objecter->check_latest_map_ops.end()) {
+ lgeneric_subdout(objecter->cct, objecter, 10)
+ << "op_map_latest op "<< tid << " not found" << dendl;
+ return;
+ }
+
+ Op *op = iter->second;
+ objecter->check_latest_map_ops.erase(iter);
+
+ lgeneric_subdout(objecter->cct, objecter, 20)
+ << "op_map_latest op "<< op << dendl;
+
+ if (op->map_dne_bound == 0)
+ op->map_dne_bound = latest;
+
+ unique_lock sl(op->session->lock, defer_lock);
+ objecter->_check_op_pool_dne(op, &sl);
+
+ op->put();
+}
+
+int Objecter::pool_snap_by_name(int64_t poolid, const char *snap_name,
+ snapid_t *snap) const
+{
+ shared_lock rl(rwlock);
+
+ auto& pools = osdmap->get_pools();
+ auto iter = pools.find(poolid);
+ if (iter == pools.end()) {
+ return -ENOENT;
+ }
+ const pg_pool_t& pg_pool = iter->second;
+ for (auto p = pg_pool.snaps.begin();
+ p != pg_pool.snaps.end();
+ ++p) {
+ if (p->second.name == snap_name) {
+ *snap = p->first;
+ return 0;
+ }
+ }
+ return -ENOENT;
+}
+
+int Objecter::pool_snap_get_info(int64_t poolid, snapid_t snap,
+ pool_snap_info_t *info) const
+{
+ shared_lock rl(rwlock);
+
+ auto& pools = osdmap->get_pools();
+ auto iter = pools.find(poolid);
+ if (iter == pools.end()) {
+ return -ENOENT;
+ }
+ const pg_pool_t& pg_pool = iter->second;
+ auto p = pg_pool.snaps.find(snap);
+ if (p == pg_pool.snaps.end())
+ return -ENOENT;
+ *info = p->second;
+
+ return 0;
+}
+
+int Objecter::pool_snap_list(int64_t poolid, vector<uint64_t> *snaps)
+{
+ shared_lock rl(rwlock);
+
+ const pg_pool_t *pi = osdmap->get_pg_pool(poolid);
+ if (!pi)
+ return -ENOENT;
+ for (auto p = pi->snaps.begin();
+ p != pi->snaps.end();
+ ++p) {
+ snaps->push_back(p->first);
+ }
+ return 0;
+}
+
+// sl may be unlocked.
+void Objecter::_check_op_pool_dne(Op *op, std::unique_lock<std::shared_mutex> *sl)
+{
+ // rwlock is locked unique
+
+ if (op->target.pool_ever_existed) {
+ // the pool previously existed and now it does not, which means it
+ // was deleted.
+ op->map_dne_bound = osdmap->get_epoch();
+ ldout(cct, 10) << "check_op_pool_dne tid " << op->tid
+ << " pool previously exists but now does not"
+ << dendl;
+ } else {
+ ldout(cct, 10) << "check_op_pool_dne tid " << op->tid
+ << " current " << osdmap->get_epoch()
+ << " map_dne_bound " << op->map_dne_bound
+ << dendl;
+ }
+ if (op->map_dne_bound > 0) {
+ if (osdmap->get_epoch() >= op->map_dne_bound) {
+ // we had a new enough map
+ ldout(cct, 10) << "check_op_pool_dne tid " << op->tid
+ << " concluding pool " << op->target.base_pgid.pool()
+ << " dne" << dendl;
+ if (op->has_completion()) {
+ num_in_flight--;
+ op->complete(osdc_errc::pool_dne, -ENOENT);
+ }
+
+ OSDSession *s = op->session;
+ if (s) {
+ ceph_assert(s != NULL);
+ ceph_assert(sl->mutex() == &s->lock);
+ bool session_locked = sl->owns_lock();
+ if (!session_locked) {
+ sl->lock();
+ }
+ _finish_op(op, 0);
+ if (!session_locked) {
+ sl->unlock();
+ }
+ } else {
+ _finish_op(op, 0); // no session
+ }
+ }
+ } else {
+ _send_op_map_check(op);
+ }
+}
+
+void Objecter::_send_op_map_check(Op *op)
+{
+ // rwlock is locked unique
+ // ask the monitor
+ if (check_latest_map_ops.count(op->tid) == 0) {
+ op->get();
+ check_latest_map_ops[op->tid] = op;
+ monc->get_version("osdmap", CB_Op_Map_Latest(this, op->tid));
+ }
+}
+
+void Objecter::_op_cancel_map_check(Op *op)
+{
+ // rwlock is locked unique
+ auto iter = check_latest_map_ops.find(op->tid);
+ if (iter != check_latest_map_ops.end()) {
+ Op *op = iter->second;
+ op->put();
+ check_latest_map_ops.erase(iter);
+ }
+}
+
+// linger pool check
+
+void Objecter::CB_Linger_Map_Latest::operator()(bs::error_code e,
+ version_t latest,
+ version_t)
+{
+ if (e == bs::errc::resource_unavailable_try_again ||
+ e == bs::errc::operation_canceled) {
+ // ignore callback; we will retry in resend_mon_ops()
+ return;
+ }
+
+ unique_lock wl(objecter->rwlock);
+
+ auto iter = objecter->check_latest_map_lingers.find(linger_id);
+ if (iter == objecter->check_latest_map_lingers.end()) {
+ return;
+ }
+
+ auto op = iter->second;
+ objecter->check_latest_map_lingers.erase(iter);
+
+ if (op->map_dne_bound == 0)
+ op->map_dne_bound = latest;
+
+ bool unregister;
+ objecter->_check_linger_pool_dne(op, &unregister);
+
+ if (unregister) {
+ objecter->_linger_cancel(op);
+ }
+
+ op->put();
+}
+
+void Objecter::_check_linger_pool_dne(LingerOp *op, bool *need_unregister)
+{
+ // rwlock is locked unique
+
+ *need_unregister = false;
+
+ if (op->register_gen > 0) {
+ ldout(cct, 10) << "_check_linger_pool_dne linger_id " << op->linger_id
+ << " pool previously existed but now does not"
+ << dendl;
+ op->map_dne_bound = osdmap->get_epoch();
+ } else {
+ ldout(cct, 10) << "_check_linger_pool_dne linger_id " << op->linger_id
+ << " current " << osdmap->get_epoch()
+ << " map_dne_bound " << op->map_dne_bound
+ << dendl;
+ }
+ if (op->map_dne_bound > 0) {
+ if (osdmap->get_epoch() >= op->map_dne_bound) {
+ std::unique_lock wl{op->watch_lock};
+ if (op->on_reg_commit) {
+ op->on_reg_commit->defer(std::move(op->on_reg_commit),
+ osdc_errc::pool_dne, cb::list{});
+ op->on_reg_commit = nullptr;
+ }
+ if (op->on_notify_finish) {
+ op->on_notify_finish->defer(std::move(op->on_notify_finish),
+ osdc_errc::pool_dne, cb::list{});
+ op->on_notify_finish = nullptr;
+ }
+ *need_unregister = true;
+ }
+ } else {
+ _send_linger_map_check(op);
+ }
+}
+
+void Objecter::_send_linger_map_check(LingerOp *op)
+{
+ // ask the monitor
+ if (check_latest_map_lingers.count(op->linger_id) == 0) {
+ op->get();
+ check_latest_map_lingers[op->linger_id] = op;
+ monc->get_version("osdmap", CB_Linger_Map_Latest(this, op->linger_id));
+ }
+}
+
+void Objecter::_linger_cancel_map_check(LingerOp *op)
+{
+ // rwlock is locked unique
+
+ auto iter = check_latest_map_lingers.find(op->linger_id);
+ if (iter != check_latest_map_lingers.end()) {
+ LingerOp *op = iter->second;
+ op->put();
+ check_latest_map_lingers.erase(iter);
+ }
+}
+
+// command pool check
+
+void Objecter::CB_Command_Map_Latest::operator()(bs::error_code e,
+ version_t latest, version_t)
+{
+ if (e == bs::errc::resource_unavailable_try_again ||
+ e == bs::errc::operation_canceled) {
+ // ignore callback; we will retry in resend_mon_ops()
+ return;
+ }
+
+ unique_lock wl(objecter->rwlock);
+
+ auto iter = objecter->check_latest_map_commands.find(tid);
+ if (iter == objecter->check_latest_map_commands.end()) {
+ return;
+ }
+
+ auto c = iter->second;
+ objecter->check_latest_map_commands.erase(iter);
+
+ if (c->map_dne_bound == 0)
+ c->map_dne_bound = latest;
+
+ unique_lock sul(c->session->lock);
+ objecter->_check_command_map_dne(c);
+ sul.unlock();
+
+ c->put();
+}
+
+void Objecter::_check_command_map_dne(CommandOp *c)
+{
+ // rwlock is locked unique
+ // session is locked unique
+
+ ldout(cct, 10) << "_check_command_map_dne tid " << c->tid
+ << " current " << osdmap->get_epoch()
+ << " map_dne_bound " << c->map_dne_bound
+ << dendl;
+ if (c->map_dne_bound > 0) {
+ if (osdmap->get_epoch() >= c->map_dne_bound) {
+ _finish_command(c, osdcode(c->map_check_error),
+ std::move(c->map_check_error_str), {});
+ }
+ } else {
+ _send_command_map_check(c);
+ }
+}
+
+void Objecter::_send_command_map_check(CommandOp *c)
+{
+ // rwlock is locked unique
+ // session is locked unique
+
+ // ask the monitor
+ if (check_latest_map_commands.count(c->tid) == 0) {
+ c->get();
+ check_latest_map_commands[c->tid] = c;
+ monc->get_version("osdmap", CB_Command_Map_Latest(this, c->tid));
+ }
+}
+
+void Objecter::_command_cancel_map_check(CommandOp *c)
+{
+ // rwlock is locked uniqe
+
+ auto iter = check_latest_map_commands.find(c->tid);
+ if (iter != check_latest_map_commands.end()) {
+ auto c = iter->second;
+ c->put();
+ check_latest_map_commands.erase(iter);
+ }
+}
+
+
+/**
+ * Look up OSDSession by OSD id.
+ *
+ * @returns 0 on success, or -EAGAIN if the lock context requires
+ * promotion to write.
+ */
+int Objecter::_get_session(int osd, OSDSession **session,
+ shunique_lock<ceph::shared_mutex>& sul)
+{
+ ceph_assert(sul && sul.mutex() == &rwlock);
+
+ if (osd < 0) {
+ *session = homeless_session;
+ ldout(cct, 20) << __func__ << " osd=" << osd << " returning homeless"
+ << dendl;
+ return 0;
+ }
+
+ auto p = osd_sessions.find(osd);
+ if (p != osd_sessions.end()) {
+ auto s = p->second;
+ s->get();
+ *session = s;
+ ldout(cct, 20) << __func__ << " s=" << s << " osd=" << osd << " "
+ << s->get_nref() << dendl;
+ return 0;
+ }
+ if (!sul.owns_lock()) {
+ return -EAGAIN;
+ }
+ auto s = new OSDSession(cct, osd);
+ osd_sessions[osd] = s;
+ s->con = messenger->connect_to_osd(osdmap->get_addrs(osd));
+ s->con->set_priv(RefCountedPtr{s});
+ logger->inc(l_osdc_osd_session_open);
+ logger->set(l_osdc_osd_sessions, osd_sessions.size());
+ s->get();
+ *session = s;
+ ldout(cct, 20) << __func__ << " s=" << s << " osd=" << osd << " "
+ << s->get_nref() << dendl;
+ return 0;
+}
+
+void Objecter::put_session(Objecter::OSDSession *s)
+{
+ if (s && !s->is_homeless()) {
+ ldout(cct, 20) << __func__ << " s=" << s << " osd=" << s->osd << " "
+ << s->get_nref() << dendl;
+ s->put();
+ }
+}
+
+void Objecter::get_session(Objecter::OSDSession *s)
+{
+ ceph_assert(s != NULL);
+
+ if (!s->is_homeless()) {
+ ldout(cct, 20) << __func__ << " s=" << s << " osd=" << s->osd << " "
+ << s->get_nref() << dendl;
+ s->get();
+ }
+}
+
+void Objecter::_reopen_session(OSDSession *s)
+{
+ // rwlock is locked unique
+ // s->lock is locked
+
+ auto addrs = osdmap->get_addrs(s->osd);
+ ldout(cct, 10) << "reopen_session osd." << s->osd << " session, addr now "
+ << addrs << dendl;
+ if (s->con) {
+ s->con->set_priv(NULL);
+ s->con->mark_down();
+ logger->inc(l_osdc_osd_session_close);
+ }
+ s->con = messenger->connect_to_osd(addrs);
+ s->con->set_priv(RefCountedPtr{s});
+ s->incarnation++;
+ logger->inc(l_osdc_osd_session_open);
+}
+
+void Objecter::close_session(OSDSession *s)
+{
+ // rwlock is locked unique
+
+ ldout(cct, 10) << "close_session for osd." << s->osd << dendl;
+ if (s->con) {
+ s->con->set_priv(NULL);
+ s->con->mark_down();
+ logger->inc(l_osdc_osd_session_close);
+ }
+ unique_lock sl(s->lock);
+
+ std::list<LingerOp*> homeless_lingers;
+ std::list<CommandOp*> homeless_commands;
+ std::list<Op*> homeless_ops;
+
+ while (!s->linger_ops.empty()) {
+ auto i = s->linger_ops.begin();
+ ldout(cct, 10) << " linger_op " << i->first << dendl;
+ homeless_lingers.push_back(i->second);
+ _session_linger_op_remove(s, i->second);
+ }
+
+ while (!s->ops.empty()) {
+ auto i = s->ops.begin();
+ ldout(cct, 10) << " op " << i->first << dendl;
+ homeless_ops.push_back(i->second);
+ _session_op_remove(s, i->second);
+ }
+
+ while (!s->command_ops.empty()) {
+ auto i = s->command_ops.begin();
+ ldout(cct, 10) << " command_op " << i->first << dendl;
+ homeless_commands.push_back(i->second);
+ _session_command_op_remove(s, i->second);
+ }
+
+ osd_sessions.erase(s->osd);
+ sl.unlock();
+ put_session(s);
+
+ // Assign any leftover ops to the homeless session
+ {
+ unique_lock hsl(homeless_session->lock);
+ for (auto i = homeless_lingers.begin();
+ i != homeless_lingers.end(); ++i) {
+ _session_linger_op_assign(homeless_session, *i);
+ }
+ for (auto i = homeless_ops.begin();
+ i != homeless_ops.end(); ++i) {
+ _session_op_assign(homeless_session, *i);
+ }
+ for (auto i = homeless_commands.begin();
+ i != homeless_commands.end(); ++i) {
+ _session_command_op_assign(homeless_session, *i);
+ }
+ }
+
+ logger->set(l_osdc_osd_sessions, osd_sessions.size());
+}
+
+void Objecter::wait_for_osd_map(epoch_t e)
+{
+ unique_lock l(rwlock);
+ if (osdmap->get_epoch() >= e) {
+ l.unlock();
+ return;
+ }
+
+ ca::waiter<bs::error_code> w;
+ waiting_for_map[e].emplace_back(OpCompletion::create(
+ service.get_executor(),
+ w.ref()),
+ bs::error_code{});
+ l.unlock();
+ w.wait();
+}
+
+void Objecter::_get_latest_version(epoch_t oldest, epoch_t newest,
+ std::unique_ptr<OpCompletion> fin,
+ std::unique_lock<ceph::shared_mutex>&& l)
+{
+ ceph_assert(fin);
+ if (osdmap->get_epoch() >= newest) {
+ ldout(cct, 10) << __func__ << " latest " << newest << ", have it" << dendl;
+ l.unlock();
+ ca::defer(std::move(fin), bs::error_code{});
+ } else {
+ ldout(cct, 10) << __func__ << " latest " << newest << ", waiting" << dendl;
+ _wait_for_new_map(std::move(fin), newest, bs::error_code{});
+ l.unlock();
+ }
+}
+
+void Objecter::maybe_request_map()
+{
+ shared_lock rl(rwlock);
+ _maybe_request_map();
+}
+
+void Objecter::_maybe_request_map()
+{
+ // rwlock is locked
+ int flag = 0;
+ if (_osdmap_full_flag()
+ || osdmap->test_flag(CEPH_OSDMAP_PAUSERD)
+ || osdmap->test_flag(CEPH_OSDMAP_PAUSEWR)) {
+ ldout(cct, 10) << "_maybe_request_map subscribing (continuous) to next "
+ "osd map (FULL flag is set)" << dendl;
+ } else {
+ ldout(cct, 10)
+ << "_maybe_request_map subscribing (onetime) to next osd map" << dendl;
+ flag = CEPH_SUBSCRIBE_ONETIME;
+ }
+ epoch_t epoch = osdmap->get_epoch() ? osdmap->get_epoch()+1 : 0;
+ if (monc->sub_want("osdmap", epoch, flag)) {
+ monc->renew_subs();
+ }
+}
+
+void Objecter::_wait_for_new_map(std::unique_ptr<OpCompletion> c, epoch_t epoch,
+ bs::error_code ec)
+{
+ // rwlock is locked unique
+ waiting_for_map[epoch].emplace_back(std::move(c), ec);
+ _maybe_request_map();
+}
+
+
+/**
+ * Use this together with wait_for_map: this is a pre-check to avoid
+ * allocating a Context for wait_for_map if we can see that we
+ * definitely already have the epoch.
+ *
+ * This does *not* replace the need to handle the return value of
+ * wait_for_map: just because we don't have it in this pre-check
+ * doesn't mean we won't have it when calling back into wait_for_map,
+ * since the objecter lock is dropped in between.
+ */
+bool Objecter::have_map(const epoch_t epoch)
+{
+ shared_lock rl(rwlock);
+ if (osdmap->get_epoch() >= epoch) {
+ return true;
+ } else {
+ return false;
+ }
+}
+
+void Objecter::_kick_requests(OSDSession *session,
+ map<uint64_t, LingerOp *>& lresend)
+{
+ // rwlock is locked unique
+
+ // clear backoffs
+ session->backoffs.clear();
+ session->backoffs_by_id.clear();
+
+ // resend ops
+ map<ceph_tid_t,Op*> resend; // resend in tid order
+ for (auto p = session->ops.begin(); p != session->ops.end();) {
+ Op *op = p->second;
+ ++p;
+ if (op->should_resend) {
+ if (!op->target.paused)
+ resend[op->tid] = op;
+ } else {
+ _op_cancel_map_check(op);
+ _cancel_linger_op(op);
+ }
+ }
+
+ logger->inc(l_osdc_op_resend, resend.size());
+ while (!resend.empty()) {
+ _send_op(resend.begin()->second);
+ resend.erase(resend.begin());
+ }
+
+ // resend lingers
+ logger->inc(l_osdc_linger_resend, session->linger_ops.size());
+ for (auto j = session->linger_ops.begin();
+ j != session->linger_ops.end(); ++j) {
+ LingerOp *op = j->second;
+ op->get();
+ ceph_assert(lresend.count(j->first) == 0);
+ lresend[j->first] = op;
+ }
+
+ // resend commands
+ logger->inc(l_osdc_command_resend, session->command_ops.size());
+ map<uint64_t,CommandOp*> cresend; // resend in order
+ for (auto k = session->command_ops.begin();
+ k != session->command_ops.end(); ++k) {
+ cresend[k->first] = k->second;
+ }
+ while (!cresend.empty()) {
+ _send_command(cresend.begin()->second);
+ cresend.erase(cresend.begin());
+ }
+}
+
+void Objecter::_linger_ops_resend(map<uint64_t, LingerOp *>& lresend,
+ unique_lock<ceph::shared_mutex>& ul)
+{
+ ceph_assert(ul.owns_lock());
+ shunique_lock sul(std::move(ul));
+ while (!lresend.empty()) {
+ LingerOp *op = lresend.begin()->second;
+ if (!op->canceled) {
+ _send_linger(op, sul);
+ }
+ op->put();
+ lresend.erase(lresend.begin());
+ }
+ ul = sul.release_to_unique();
+}
+
+void Objecter::start_tick()
+{
+ ceph_assert(tick_event == 0);
+ tick_event =
+ timer.add_event(ceph::make_timespan(cct->_conf->objecter_tick_interval),
+ &Objecter::tick, this);
+}
+
+void Objecter::tick()
+{
+ shared_lock rl(rwlock);
+
+ ldout(cct, 10) << "tick" << dendl;
+
+ // we are only called by C_Tick
+ tick_event = 0;
+
+ if (!initialized) {
+ // we raced with shutdown
+ ldout(cct, 10) << __func__ << " raced with shutdown" << dendl;
+ return;
+ }
+
+ set<OSDSession*> toping;
+
+
+ // look for laggy requests
+ auto cutoff = ceph::coarse_mono_clock::now();
+ cutoff -= ceph::make_timespan(cct->_conf->objecter_timeout); // timeout
+
+ unsigned laggy_ops = 0;
+
+ for (auto siter = osd_sessions.begin();
+ siter != osd_sessions.end(); ++siter) {
+ auto s = siter->second;
+ scoped_lock l(s->lock);
+ bool found = false;
+ for (auto p = s->ops.begin(); p != s->ops.end(); ++p) {
+ auto op = p->second;
+ ceph_assert(op->session);
+ if (op->stamp < cutoff) {
+ ldout(cct, 2) << " tid " << p->first << " on osd." << op->session->osd
+ << " is laggy" << dendl;
+ found = true;
+ ++laggy_ops;
+ }
+ }
+ for (auto p = s->linger_ops.begin();
+ p != s->linger_ops.end();
+ ++p) {
+ auto op = p->second;
+ std::unique_lock wl(op->watch_lock);
+ ceph_assert(op->session);
+ ldout(cct, 10) << " pinging osd that serves lingering tid " << p->first
+ << " (osd." << op->session->osd << ")" << dendl;
+ found = true;
+ if (op->is_watch && op->registered && !op->last_error)
+ _send_linger_ping(op);
+ }
+ for (auto p = s->command_ops.begin();
+ p != s->command_ops.end();
+ ++p) {
+ auto op = p->second;
+ ceph_assert(op->session);
+ ldout(cct, 10) << " pinging osd that serves command tid " << p->first
+ << " (osd." << op->session->osd << ")" << dendl;
+ found = true;
+ }
+ if (found)
+ toping.insert(s);
+ }
+ if (num_homeless_ops || !toping.empty()) {
+ _maybe_request_map();
+ }
+
+ logger->set(l_osdc_op_laggy, laggy_ops);
+ logger->set(l_osdc_osd_laggy, toping.size());
+
+ if (!toping.empty()) {
+ // send a ping to these osds, to ensure we detect any session resets
+ // (osd reply message policy is lossy)
+ for (auto i = toping.begin(); i != toping.end(); ++i) {
+ (*i)->con->send_message(new MPing);
+ }
+ }
+
+ // Make sure we don't reschedule if we wake up after shutdown
+ if (initialized) {
+ tick_event = timer.reschedule_me(ceph::make_timespan(
+ cct->_conf->objecter_tick_interval));
+ }
+}
+
+void Objecter::resend_mon_ops()
+{
+ unique_lock wl(rwlock);
+
+ ldout(cct, 10) << "resend_mon_ops" << dendl;
+
+ for (auto p = poolstat_ops.begin(); p != poolstat_ops.end(); ++p) {
+ _poolstat_submit(p->second);
+ logger->inc(l_osdc_poolstat_resend);
+ }
+
+ for (auto p = statfs_ops.begin(); p != statfs_ops.end(); ++p) {
+ _fs_stats_submit(p->second);
+ logger->inc(l_osdc_statfs_resend);
+ }
+
+ for (auto p = pool_ops.begin(); p != pool_ops.end(); ++p) {
+ _pool_op_submit(p->second);
+ logger->inc(l_osdc_poolop_resend);
+ }
+
+ for (auto p = check_latest_map_ops.begin();
+ p != check_latest_map_ops.end();
+ ++p) {
+ monc->get_version("osdmap", CB_Op_Map_Latest(this, p->second->tid));
+ }
+
+ for (auto p = check_latest_map_lingers.begin();
+ p != check_latest_map_lingers.end();
+ ++p) {
+ monc->get_version("osdmap", CB_Linger_Map_Latest(this, p->second->linger_id));
+ }
+
+ for (auto p = check_latest_map_commands.begin();
+ p != check_latest_map_commands.end();
+ ++p) {
+ monc->get_version("osdmap", CB_Command_Map_Latest(this, p->second->tid));
+ }
+}
+
+// read | write ---------------------------
+
+void Objecter::op_submit(Op *op, ceph_tid_t *ptid, int *ctx_budget)
+{
+ shunique_lock rl(rwlock, ceph::acquire_shared);
+ ceph_tid_t tid = 0;
+ if (!ptid)
+ ptid = &tid;
+ op->trace.event("op submit");
+ _op_submit_with_budget(op, rl, ptid, ctx_budget);
+}
+
+void Objecter::_op_submit_with_budget(Op *op,
+ shunique_lock<ceph::shared_mutex>& sul,
+ ceph_tid_t *ptid,
+ int *ctx_budget)
+{
+ ceph_assert(initialized);
+
+ ceph_assert(op->ops.size() == op->out_bl.size());
+ ceph_assert(op->ops.size() == op->out_rval.size());
+ ceph_assert(op->ops.size() == op->out_handler.size());
+
+ // throttle. before we look at any state, because
+ // _take_op_budget() may drop our lock while it blocks.
+ if (!op->ctx_budgeted || (ctx_budget && (*ctx_budget == -1))) {
+ int op_budget = _take_op_budget(op, sul);
+ // take and pass out the budget for the first OP
+ // in the context session
+ if (ctx_budget && (*ctx_budget == -1)) {
+ *ctx_budget = op_budget;
+ }
+ }
+
+ if (osd_timeout > timespan(0)) {
+ if (op->tid == 0)
+ op->tid = ++last_tid;
+ auto tid = op->tid;
+ op->ontimeout = timer.add_event(osd_timeout,
+ [this, tid]() {
+ op_cancel(tid, -ETIMEDOUT); });
+ }
+
+ _op_submit(op, sul, ptid);
+}
+
+void Objecter::_send_op_account(Op *op)
+{
+ inflight_ops++;
+
+ // add to gather set(s)
+ if (op->has_completion()) {
+ num_in_flight++;
+ } else {
+ ldout(cct, 20) << " note: not requesting reply" << dendl;
+ }
+
+ logger->inc(l_osdc_op_active);
+ logger->inc(l_osdc_op);
+ logger->inc(l_osdc_oplen_avg, op->ops.size());
+
+ if ((op->target.flags & (CEPH_OSD_FLAG_READ | CEPH_OSD_FLAG_WRITE)) ==
+ (CEPH_OSD_FLAG_READ|CEPH_OSD_FLAG_WRITE))
+ logger->inc(l_osdc_op_rmw);
+ else if (op->target.flags & CEPH_OSD_FLAG_WRITE)
+ logger->inc(l_osdc_op_w);
+ else if (op->target.flags & CEPH_OSD_FLAG_READ)
+ logger->inc(l_osdc_op_r);
+
+ if (op->target.flags & CEPH_OSD_FLAG_PGOP)
+ logger->inc(l_osdc_op_pg);
+
+ for (auto p = op->ops.begin(); p != op->ops.end(); ++p) {
+ int code = l_osdc_osdop_other;
+ switch (p->op.op) {
+ case CEPH_OSD_OP_STAT: code = l_osdc_osdop_stat; break;
+ case CEPH_OSD_OP_CREATE: code = l_osdc_osdop_create; break;
+ case CEPH_OSD_OP_READ: code = l_osdc_osdop_read; break;
+ case CEPH_OSD_OP_WRITE: code = l_osdc_osdop_write; break;
+ case CEPH_OSD_OP_WRITEFULL: code = l_osdc_osdop_writefull; break;
+ case CEPH_OSD_OP_WRITESAME: code = l_osdc_osdop_writesame; break;
+ case CEPH_OSD_OP_APPEND: code = l_osdc_osdop_append; break;
+ case CEPH_OSD_OP_ZERO: code = l_osdc_osdop_zero; break;
+ case CEPH_OSD_OP_TRUNCATE: code = l_osdc_osdop_truncate; break;
+ case CEPH_OSD_OP_DELETE: code = l_osdc_osdop_delete; break;
+ case CEPH_OSD_OP_MAPEXT: code = l_osdc_osdop_mapext; break;
+ case CEPH_OSD_OP_SPARSE_READ: code = l_osdc_osdop_sparse_read; break;
+ case CEPH_OSD_OP_GETXATTR: code = l_osdc_osdop_getxattr; break;
+ case CEPH_OSD_OP_SETXATTR: code = l_osdc_osdop_setxattr; break;
+ case CEPH_OSD_OP_CMPXATTR: code = l_osdc_osdop_cmpxattr; break;
+ case CEPH_OSD_OP_RMXATTR: code = l_osdc_osdop_rmxattr; break;
+ case CEPH_OSD_OP_RESETXATTRS: code = l_osdc_osdop_resetxattrs; break;
+
+ // OMAP read operations
+ case CEPH_OSD_OP_OMAPGETVALS:
+ case CEPH_OSD_OP_OMAPGETKEYS:
+ case CEPH_OSD_OP_OMAPGETHEADER:
+ case CEPH_OSD_OP_OMAPGETVALSBYKEYS:
+ case CEPH_OSD_OP_OMAP_CMP: code = l_osdc_osdop_omap_rd; break;
+
+ // OMAP write operations
+ case CEPH_OSD_OP_OMAPSETVALS:
+ case CEPH_OSD_OP_OMAPSETHEADER: code = l_osdc_osdop_omap_wr; break;
+
+ // OMAP del operations
+ case CEPH_OSD_OP_OMAPCLEAR:
+ case CEPH_OSD_OP_OMAPRMKEYS: code = l_osdc_osdop_omap_del; break;
+
+ case CEPH_OSD_OP_CALL: code = l_osdc_osdop_call; break;
+ case CEPH_OSD_OP_WATCH: code = l_osdc_osdop_watch; break;
+ case CEPH_OSD_OP_NOTIFY: code = l_osdc_osdop_notify; break;
+ }
+ if (code)
+ logger->inc(code);
+ }
+}
+
+void Objecter::_op_submit(Op *op, shunique_lock<ceph::shared_mutex>& sul, ceph_tid_t *ptid)
+{
+ // rwlock is locked
+
+ ldout(cct, 10) << __func__ << " op " << op << dendl;
+
+ // pick target
+ ceph_assert(op->session == NULL);
+ OSDSession *s = NULL;
+
+ bool check_for_latest_map = _calc_target(&op->target, nullptr)
+ == RECALC_OP_TARGET_POOL_DNE;
+
+ // Try to get a session, including a retry if we need to take write lock
+ int r = _get_session(op->target.osd, &s, sul);
+ if (r == -EAGAIN ||
+ (check_for_latest_map && sul.owns_lock_shared()) ||
+ cct->_conf->objecter_debug_inject_relock_delay) {
+ epoch_t orig_epoch = osdmap->get_epoch();
+ sul.unlock();
+ if (cct->_conf->objecter_debug_inject_relock_delay) {
+ sleep(1);
+ }
+ sul.lock();
+ if (orig_epoch != osdmap->get_epoch()) {
+ // map changed; recalculate mapping
+ ldout(cct, 10) << __func__ << " relock raced with osdmap, recalc target"
+ << dendl;
+ check_for_latest_map = _calc_target(&op->target, nullptr)
+ == RECALC_OP_TARGET_POOL_DNE;
+ if (s) {
+ put_session(s);
+ s = NULL;
+ r = -EAGAIN;
+ }
+ }
+ }
+ if (r == -EAGAIN) {
+ ceph_assert(s == NULL);
+ r = _get_session(op->target.osd, &s, sul);
+ }
+ ceph_assert(r == 0);
+ ceph_assert(s); // may be homeless
+
+ _send_op_account(op);
+
+ // send?
+
+ ceph_assert(op->target.flags & (CEPH_OSD_FLAG_READ|CEPH_OSD_FLAG_WRITE));
+
+ bool need_send = false;
+ if (op->target.paused) {
+ ldout(cct, 10) << " tid " << op->tid << " op " << op << " is paused"
+ << dendl;
+ _maybe_request_map();
+ } else if (!s->is_homeless()) {
+ need_send = true;
+ } else {
+ _maybe_request_map();
+ }
+
+ unique_lock sl(s->lock);
+ if (op->tid == 0)
+ op->tid = ++last_tid;
+
+ ldout(cct, 10) << "_op_submit oid " << op->target.base_oid
+ << " '" << op->target.base_oloc << "' '"
+ << op->target.target_oloc << "' " << op->ops << " tid "
+ << op->tid << " osd." << (!s->is_homeless() ? s->osd : -1)
+ << dendl;
+
+ _session_op_assign(s, op);
+
+ if (need_send) {
+ _send_op(op);
+ }
+
+ // Last chance to touch Op here, after giving up session lock it can
+ // be freed at any time by response handler.
+ ceph_tid_t tid = op->tid;
+ if (check_for_latest_map) {
+ _send_op_map_check(op);
+ }
+ if (ptid)
+ *ptid = tid;
+ op = NULL;
+
+ sl.unlock();
+ put_session(s);
+
+ ldout(cct, 5) << num_in_flight << " in flight" << dendl;
+}
+
+int Objecter::op_cancel(OSDSession *s, ceph_tid_t tid, int r)
+{
+ ceph_assert(initialized);
+
+ unique_lock sl(s->lock);
+
+ auto p = s->ops.find(tid);
+ if (p == s->ops.end()) {
+ ldout(cct, 10) << __func__ << " tid " << tid << " dne in session "
+ << s->osd << dendl;
+ return -ENOENT;
+ }
+
+#if 0
+ if (s->con) {
+ ldout(cct, 20) << " revoking rx ceph::buffer for " << tid
+ << " on " << s->con << dendl;
+ s->con->revoke_rx_buffer(tid);
+ }
+#endif
+
+ ldout(cct, 10) << __func__ << " tid " << tid << " in session " << s->osd
+ << dendl;
+ Op *op = p->second;
+ if (op->has_completion()) {
+ num_in_flight--;
+ op->complete(osdcode(r), r);
+ }
+ _op_cancel_map_check(op);
+ _finish_op(op, r);
+ sl.unlock();
+
+ return 0;
+}
+
+int Objecter::op_cancel(ceph_tid_t tid, int r)
+{
+ int ret = 0;
+
+ unique_lock wl(rwlock);
+ ret = _op_cancel(tid, r);
+
+ return ret;
+}
+
+int Objecter::op_cancel(const vector<ceph_tid_t>& tids, int r)
+{
+ unique_lock wl(rwlock);
+ ldout(cct,10) << __func__ << " " << tids << dendl;
+ for (auto tid : tids) {
+ _op_cancel(tid, r);
+ }
+ return 0;
+}
+
+int Objecter::_op_cancel(ceph_tid_t tid, int r)
+{
+ int ret = 0;
+
+ ldout(cct, 5) << __func__ << ": cancelling tid " << tid << " r=" << r
+ << dendl;
+
+start:
+
+ for (auto siter = osd_sessions.begin();
+ siter != osd_sessions.end(); ++siter) {
+ OSDSession *s = siter->second;
+ shared_lock sl(s->lock);
+ if (s->ops.find(tid) != s->ops.end()) {
+ sl.unlock();
+ ret = op_cancel(s, tid, r);
+ if (ret == -ENOENT) {
+ /* oh no! raced, maybe tid moved to another session, restarting */
+ goto start;
+ }
+ return ret;
+ }
+ }
+
+ ldout(cct, 5) << __func__ << ": tid " << tid
+ << " not found in live sessions" << dendl;
+
+ // Handle case where the op is in homeless session
+ shared_lock sl(homeless_session->lock);
+ if (homeless_session->ops.find(tid) != homeless_session->ops.end()) {
+ sl.unlock();
+ ret = op_cancel(homeless_session, tid, r);
+ if (ret == -ENOENT) {
+ /* oh no! raced, maybe tid moved to another session, restarting */
+ goto start;
+ } else {
+ return ret;
+ }
+ } else {
+ sl.unlock();
+ }
+
+ ldout(cct, 5) << __func__ << ": tid " << tid
+ << " not found in homeless session" << dendl;
+
+ return ret;
+}
+
+
+epoch_t Objecter::op_cancel_writes(int r, int64_t pool)
+{
+ unique_lock wl(rwlock);
+
+ std::vector<ceph_tid_t> to_cancel;
+ bool found = false;
+
+ for (auto siter = osd_sessions.begin();
+ siter != osd_sessions.end(); ++siter) {
+ OSDSession *s = siter->second;
+ shared_lock sl(s->lock);
+ for (auto op_i = s->ops.begin();
+ op_i != s->ops.end(); ++op_i) {
+ if (op_i->second->target.flags & CEPH_OSD_FLAG_WRITE
+ && (pool == -1 || op_i->second->target.target_oloc.pool == pool)) {
+ to_cancel.push_back(op_i->first);
+ }
+ }
+ sl.unlock();
+
+ for (auto titer = to_cancel.begin(); titer != to_cancel.end(); ++titer) {
+ int cancel_result = op_cancel(s, *titer, r);
+ // We hold rwlock across search and cancellation, so cancels
+ // should always succeed
+ ceph_assert(cancel_result == 0);
+ }
+ if (!found && to_cancel.size())
+ found = true;
+ to_cancel.clear();
+ }
+
+ const epoch_t epoch = osdmap->get_epoch();
+
+ wl.unlock();
+
+ if (found) {
+ return epoch;
+ } else {
+ return -1;
+ }
+}
+
+bool Objecter::is_pg_changed(
+ int oldprimary,
+ const vector<int>& oldacting,
+ int newprimary,
+ const vector<int>& newacting,
+ bool any_change)
+{
+ if (OSDMap::primary_changed_broken( // https://tracker.ceph.com/issues/43213
+ oldprimary,
+ oldacting,
+ newprimary,
+ newacting))
+ return true;
+ if (any_change && oldacting != newacting)
+ return true;
+ return false; // same primary (tho replicas may have changed)
+}
+
+bool Objecter::target_should_be_paused(op_target_t *t)
+{
+ const pg_pool_t *pi = osdmap->get_pg_pool(t->base_oloc.pool);
+ bool pauserd = osdmap->test_flag(CEPH_OSDMAP_PAUSERD);
+ bool pausewr = osdmap->test_flag(CEPH_OSDMAP_PAUSEWR) ||
+ (t->respects_full() && (_osdmap_full_flag() || _osdmap_pool_full(*pi)));
+
+ return (t->flags & CEPH_OSD_FLAG_READ && pauserd) ||
+ (t->flags & CEPH_OSD_FLAG_WRITE && pausewr) ||
+ (osdmap->get_epoch() < epoch_barrier);
+}
+
+/**
+ * Locking public accessor for _osdmap_full_flag
+ */
+bool Objecter::osdmap_full_flag() const
+{
+ shared_lock rl(rwlock);
+
+ return _osdmap_full_flag();
+}
+
+bool Objecter::osdmap_pool_full(const int64_t pool_id) const
+{
+ shared_lock rl(rwlock);
+
+ if (_osdmap_full_flag()) {
+ return true;
+ }
+
+ return _osdmap_pool_full(pool_id);
+}
+
+bool Objecter::_osdmap_pool_full(const int64_t pool_id) const
+{
+ const pg_pool_t *pool = osdmap->get_pg_pool(pool_id);
+ if (pool == NULL) {
+ ldout(cct, 4) << __func__ << ": DNE pool " << pool_id << dendl;
+ return false;
+ }
+
+ return _osdmap_pool_full(*pool);
+}
+
+bool Objecter::_osdmap_has_pool_full() const
+{
+ for (auto it = osdmap->get_pools().begin();
+ it != osdmap->get_pools().end(); ++it) {
+ if (_osdmap_pool_full(it->second))
+ return true;
+ }
+ return false;
+}
+
+/**
+ * Wrapper around osdmap->test_flag for special handling of the FULL flag.
+ */
+bool Objecter::_osdmap_full_flag() const
+{
+ // Ignore the FULL flag if the caller does not have honor_osdmap_full
+ return osdmap->test_flag(CEPH_OSDMAP_FULL) && honor_pool_full;
+}
+
+void Objecter::update_pool_full_map(map<int64_t, bool>& pool_full_map)
+{
+ for (map<int64_t, pg_pool_t>::const_iterator it
+ = osdmap->get_pools().begin();
+ it != osdmap->get_pools().end(); ++it) {
+ if (pool_full_map.find(it->first) == pool_full_map.end()) {
+ pool_full_map[it->first] = _osdmap_pool_full(it->second);
+ } else {
+ pool_full_map[it->first] = _osdmap_pool_full(it->second) ||
+ pool_full_map[it->first];
+ }
+ }
+}
+
+int64_t Objecter::get_object_hash_position(int64_t pool, const string& key,
+ const string& ns)
+{
+ shared_lock rl(rwlock);
+ const pg_pool_t *p = osdmap->get_pg_pool(pool);
+ if (!p)
+ return -ENOENT;
+ return p->hash_key(key, ns);
+}
+
+int64_t Objecter::get_object_pg_hash_position(int64_t pool, const string& key,
+ const string& ns)
+{
+ shared_lock rl(rwlock);
+ const pg_pool_t *p = osdmap->get_pg_pool(pool);
+ if (!p)
+ return -ENOENT;
+ return p->raw_hash_to_pg(p->hash_key(key, ns));
+}
+
+void Objecter::_prune_snapc(
+ const mempool::osdmap::map<int64_t,
+ snap_interval_set_t>& new_removed_snaps,
+ Op *op)
+{
+ bool match = false;
+ auto i = new_removed_snaps.find(op->target.base_pgid.pool());
+ if (i != new_removed_snaps.end()) {
+ for (auto s : op->snapc.snaps) {
+ if (i->second.contains(s)) {
+ match = true;
+ break;
+ }
+ }
+ if (match) {
+ vector<snapid_t> new_snaps;
+ for (auto s : op->snapc.snaps) {
+ if (!i->second.contains(s)) {
+ new_snaps.push_back(s);
+ }
+ }
+ op->snapc.snaps.swap(new_snaps);
+ ldout(cct,10) << __func__ << " op " << op->tid << " snapc " << op->snapc
+ << " (was " << new_snaps << ")" << dendl;
+ }
+ }
+}
+
+int Objecter::_calc_target(op_target_t *t, Connection *con, bool any_change)
+{
+ // rwlock is locked
+ bool is_read = t->flags & CEPH_OSD_FLAG_READ;
+ bool is_write = t->flags & CEPH_OSD_FLAG_WRITE;
+ t->epoch = osdmap->get_epoch();
+ ldout(cct,20) << __func__ << " epoch " << t->epoch
+ << " base " << t->base_oid << " " << t->base_oloc
+ << " precalc_pgid " << (int)t->precalc_pgid
+ << " pgid " << t->base_pgid
+ << (is_read ? " is_read" : "")
+ << (is_write ? " is_write" : "")
+ << dendl;
+
+ const pg_pool_t *pi = osdmap->get_pg_pool(t->base_oloc.pool);
+ if (!pi) {
+ t->osd = -1;
+ return RECALC_OP_TARGET_POOL_DNE;
+ }
+ ldout(cct,30) << __func__ << " base pi " << pi
+ << " pg_num " << pi->get_pg_num() << dendl;
+
+ bool force_resend = false;
+ if (osdmap->get_epoch() == pi->last_force_op_resend) {
+ if (t->last_force_resend < pi->last_force_op_resend) {
+ t->last_force_resend = pi->last_force_op_resend;
+ force_resend = true;
+ } else if (t->last_force_resend == 0) {
+ force_resend = true;
+ }
+ }
+
+ // apply tiering
+ t->target_oid = t->base_oid;
+ t->target_oloc = t->base_oloc;
+ if ((t->flags & CEPH_OSD_FLAG_IGNORE_OVERLAY) == 0) {
+ if (is_read && pi->has_read_tier())
+ t->target_oloc.pool = pi->read_tier;
+ if (is_write && pi->has_write_tier())
+ t->target_oloc.pool = pi->write_tier;
+ pi = osdmap->get_pg_pool(t->target_oloc.pool);
+ if (!pi) {
+ t->osd = -1;
+ return RECALC_OP_TARGET_POOL_DNE;
+ }
+ }
+
+ pg_t pgid;
+ if (t->precalc_pgid) {
+ ceph_assert(t->flags & CEPH_OSD_FLAG_IGNORE_OVERLAY);
+ ceph_assert(t->base_oid.name.empty()); // make sure this is a pg op
+ ceph_assert(t->base_oloc.pool == (int64_t)t->base_pgid.pool());
+ pgid = t->base_pgid;
+ } else {
+ int ret = osdmap->object_locator_to_pg(t->target_oid, t->target_oloc,
+ pgid);
+ if (ret == -ENOENT) {
+ t->osd = -1;
+ return RECALC_OP_TARGET_POOL_DNE;
+ }
+ }
+ ldout(cct,20) << __func__ << " target " << t->target_oid << " "
+ << t->target_oloc << " -> pgid " << pgid << dendl;
+ ldout(cct,30) << __func__ << " target pi " << pi
+ << " pg_num " << pi->get_pg_num() << dendl;
+ t->pool_ever_existed = true;
+
+ int size = pi->size;
+ int min_size = pi->min_size;
+ unsigned pg_num = pi->get_pg_num();
+ unsigned pg_num_mask = pi->get_pg_num_mask();
+ unsigned pg_num_pending = pi->get_pg_num_pending();
+ int up_primary, acting_primary;
+ vector<int> up, acting;
+ ps_t actual_ps = ceph_stable_mod(pgid.ps(), pg_num, pg_num_mask);
+ pg_t actual_pgid(actual_ps, pgid.pool());
+ pg_mapping_t pg_mapping;
+ pg_mapping.epoch = osdmap->get_epoch();
+ if (lookup_pg_mapping(actual_pgid, &pg_mapping)) {
+ up = pg_mapping.up;
+ up_primary = pg_mapping.up_primary;
+ acting = pg_mapping.acting;
+ acting_primary = pg_mapping.acting_primary;
+ } else {
+ osdmap->pg_to_up_acting_osds(actual_pgid, &up, &up_primary,
+ &acting, &acting_primary);
+ pg_mapping_t pg_mapping(osdmap->get_epoch(),
+ up, up_primary, acting, acting_primary);
+ update_pg_mapping(actual_pgid, std::move(pg_mapping));
+ }
+ bool sort_bitwise = osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE);
+ bool recovery_deletes = osdmap->test_flag(CEPH_OSDMAP_RECOVERY_DELETES);
+ unsigned prev_seed = ceph_stable_mod(pgid.ps(), t->pg_num, t->pg_num_mask);
+ pg_t prev_pgid(prev_seed, pgid.pool());
+ if (any_change && PastIntervals::is_new_interval(
+ t->acting_primary,
+ acting_primary,
+ t->acting,
+ acting,
+ t->up_primary,
+ up_primary,
+ t->up,
+ up,
+ t->size,
+ size,
+ t->min_size,
+ min_size,
+ t->pg_num,
+ pg_num,
+ t->pg_num_pending,
+ pg_num_pending,
+ t->sort_bitwise,
+ sort_bitwise,
+ t->recovery_deletes,
+ recovery_deletes,
+ t->peering_crush_bucket_count,
+ pi->peering_crush_bucket_count,
+ t->peering_crush_bucket_target,
+ pi->peering_crush_bucket_target,
+ t->peering_crush_bucket_barrier,
+ pi->peering_crush_bucket_barrier,
+ t->peering_crush_mandatory_member,
+ pi->peering_crush_mandatory_member,
+ prev_pgid)) {
+ force_resend = true;
+ }
+
+ bool unpaused = false;
+ bool should_be_paused = target_should_be_paused(t);
+ if (t->paused && !should_be_paused) {
+ unpaused = true;
+ }
+ if (t->paused != should_be_paused) {
+ ldout(cct, 10) << __func__ << " paused " << t->paused
+ << " -> " << should_be_paused << dendl;
+ t->paused = should_be_paused;
+ }
+
+ bool legacy_change =
+ t->pgid != pgid ||
+ is_pg_changed(
+ t->acting_primary, t->acting, acting_primary, acting,
+ t->used_replica || any_change);
+ bool split_or_merge = false;
+ if (t->pg_num) {
+ split_or_merge =
+ prev_pgid.is_split(t->pg_num, pg_num, nullptr) ||
+ prev_pgid.is_merge_source(t->pg_num, pg_num, nullptr) ||
+ prev_pgid.is_merge_target(t->pg_num, pg_num);
+ }
+
+ if (legacy_change || split_or_merge || force_resend) {
+ t->pgid = pgid;
+ t->acting = acting;
+ t->acting_primary = acting_primary;
+ t->up_primary = up_primary;
+ t->up = up;
+ t->size = size;
+ t->min_size = min_size;
+ t->pg_num = pg_num;
+ t->pg_num_mask = pg_num_mask;
+ t->pg_num_pending = pg_num_pending;
+ spg_t spgid(actual_pgid);
+ if (pi->is_erasure()) {
+ for (uint8_t i = 0; i < acting.size(); ++i) {
+ if (acting[i] == acting_primary) {
+ spgid.reset_shard(shard_id_t(i));
+ break;
+ }
+ }
+ }
+ t->actual_pgid = spgid;
+ t->sort_bitwise = sort_bitwise;
+ t->recovery_deletes = recovery_deletes;
+ t->peering_crush_bucket_count = pi->peering_crush_bucket_count;
+ t->peering_crush_bucket_target = pi->peering_crush_bucket_target;
+ t->peering_crush_bucket_barrier = pi->peering_crush_bucket_barrier;
+ t->peering_crush_mandatory_member = pi->peering_crush_mandatory_member;
+ ldout(cct, 10) << __func__ << " "
+ << " raw pgid " << pgid << " -> actual " << t->actual_pgid
+ << " acting " << acting
+ << " primary " << acting_primary << dendl;
+ t->used_replica = false;
+ if ((t->flags & (CEPH_OSD_FLAG_BALANCE_READS |
+ CEPH_OSD_FLAG_LOCALIZE_READS)) &&
+ !is_write && pi->is_replicated() && acting.size() > 1) {
+ int osd;
+ ceph_assert(is_read && acting[0] == acting_primary);
+ if (t->flags & CEPH_OSD_FLAG_BALANCE_READS) {
+ int p = rand() % acting.size();
+ if (p)
+ t->used_replica = true;
+ osd = acting[p];
+ ldout(cct, 10) << " chose random osd." << osd << " of " << acting
+ << dendl;
+ } else {
+ // look for a local replica. prefer the primary if the
+ // distance is the same.
+ int best = -1;
+ int best_locality = 0;
+ for (unsigned i = 0; i < acting.size(); ++i) {
+ int locality = osdmap->crush->get_common_ancestor_distance(
+ cct, acting[i], crush_location);
+ ldout(cct, 20) << __func__ << " localize: rank " << i
+ << " osd." << acting[i]
+ << " locality " << locality << dendl;
+ if (i == 0 ||
+ (locality >= 0 && best_locality >= 0 &&
+ locality < best_locality) ||
+ (best_locality < 0 && locality >= 0)) {
+ best = i;
+ best_locality = locality;
+ if (i)
+ t->used_replica = true;
+ }
+ }
+ ceph_assert(best >= 0);
+ osd = acting[best];
+ }
+ t->osd = osd;
+ } else {
+ t->osd = acting_primary;
+ }
+ }
+ if (legacy_change || unpaused || force_resend) {
+ return RECALC_OP_TARGET_NEED_RESEND;
+ }
+ if (split_or_merge &&
+ (osdmap->require_osd_release >= ceph_release_t::luminous ||
+ HAVE_FEATURE(osdmap->get_xinfo(acting_primary).features,
+ RESEND_ON_SPLIT))) {
+ return RECALC_OP_TARGET_NEED_RESEND;
+ }
+ return RECALC_OP_TARGET_NO_ACTION;
+}
+
+int Objecter::_map_session(op_target_t *target, OSDSession **s,
+ shunique_lock<ceph::shared_mutex>& sul)
+{
+ _calc_target(target, nullptr);
+ return _get_session(target->osd, s, sul);
+}
+
+void Objecter::_session_op_assign(OSDSession *to, Op *op)
+{
+ // to->lock is locked
+ ceph_assert(op->session == NULL);
+ ceph_assert(op->tid);
+
+ get_session(to);
+ op->session = to;
+ to->ops[op->tid] = op;
+
+ if (to->is_homeless()) {
+ num_homeless_ops++;
+ }
+
+ ldout(cct, 15) << __func__ << " " << to->osd << " " << op->tid << dendl;
+}
+
+void Objecter::_session_op_remove(OSDSession *from, Op *op)
+{
+ ceph_assert(op->session == from);
+ // from->lock is locked
+
+ if (from->is_homeless()) {
+ num_homeless_ops--;
+ }
+
+ from->ops.erase(op->tid);
+ put_session(from);
+ op->session = NULL;
+
+ ldout(cct, 15) << __func__ << " " << from->osd << " " << op->tid << dendl;
+}
+
+void Objecter::_session_linger_op_assign(OSDSession *to, LingerOp *op)
+{
+ // to lock is locked unique
+ ceph_assert(op->session == NULL);
+
+ if (to->is_homeless()) {
+ num_homeless_ops++;
+ }
+
+ get_session(to);
+ op->session = to;
+ to->linger_ops[op->linger_id] = op;
+
+ ldout(cct, 15) << __func__ << " " << to->osd << " " << op->linger_id
+ << dendl;
+}
+
+void Objecter::_session_linger_op_remove(OSDSession *from, LingerOp *op)
+{
+ ceph_assert(from == op->session);
+ // from->lock is locked unique
+
+ if (from->is_homeless()) {
+ num_homeless_ops--;
+ }
+
+ from->linger_ops.erase(op->linger_id);
+ put_session(from);
+ op->session = NULL;
+
+ ldout(cct, 15) << __func__ << " " << from->osd << " " << op->linger_id
+ << dendl;
+}
+
+void Objecter::_session_command_op_remove(OSDSession *from, CommandOp *op)
+{
+ ceph_assert(from == op->session);
+ // from->lock is locked
+
+ if (from->is_homeless()) {
+ num_homeless_ops--;
+ }
+
+ from->command_ops.erase(op->tid);
+ put_session(from);
+ op->session = NULL;
+
+ ldout(cct, 15) << __func__ << " " << from->osd << " " << op->tid << dendl;
+}
+
+void Objecter::_session_command_op_assign(OSDSession *to, CommandOp *op)
+{
+ // to->lock is locked
+ ceph_assert(op->session == NULL);
+ ceph_assert(op->tid);
+
+ if (to->is_homeless()) {
+ num_homeless_ops++;
+ }
+
+ get_session(to);
+ op->session = to;
+ to->command_ops[op->tid] = op;
+
+ ldout(cct, 15) << __func__ << " " << to->osd << " " << op->tid << dendl;
+}
+
+int Objecter::_recalc_linger_op_target(LingerOp *linger_op,
+ shunique_lock<ceph::shared_mutex>& sul)
+{
+ // rwlock is locked unique
+
+ int r = _calc_target(&linger_op->target, nullptr, true);
+ if (r == RECALC_OP_TARGET_NEED_RESEND) {
+ ldout(cct, 10) << "recalc_linger_op_target tid " << linger_op->linger_id
+ << " pgid " << linger_op->target.pgid
+ << " acting " << linger_op->target.acting << dendl;
+
+ OSDSession *s = NULL;
+ r = _get_session(linger_op->target.osd, &s, sul);
+ ceph_assert(r == 0);
+
+ if (linger_op->session != s) {
+ // NB locking two sessions (s and linger_op->session) at the
+ // same time here is only safe because we are the only one that
+ // takes two, and we are holding rwlock for write. We use
+ // std::shared_mutex in OSDSession because lockdep doesn't know
+ // that.
+ unique_lock sl(s->lock);
+ _session_linger_op_remove(linger_op->session, linger_op);
+ _session_linger_op_assign(s, linger_op);
+ }
+
+ put_session(s);
+ return RECALC_OP_TARGET_NEED_RESEND;
+ }
+ return r;
+}
+
+void Objecter::_cancel_linger_op(Op *op)
+{
+ ldout(cct, 15) << "cancel_op " << op->tid << dendl;
+
+ ceph_assert(!op->should_resend);
+ if (op->has_completion()) {
+ op->onfinish = nullptr;
+ num_in_flight--;
+ }
+
+ _finish_op(op, 0);
+}
+
+void Objecter::_finish_op(Op *op, int r)
+{
+ ldout(cct, 15) << __func__ << " " << op->tid << dendl;
+
+ // op->session->lock is locked unique or op->session is null
+
+ if (!op->ctx_budgeted && op->budget >= 0) {
+ put_op_budget_bytes(op->budget);
+ op->budget = -1;
+ }
+
+ if (op->ontimeout && r != -ETIMEDOUT)
+ timer.cancel_event(op->ontimeout);
+
+ if (op->session) {
+ _session_op_remove(op->session, op);
+ }
+
+ logger->dec(l_osdc_op_active);
+
+ ceph_assert(check_latest_map_ops.find(op->tid) == check_latest_map_ops.end());
+
+ inflight_ops--;
+
+ op->put();
+}
+
+Objecter::MOSDOp *Objecter::_prepare_osd_op(Op *op)
+{
+ // rwlock is locked
+
+ int flags = op->target.flags;
+ flags |= CEPH_OSD_FLAG_KNOWN_REDIR;
+
+ // Nothing checks this any longer, but needed for compatibility with
+ // pre-luminous osds
+ flags |= CEPH_OSD_FLAG_ONDISK;
+
+ if (!honor_pool_full)
+ flags |= CEPH_OSD_FLAG_FULL_FORCE;
+
+ op->target.paused = false;
+ op->stamp = ceph::coarse_mono_clock::now();
+
+ hobject_t hobj = op->target.get_hobj();
+ auto m = new MOSDOp(client_inc, op->tid,
+ hobj, op->target.actual_pgid,
+ osdmap->get_epoch(),
+ flags, op->features);
+
+ m->set_snapid(op->snapid);
+ m->set_snap_seq(op->snapc.seq);
+ m->set_snaps(op->snapc.snaps);
+
+ m->ops = op->ops;
+ m->set_mtime(op->mtime);
+ m->set_retry_attempt(op->attempts++);
+
+ if (!op->trace.valid() && cct->_conf->osdc_blkin_trace_all) {
+ op->trace.init("op", &trace_endpoint);
+ }
+
+ if (op->priority)
+ m->set_priority(op->priority);
+ else
+ m->set_priority(cct->_conf->osd_client_op_priority);
+
+ if (op->reqid != osd_reqid_t()) {
+ m->set_reqid(op->reqid);
+ }
+
+ logger->inc(l_osdc_op_send);
+ ssize_t sum = 0;
+ for (unsigned i = 0; i < m->ops.size(); i++) {
+ sum += m->ops[i].indata.length();
+ }
+ logger->inc(l_osdc_op_send_bytes, sum);
+
+ return m;
+}
+
+void Objecter::_send_op(Op *op)
+{
+ // rwlock is locked
+ // op->session->lock is locked
+
+ // backoff?
+ auto p = op->session->backoffs.find(op->target.actual_pgid);
+ if (p != op->session->backoffs.end()) {
+ hobject_t hoid = op->target.get_hobj();
+ auto q = p->second.lower_bound(hoid);
+ if (q != p->second.begin()) {
+ --q;
+ if (hoid >= q->second.end) {
+ ++q;
+ }
+ }
+ if (q != p->second.end()) {
+ ldout(cct, 20) << __func__ << " ? " << q->first << " [" << q->second.begin
+ << "," << q->second.end << ")" << dendl;
+ int r = cmp(hoid, q->second.begin);
+ if (r == 0 || (r > 0 && hoid < q->second.end)) {
+ ldout(cct, 10) << __func__ << " backoff " << op->target.actual_pgid
+ << " id " << q->second.id << " on " << hoid
+ << ", queuing " << op << " tid " << op->tid << dendl;
+ return;
+ }
+ }
+ }
+
+ ceph_assert(op->tid > 0);
+ MOSDOp *m = _prepare_osd_op(op);
+
+ if (op->target.actual_pgid != m->get_spg()) {
+ ldout(cct, 10) << __func__ << " " << op->tid << " pgid change from "
+ << m->get_spg() << " to " << op->target.actual_pgid
+ << ", updating and reencoding" << dendl;
+ m->set_spg(op->target.actual_pgid);
+ m->clear_payload(); // reencode
+ }
+
+ ldout(cct, 15) << "_send_op " << op->tid << " to "
+ << op->target.actual_pgid << " on osd." << op->session->osd
+ << dendl;
+
+ ConnectionRef con = op->session->con;
+ ceph_assert(con);
+
+#if 0
+ // preallocated rx ceph::buffer?
+ if (op->con) {
+ ldout(cct, 20) << " revoking rx ceph::buffer for " << op->tid << " on "
+ << op->con << dendl;
+ op->con->revoke_rx_buffer(op->tid);
+ }
+ if (op->outbl &&
+ op->ontimeout == 0 && // only post rx_buffer if no timeout; see #9582
+ op->outbl->length()) {
+ op->outbl->invalidate_crc(); // messenger writes through c_str()
+ ldout(cct, 20) << " posting rx ceph::buffer for " << op->tid << " on " << con
+ << dendl;
+ op->con = con;
+ op->con->post_rx_buffer(op->tid, *op->outbl);
+ }
+#endif
+
+ op->incarnation = op->session->incarnation;
+
+ if (op->trace.valid()) {
+ m->trace.init("op msg", nullptr, &op->trace);
+ }
+ op->session->con->send_message(m);
+}
+
+int Objecter::calc_op_budget(const bc::small_vector_base<OSDOp>& ops)
+{
+ int op_budget = 0;
+ for (auto i = ops.begin(); i != ops.end(); ++i) {
+ if (i->op.op & CEPH_OSD_OP_MODE_WR) {
+ op_budget += i->indata.length();
+ } else if (ceph_osd_op_mode_read(i->op.op)) {
+ if (ceph_osd_op_uses_extent(i->op.op)) {
+ if ((int64_t)i->op.extent.length > 0)
+ op_budget += (int64_t)i->op.extent.length;
+ } else if (ceph_osd_op_type_attr(i->op.op)) {
+ op_budget += i->op.xattr.name_len + i->op.xattr.value_len;
+ }
+ }
+ }
+ return op_budget;
+}
+
+void Objecter::_throttle_op(Op *op,
+ shunique_lock<ceph::shared_mutex>& sul,
+ int op_budget)
+{
+ ceph_assert(sul && sul.mutex() == &rwlock);
+ bool locked_for_write = sul.owns_lock();
+
+ if (!op_budget)
+ op_budget = calc_op_budget(op->ops);
+ if (!op_throttle_bytes.get_or_fail(op_budget)) { //couldn't take right now
+ sul.unlock();
+ op_throttle_bytes.get(op_budget);
+ if (locked_for_write)
+ sul.lock();
+ else
+ sul.lock_shared();
+ }
+ if (!op_throttle_ops.get_or_fail(1)) { //couldn't take right now
+ sul.unlock();
+ op_throttle_ops.get(1);
+ if (locked_for_write)
+ sul.lock();
+ else
+ sul.lock_shared();
+ }
+}
+
+int Objecter::take_linger_budget(LingerOp *info)
+{
+ return 1;
+}
+
+/* This function DOES put the passed message before returning */
+void Objecter::handle_osd_op_reply(MOSDOpReply *m)
+{
+ ldout(cct, 10) << "in handle_osd_op_reply" << dendl;
+
+ // get pio
+ ceph_tid_t tid = m->get_tid();
+
+ shunique_lock sul(rwlock, ceph::acquire_shared);
+ if (!initialized) {
+ m->put();
+ return;
+ }
+
+ ConnectionRef con = m->get_connection();
+ auto priv = con->get_priv();
+ auto s = static_cast<OSDSession*>(priv.get());
+ if (!s || s->con != con) {
+ ldout(cct, 7) << __func__ << " no session on con " << con << dendl;
+ m->put();
+ return;
+ }
+
+ unique_lock sl(s->lock);
+
+ map<ceph_tid_t, Op *>::iterator iter = s->ops.find(tid);
+ if (iter == s->ops.end()) {
+ ldout(cct, 7) << "handle_osd_op_reply " << tid
+ << (m->is_ondisk() ? " ondisk" : (m->is_onnvram() ?
+ " onnvram" : " ack"))
+ << " ... stray" << dendl;
+ sl.unlock();
+ m->put();
+ return;
+ }
+
+ ldout(cct, 7) << "handle_osd_op_reply " << tid
+ << (m->is_ondisk() ? " ondisk" :
+ (m->is_onnvram() ? " onnvram" : " ack"))
+ << " uv " << m->get_user_version()
+ << " in " << m->get_pg()
+ << " attempt " << m->get_retry_attempt()
+ << dendl;
+ Op *op = iter->second;
+ op->trace.event("osd op reply");
+
+ if (retry_writes_after_first_reply && op->attempts == 1 &&
+ (op->target.flags & CEPH_OSD_FLAG_WRITE)) {
+ ldout(cct, 7) << "retrying write after first reply: " << tid << dendl;
+ if (op->has_completion()) {
+ num_in_flight--;
+ }
+ _session_op_remove(s, op);
+ sl.unlock();
+
+ _op_submit(op, sul, NULL);
+ m->put();
+ return;
+ }
+
+ if (m->get_retry_attempt() >= 0) {
+ if (m->get_retry_attempt() != (op->attempts - 1)) {
+ ldout(cct, 7) << " ignoring reply from attempt "
+ << m->get_retry_attempt()
+ << " from " << m->get_source_inst()
+ << "; last attempt " << (op->attempts - 1) << " sent to "
+ << op->session->con->get_peer_addr() << dendl;
+ m->put();
+ sl.unlock();
+ return;
+ }
+ } else {
+ // we don't know the request attempt because the server is old, so
+ // just accept this one. we may do ACK callbacks we shouldn't
+ // have, but that is better than doing callbacks out of order.
+ }
+
+ decltype(op->onfinish) onfinish;
+
+ int rc = m->get_result();
+
+ if (m->is_redirect_reply()) {
+ ldout(cct, 5) << " got redirect reply; redirecting" << dendl;
+ if (op->has_completion())
+ num_in_flight--;
+ _session_op_remove(s, op);
+ sl.unlock();
+
+ // FIXME: two redirects could race and reorder
+
+ op->tid = 0;
+ m->get_redirect().combine_with_locator(op->target.target_oloc,
+ op->target.target_oid.name);
+ op->target.flags |= (CEPH_OSD_FLAG_REDIRECTED |
+ CEPH_OSD_FLAG_IGNORE_CACHE |
+ CEPH_OSD_FLAG_IGNORE_OVERLAY);
+ _op_submit(op, sul, NULL);
+ m->put();
+ return;
+ }
+
+ if (rc == -EAGAIN) {
+ ldout(cct, 7) << " got -EAGAIN, resubmitting" << dendl;
+ if (op->has_completion())
+ num_in_flight--;
+ _session_op_remove(s, op);
+ sl.unlock();
+
+ op->tid = 0;
+ op->target.flags &= ~(CEPH_OSD_FLAG_BALANCE_READS |
+ CEPH_OSD_FLAG_LOCALIZE_READS);
+ op->target.pgid = pg_t();
+ _op_submit(op, sul, NULL);
+ m->put();
+ return;
+ }
+
+ sul.unlock();
+
+ if (op->objver)
+ *op->objver = m->get_user_version();
+ if (op->reply_epoch)
+ *op->reply_epoch = m->get_map_epoch();
+ if (op->data_offset)
+ *op->data_offset = m->get_header().data_off;
+
+ // got data?
+ if (op->outbl) {
+#if 0
+ if (op->con)
+ op->con->revoke_rx_buffer(op->tid);
+#endif
+ auto& bl = m->get_data();
+ if (op->outbl->length() == bl.length() &&
+ bl.get_num_buffers() <= 1) {
+ // this is here to keep previous users to *relied* on getting data
+ // read into existing buffers happy. Notably,
+ // libradosstriper::RadosStriperImpl::aio_read().
+ ldout(cct,10) << __func__ << " copying resulting " << bl.length()
+ << " into existing ceph::buffer of length " << op->outbl->length()
+ << dendl;
+ cb::list t;
+ t = std::move(*op->outbl);
+ t.invalidate_crc(); // we're overwriting the raw buffers via c_str()
+ bl.begin().copy(bl.length(), t.c_str());
+ op->outbl->substr_of(t, 0, bl.length());
+ } else {
+ m->claim_data(*op->outbl);
+ }
+ op->outbl = 0;
+ }
+
+ // per-op result demuxing
+ vector<OSDOp> out_ops;
+ m->claim_ops(out_ops);
+
+ if (out_ops.size() != op->ops.size())
+ ldout(cct, 0) << "WARNING: tid " << op->tid << " reply ops " << out_ops
+ << " != request ops " << op->ops
+ << " from " << m->get_source_inst() << dendl;
+
+ ceph_assert(op->ops.size() == op->out_bl.size());
+ ceph_assert(op->ops.size() == op->out_rval.size());
+ ceph_assert(op->ops.size() == op->out_ec.size());
+ ceph_assert(op->ops.size() == op->out_handler.size());
+ auto pb = op->out_bl.begin();
+ auto pr = op->out_rval.begin();
+ auto pe = op->out_ec.begin();
+ auto ph = op->out_handler.begin();
+ ceph_assert(op->out_bl.size() == op->out_rval.size());
+ ceph_assert(op->out_bl.size() == op->out_handler.size());
+ auto p = out_ops.begin();
+ for (unsigned i = 0;
+ p != out_ops.end() && pb != op->out_bl.end();
+ ++i, ++p, ++pb, ++pr, ++pe, ++ph) {
+ ldout(cct, 10) << " op " << i << " rval " << p->rval
+ << " len " << p->outdata.length() << dendl;
+ if (*pb)
+ **pb = p->outdata;
+ // set rval before running handlers so that handlers
+ // can change it if e.g. decoding fails
+ if (*pr)
+ **pr = ceph_to_hostos_errno(p->rval);
+ if (*pe)
+ **pe = p->rval < 0 ? bs::error_code(-p->rval, osd_category()) :
+ bs::error_code();
+ if (*ph) {
+ std::move((*ph))(p->rval < 0 ?
+ bs::error_code(-p->rval, osd_category()) :
+ bs::error_code(),
+ p->rval, p->outdata);
+ }
+ }
+
+ // NOTE: we assume that since we only request ONDISK ever we will
+ // only ever get back one (type of) ack ever.
+
+ if (op->has_completion()) {
+ num_in_flight--;
+ onfinish = std::move(op->onfinish);
+ op->onfinish = nullptr;
+ }
+ logger->inc(l_osdc_op_reply);
+
+ /* get it before we call _finish_op() */
+ auto completion_lock = s->get_lock(op->target.base_oid);
+
+ ldout(cct, 15) << "handle_osd_op_reply completed tid " << tid << dendl;
+ _finish_op(op, 0);
+
+ ldout(cct, 5) << num_in_flight << " in flight" << dendl;
+
+ // serialize completions
+ if (completion_lock.mutex()) {
+ completion_lock.lock();
+ }
+ sl.unlock();
+
+ // do callbacks
+ if (Op::has_completion(onfinish)) {
+ Op::complete(std::move(onfinish), osdcode(rc), rc);
+ }
+ if (completion_lock.mutex()) {
+ completion_lock.unlock();
+ }
+
+ m->put();
+}
+
+void Objecter::handle_osd_backoff(MOSDBackoff *m)
+{
+ ldout(cct, 10) << __func__ << " " << *m << dendl;
+ shunique_lock sul(rwlock, ceph::acquire_shared);
+ if (!initialized) {
+ m->put();
+ return;
+ }
+
+ ConnectionRef con = m->get_connection();
+ auto priv = con->get_priv();
+ auto s = static_cast<OSDSession*>(priv.get());
+ if (!s || s->con != con) {
+ ldout(cct, 7) << __func__ << " no session on con " << con << dendl;
+ m->put();
+ return;
+ }
+
+ get_session(s);
+
+ unique_lock sl(s->lock);
+
+ switch (m->op) {
+ case CEPH_OSD_BACKOFF_OP_BLOCK:
+ {
+ // register
+ OSDBackoff& b = s->backoffs[m->pgid][m->begin];
+ s->backoffs_by_id.insert(make_pair(m->id, &b));
+ b.pgid = m->pgid;
+ b.id = m->id;
+ b.begin = m->begin;
+ b.end = m->end;
+
+ // ack with original backoff's epoch so that the osd can discard this if
+ // there was a pg split.
+ auto r = new MOSDBackoff(m->pgid, m->map_epoch,
+ CEPH_OSD_BACKOFF_OP_ACK_BLOCK,
+ m->id, m->begin, m->end);
+ // this priority must match the MOSDOps from _prepare_osd_op
+ r->set_priority(cct->_conf->osd_client_op_priority);
+ con->send_message(r);
+ }
+ break;
+
+ case CEPH_OSD_BACKOFF_OP_UNBLOCK:
+ {
+ auto p = s->backoffs_by_id.find(m->id);
+ if (p != s->backoffs_by_id.end()) {
+ OSDBackoff *b = p->second;
+ if (b->begin != m->begin &&
+ b->end != m->end) {
+ lderr(cct) << __func__ << " got " << m->pgid << " id " << m->id
+ << " unblock on ["
+ << m->begin << "," << m->end << ") but backoff is ["
+ << b->begin << "," << b->end << ")" << dendl;
+ // hrmpf, unblock it anyway.
+ }
+ ldout(cct, 10) << __func__ << " unblock backoff " << b->pgid
+ << " id " << b->id
+ << " [" << b->begin << "," << b->end
+ << ")" << dendl;
+ auto spgp = s->backoffs.find(b->pgid);
+ ceph_assert(spgp != s->backoffs.end());
+ spgp->second.erase(b->begin);
+ if (spgp->second.empty()) {
+ s->backoffs.erase(spgp);
+ }
+ s->backoffs_by_id.erase(p);
+
+ // check for any ops to resend
+ for (auto& q : s->ops) {
+ if (q.second->target.actual_pgid == m->pgid) {
+ int r = q.second->target.contained_by(m->begin, m->end);
+ ldout(cct, 20) << __func__ << " contained_by " << r << " on "
+ << q.second->target.get_hobj() << dendl;
+ if (r) {
+ _send_op(q.second);
+ }
+ }
+ }
+ } else {
+ lderr(cct) << __func__ << " " << m->pgid << " id " << m->id
+ << " unblock on ["
+ << m->begin << "," << m->end << ") but backoff dne" << dendl;
+ }
+ }
+ break;
+
+ default:
+ ldout(cct, 10) << __func__ << " unrecognized op " << (int)m->op << dendl;
+ }
+
+ sul.unlock();
+ sl.unlock();
+
+ m->put();
+ put_session(s);
+}
+
+uint32_t Objecter::list_nobjects_seek(NListContext *list_context,
+ uint32_t pos)
+{
+ shared_lock rl(rwlock);
+ list_context->pos = hobject_t(object_t(), string(), CEPH_NOSNAP,
+ pos, list_context->pool_id, string());
+ ldout(cct, 10) << __func__ << " " << list_context
+ << " pos " << pos << " -> " << list_context->pos << dendl;
+ pg_t actual = osdmap->raw_pg_to_pg(pg_t(pos, list_context->pool_id));
+ list_context->current_pg = actual.ps();
+ list_context->at_end_of_pool = false;
+ return pos;
+}
+
+uint32_t Objecter::list_nobjects_seek(NListContext *list_context,
+ const hobject_t& cursor)
+{
+ shared_lock rl(rwlock);
+ ldout(cct, 10) << "list_nobjects_seek " << list_context << dendl;
+ list_context->pos = cursor;
+ list_context->at_end_of_pool = false;
+ pg_t actual = osdmap->raw_pg_to_pg(pg_t(cursor.get_hash(), list_context->pool_id));
+ list_context->current_pg = actual.ps();
+ list_context->sort_bitwise = true;
+ return list_context->current_pg;
+}
+
+void Objecter::list_nobjects_get_cursor(NListContext *list_context,
+ hobject_t *cursor)
+{
+ shared_lock rl(rwlock);
+ if (list_context->list.empty()) {
+ *cursor = list_context->pos;
+ } else {
+ const librados::ListObjectImpl& entry = list_context->list.front();
+ const string *key = (entry.locator.empty() ? &entry.oid : &entry.locator);
+ uint32_t h = osdmap->get_pg_pool(list_context->pool_id)->hash_key(*key, entry.nspace);
+ *cursor = hobject_t(entry.oid, entry.locator, list_context->pool_snap_seq, h, list_context->pool_id, entry.nspace);
+ }
+}
+
+void Objecter::list_nobjects(NListContext *list_context, Context *onfinish)
+{
+ ldout(cct, 10) << __func__ << " pool_id " << list_context->pool_id
+ << " pool_snap_seq " << list_context->pool_snap_seq
+ << " max_entries " << list_context->max_entries
+ << " list_context " << list_context
+ << " onfinish " << onfinish
+ << " current_pg " << list_context->current_pg
+ << " pos " << list_context->pos << dendl;
+
+ shared_lock rl(rwlock);
+ const pg_pool_t *pool = osdmap->get_pg_pool(list_context->pool_id);
+ if (!pool) { // pool is gone
+ rl.unlock();
+ put_nlist_context_budget(list_context);
+ onfinish->complete(-ENOENT);
+ return;
+ }
+ int pg_num = pool->get_pg_num();
+ bool sort_bitwise = osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE);
+
+ if (list_context->pos.is_min()) {
+ list_context->starting_pg_num = 0;
+ list_context->sort_bitwise = sort_bitwise;
+ list_context->starting_pg_num = pg_num;
+ }
+ if (list_context->sort_bitwise != sort_bitwise) {
+ list_context->pos = hobject_t(
+ object_t(), string(), CEPH_NOSNAP,
+ list_context->current_pg, list_context->pool_id, string());
+ list_context->sort_bitwise = sort_bitwise;
+ ldout(cct, 10) << " hobject sort order changed, restarting this pg at "
+ << list_context->pos << dendl;
+ }
+ if (list_context->starting_pg_num != pg_num) {
+ if (!sort_bitwise) {
+ // start reading from the beginning; the pgs have changed
+ ldout(cct, 10) << " pg_num changed; restarting with " << pg_num << dendl;
+ list_context->pos = collection_list_handle_t();
+ }
+ list_context->starting_pg_num = pg_num;
+ }
+
+ if (list_context->pos.is_max()) {
+ ldout(cct, 20) << __func__ << " end of pool, list "
+ << list_context->list << dendl;
+ if (list_context->list.empty()) {
+ list_context->at_end_of_pool = true;
+ }
+ // release the listing context's budget once all
+ // OPs (in the session) are finished
+ put_nlist_context_budget(list_context);
+ onfinish->complete(0);
+ return;
+ }
+
+ ObjectOperation op;
+ op.pg_nls(list_context->max_entries, list_context->filter,
+ list_context->pos, osdmap->get_epoch());
+ list_context->bl.clear();
+ auto onack = new C_NList(list_context, onfinish, this);
+ object_locator_t oloc(list_context->pool_id, list_context->nspace);
+
+ // note current_pg in case we don't have (or lose) SORTBITWISE
+ list_context->current_pg = pool->raw_hash_to_pg(list_context->pos.get_hash());
+ rl.unlock();
+
+ pg_read(list_context->current_pg, oloc, op,
+ &list_context->bl, 0, onack, &onack->epoch,
+ &list_context->ctx_budget);
+}
+
+void Objecter::_nlist_reply(NListContext *list_context, int r,
+ Context *final_finish, epoch_t reply_epoch)
+{
+ ldout(cct, 10) << __func__ << " " << list_context << dendl;
+
+ auto iter = list_context->bl.cbegin();
+ pg_nls_response_t response;
+ decode(response, iter);
+ if (!iter.end()) {
+ // we do this as legacy.
+ cb::list legacy_extra_info;
+ decode(legacy_extra_info, iter);
+ }
+
+ // if the osd returns 1 (newer code), or handle MAX, it means we
+ // hit the end of the pg.
+ if ((response.handle.is_max() || r == 1) &&
+ !list_context->sort_bitwise) {
+ // legacy OSD and !sortbitwise, figure out the next PG on our own
+ ++list_context->current_pg;
+ if (list_context->current_pg == list_context->starting_pg_num) {
+ // end of pool
+ list_context->pos = hobject_t::get_max();
+ } else {
+ // next pg
+ list_context->pos = hobject_t(object_t(), string(), CEPH_NOSNAP,
+ list_context->current_pg,
+ list_context->pool_id, string());
+ }
+ } else {
+ list_context->pos = response.handle;
+ }
+
+ int response_size = response.entries.size();
+ ldout(cct, 20) << " response.entries.size " << response_size
+ << ", response.entries " << response.entries
+ << ", handle " << response.handle
+ << ", tentative new pos " << list_context->pos << dendl;
+ if (response_size) {
+ std::move(response.entries.begin(), response.entries.end(),
+ std::back_inserter(list_context->list));
+ response.entries.clear();
+ }
+
+ if (list_context->list.size() >= list_context->max_entries) {
+ ldout(cct, 20) << " hit max, returning results so far, "
+ << list_context->list << dendl;
+ // release the listing context's budget once all
+ // OPs (in the session) are finished
+ put_nlist_context_budget(list_context);
+ final_finish->complete(0);
+ return;
+ }
+
+ // continue!
+ list_nobjects(list_context, final_finish);
+}
+
+void Objecter::put_nlist_context_budget(NListContext *list_context)
+{
+ if (list_context->ctx_budget >= 0) {
+ ldout(cct, 10) << " release listing context's budget " <<
+ list_context->ctx_budget << dendl;
+ put_op_budget_bytes(list_context->ctx_budget);
+ list_context->ctx_budget = -1;
+ }
+}
+
+// snapshots
+
+void Objecter::create_pool_snap(int64_t pool, std::string_view snap_name,
+ decltype(PoolOp::onfinish)&& onfinish)
+{
+ unique_lock wl(rwlock);
+ ldout(cct, 10) << "create_pool_snap; pool: " << pool << "; snap: "
+ << snap_name << dendl;
+
+ const pg_pool_t *p = osdmap->get_pg_pool(pool);
+ if (!p) {
+ onfinish->defer(std::move(onfinish), osdc_errc::pool_dne, cb::list{});
+ return;
+ }
+ if (p->snap_exists(snap_name)) {
+ onfinish->defer(std::move(onfinish), osdc_errc::snapshot_exists,
+ cb::list{});
+ return;
+ }
+
+ auto op = new PoolOp;
+ op->tid = ++last_tid;
+ op->pool = pool;
+ op->name = snap_name;
+ op->onfinish = std::move(onfinish);
+ op->pool_op = POOL_OP_CREATE_SNAP;
+ pool_ops[op->tid] = op;
+
+ pool_op_submit(op);
+}
+
+struct CB_SelfmanagedSnap {
+ std::unique_ptr<ca::Completion<void(bs::error_code, snapid_t)>> fin;
+ CB_SelfmanagedSnap(decltype(fin)&& fin)
+ : fin(std::move(fin)) {}
+ void operator()(bs::error_code ec, const cb::list& bl) {
+ snapid_t snapid = 0;
+ if (!ec) {
+ try {
+ auto p = bl.cbegin();
+ decode(snapid, p);
+ } catch (const cb::error& e) {
+ ec = e.code();
+ }
+ }
+ fin->defer(std::move(fin), ec, snapid);
+ }
+};
+
+void Objecter::allocate_selfmanaged_snap(
+ int64_t pool,
+ std::unique_ptr<ca::Completion<void(bs::error_code, snapid_t)>> onfinish)
+{
+ unique_lock wl(rwlock);
+ ldout(cct, 10) << "allocate_selfmanaged_snap; pool: " << pool << dendl;
+ auto op = new PoolOp;
+ op->tid = ++last_tid;
+ op->pool = pool;
+ op->onfinish = PoolOp::OpComp::create(
+ service.get_executor(),
+ CB_SelfmanagedSnap(std::move(onfinish)));
+ op->pool_op = POOL_OP_CREATE_UNMANAGED_SNAP;
+ pool_ops[op->tid] = op;
+
+ pool_op_submit(op);
+}
+
+void Objecter::delete_pool_snap(
+ int64_t pool, std::string_view snap_name,
+ decltype(PoolOp::onfinish)&& onfinish)
+{
+ unique_lock wl(rwlock);
+ ldout(cct, 10) << "delete_pool_snap; pool: " << pool << "; snap: "
+ << snap_name << dendl;
+
+ const pg_pool_t *p = osdmap->get_pg_pool(pool);
+ if (!p) {
+ onfinish->defer(std::move(onfinish), osdc_errc::pool_dne, cb::list{});
+ return;
+ }
+
+ if (!p->snap_exists(snap_name)) {
+ onfinish->defer(std::move(onfinish), osdc_errc::snapshot_dne, cb::list{});
+ return;
+ }
+
+ auto op = new PoolOp;
+ op->tid = ++last_tid;
+ op->pool = pool;
+ op->name = snap_name;
+ op->onfinish = std::move(onfinish);
+ op->pool_op = POOL_OP_DELETE_SNAP;
+ pool_ops[op->tid] = op;
+
+ pool_op_submit(op);
+}
+
+void Objecter::delete_selfmanaged_snap(int64_t pool, snapid_t snap,
+ decltype(PoolOp::onfinish)&& onfinish)
+{
+ unique_lock wl(rwlock);
+ ldout(cct, 10) << "delete_selfmanaged_snap; pool: " << pool << "; snap: "
+ << snap << dendl;
+ auto op = new PoolOp;
+ op->tid = ++last_tid;
+ op->pool = pool;
+ op->onfinish = std::move(onfinish);
+ op->pool_op = POOL_OP_DELETE_UNMANAGED_SNAP;
+ op->snapid = snap;
+ pool_ops[op->tid] = op;
+
+ pool_op_submit(op);
+}
+
+void Objecter::create_pool(std::string_view name,
+ decltype(PoolOp::onfinish)&& onfinish,
+ int crush_rule)
+{
+ unique_lock wl(rwlock);
+ ldout(cct, 10) << "create_pool name=" << name << dendl;
+
+ if (osdmap->lookup_pg_pool_name(name) >= 0) {
+ onfinish->defer(std::move(onfinish), osdc_errc::pool_exists, cb::list{});
+ return;
+ }
+
+ auto op = new PoolOp;
+ op->tid = ++last_tid;
+ op->pool = 0;
+ op->name = name;
+ op->onfinish = std::move(onfinish);
+ op->pool_op = POOL_OP_CREATE;
+ pool_ops[op->tid] = op;
+ op->crush_rule = crush_rule;
+
+ pool_op_submit(op);
+}
+
+void Objecter::delete_pool(int64_t pool,
+ decltype(PoolOp::onfinish)&& onfinish)
+{
+ unique_lock wl(rwlock);
+ ldout(cct, 10) << "delete_pool " << pool << dendl;
+
+ if (!osdmap->have_pg_pool(pool))
+ onfinish->defer(std::move(onfinish), osdc_errc::pool_dne, cb::list{});
+ else
+ _do_delete_pool(pool, std::move(onfinish));
+}
+
+void Objecter::delete_pool(std::string_view pool_name,
+ decltype(PoolOp::onfinish)&& onfinish)
+{
+ unique_lock wl(rwlock);
+ ldout(cct, 10) << "delete_pool " << pool_name << dendl;
+
+ int64_t pool = osdmap->lookup_pg_pool_name(pool_name);
+ if (pool < 0)
+ // This only returns one error: -ENOENT.
+ onfinish->defer(std::move(onfinish), osdc_errc::pool_dne, cb::list{});
+ else
+ _do_delete_pool(pool, std::move(onfinish));
+}
+
+void Objecter::_do_delete_pool(int64_t pool,
+ decltype(PoolOp::onfinish)&& onfinish)
+
+{
+ auto op = new PoolOp;
+ op->tid = ++last_tid;
+ op->pool = pool;
+ op->name = "delete";
+ op->onfinish = std::move(onfinish);
+ op->pool_op = POOL_OP_DELETE;
+ pool_ops[op->tid] = op;
+ pool_op_submit(op);
+}
+
+void Objecter::pool_op_submit(PoolOp *op)
+{
+ // rwlock is locked
+ if (mon_timeout > timespan(0)) {
+ op->ontimeout = timer.add_event(mon_timeout,
+ [this, op]() {
+ pool_op_cancel(op->tid, -ETIMEDOUT); });
+ }
+ _pool_op_submit(op);
+}
+
+void Objecter::_pool_op_submit(PoolOp *op)
+{
+ // rwlock is locked unique
+
+ ldout(cct, 10) << "pool_op_submit " << op->tid << dendl;
+ auto m = new MPoolOp(monc->get_fsid(), op->tid, op->pool,
+ op->name, op->pool_op,
+ last_seen_osdmap_version);
+ if (op->snapid) m->snapid = op->snapid;
+ if (op->crush_rule) m->crush_rule = op->crush_rule;
+ monc->send_mon_message(m);
+ op->last_submit = ceph::coarse_mono_clock::now();
+
+ logger->inc(l_osdc_poolop_send);
+}
+
+/**
+ * Handle a reply to a PoolOp message. Check that we sent the message
+ * and give the caller responsibility for the returned cb::list.
+ * Then either call the finisher or stash the PoolOp, depending on if we
+ * have a new enough map.
+ * Lastly, clean up the message and PoolOp.
+ */
+void Objecter::handle_pool_op_reply(MPoolOpReply *m)
+{
+ int rc = m->replyCode;
+ auto ec = rc < 0 ? bs::error_code(-rc, mon_category()) : bs::error_code();
+ FUNCTRACE(cct);
+ shunique_lock sul(rwlock, acquire_shared);
+ if (!initialized) {
+ sul.unlock();
+ m->put();
+ return;
+ }
+
+ ldout(cct, 10) << "handle_pool_op_reply " << *m << dendl;
+ ceph_tid_t tid = m->get_tid();
+ auto iter = pool_ops.find(tid);
+ if (iter != pool_ops.end()) {
+ PoolOp *op = iter->second;
+ ldout(cct, 10) << "have request " << tid << " at " << op << " Op: "
+ << ceph_pool_op_name(op->pool_op) << dendl;
+ cb::list bl{std::move(m->response_data)};
+ if (m->version > last_seen_osdmap_version)
+ last_seen_osdmap_version = m->version;
+ if (osdmap->get_epoch() < m->epoch) {
+ sul.unlock();
+ sul.lock();
+ // recheck op existence since we have let go of rwlock
+ // (for promotion) above.
+ iter = pool_ops.find(tid);
+ if (iter == pool_ops.end())
+ goto done; // op is gone.
+ if (osdmap->get_epoch() < m->epoch) {
+ ldout(cct, 20) << "waiting for client to reach epoch " << m->epoch
+ << " before calling back" << dendl;
+ _wait_for_new_map(OpCompletion::create(
+ service.get_executor(),
+ [o = std::move(op->onfinish),
+ bl = std::move(bl)](
+ bs::error_code ec) mutable {
+ o->defer(std::move(o), ec, bl);
+ }),
+ m->epoch,
+ ec);
+ } else {
+ // map epoch changed, probably because a MOSDMap message
+ // sneaked in. Do caller-specified callback now or else
+ // we lose it forever.
+ ceph_assert(op->onfinish);
+ op->onfinish->defer(std::move(op->onfinish), ec, std::move(bl));
+ }
+ } else {
+ ceph_assert(op->onfinish);
+ op->onfinish->defer(std::move(op->onfinish), ec, std::move(bl));
+ }
+ op->onfinish = nullptr;
+ if (!sul.owns_lock()) {
+ sul.unlock();
+ sul.lock();
+ }
+ iter = pool_ops.find(tid);
+ if (iter != pool_ops.end()) {
+ _finish_pool_op(op, 0);
+ }
+ } else {
+ ldout(cct, 10) << "unknown request " << tid << dendl;
+ }
+
+done:
+ // Not strictly necessary, since we'll release it on return.
+ sul.unlock();
+
+ ldout(cct, 10) << "done" << dendl;
+ m->put();
+}
+
+int Objecter::pool_op_cancel(ceph_tid_t tid, int r)
+{
+ ceph_assert(initialized);
+
+ unique_lock wl(rwlock);
+
+ auto it = pool_ops.find(tid);
+ if (it == pool_ops.end()) {
+ ldout(cct, 10) << __func__ << " tid " << tid << " dne" << dendl;
+ return -ENOENT;
+ }
+
+ ldout(cct, 10) << __func__ << " tid " << tid << dendl;
+
+ PoolOp *op = it->second;
+ if (op->onfinish)
+ op->onfinish->defer(std::move(op->onfinish), osdcode(r), cb::list{});
+
+ _finish_pool_op(op, r);
+ return 0;
+}
+
+void Objecter::_finish_pool_op(PoolOp *op, int r)
+{
+ // rwlock is locked unique
+ pool_ops.erase(op->tid);
+ logger->set(l_osdc_poolop_active, pool_ops.size());
+
+ if (op->ontimeout && r != -ETIMEDOUT) {
+ timer.cancel_event(op->ontimeout);
+ }
+
+ delete op;
+}
+
+// pool stats
+
+void Objecter::get_pool_stats(
+ const std::vector<std::string>& pools,
+ decltype(PoolStatOp::onfinish)&& onfinish)
+{
+ ldout(cct, 10) << "get_pool_stats " << pools << dendl;
+
+ auto op = new PoolStatOp;
+ op->tid = ++last_tid;
+ op->pools = pools;
+ op->onfinish = std::move(onfinish);
+ if (mon_timeout > timespan(0)) {
+ op->ontimeout = timer.add_event(mon_timeout,
+ [this, op]() {
+ pool_stat_op_cancel(op->tid,
+ -ETIMEDOUT); });
+ } else {
+ op->ontimeout = 0;
+ }
+
+ unique_lock wl(rwlock);
+
+ poolstat_ops[op->tid] = op;
+
+ logger->set(l_osdc_poolstat_active, poolstat_ops.size());
+
+ _poolstat_submit(op);
+}
+
+void Objecter::_poolstat_submit(PoolStatOp *op)
+{
+ ldout(cct, 10) << "_poolstat_submit " << op->tid << dendl;
+ monc->send_mon_message(new MGetPoolStats(monc->get_fsid(), op->tid,
+ op->pools,
+ last_seen_pgmap_version));
+ op->last_submit = ceph::coarse_mono_clock::now();
+
+ logger->inc(l_osdc_poolstat_send);
+}
+
+void Objecter::handle_get_pool_stats_reply(MGetPoolStatsReply *m)
+{
+ ldout(cct, 10) << "handle_get_pool_stats_reply " << *m << dendl;
+ ceph_tid_t tid = m->get_tid();
+
+ unique_lock wl(rwlock);
+ if (!initialized) {
+ m->put();
+ return;
+ }
+
+ auto iter = poolstat_ops.find(tid);
+ if (iter != poolstat_ops.end()) {
+ PoolStatOp *op = poolstat_ops[tid];
+ ldout(cct, 10) << "have request " << tid << " at " << op << dendl;
+ if (m->version > last_seen_pgmap_version) {
+ last_seen_pgmap_version = m->version;
+ }
+ op->onfinish->defer(std::move(op->onfinish), bs::error_code{},
+ std::move(m->pool_stats), m->per_pool);
+ _finish_pool_stat_op(op, 0);
+ } else {
+ ldout(cct, 10) << "unknown request " << tid << dendl;
+ }
+ ldout(cct, 10) << "done" << dendl;
+ m->put();
+}
+
+int Objecter::pool_stat_op_cancel(ceph_tid_t tid, int r)
+{
+ ceph_assert(initialized);
+
+ unique_lock wl(rwlock);
+
+ auto it = poolstat_ops.find(tid);
+ if (it == poolstat_ops.end()) {
+ ldout(cct, 10) << __func__ << " tid " << tid << " dne" << dendl;
+ return -ENOENT;
+ }
+
+ ldout(cct, 10) << __func__ << " tid " << tid << dendl;
+
+ auto op = it->second;
+ if (op->onfinish)
+ op->onfinish->defer(std::move(op->onfinish), osdcode(r),
+ bc::flat_map<std::string, pool_stat_t>{}, false);
+ _finish_pool_stat_op(op, r);
+ return 0;
+}
+
+void Objecter::_finish_pool_stat_op(PoolStatOp *op, int r)
+{
+ // rwlock is locked unique
+
+ poolstat_ops.erase(op->tid);
+ logger->set(l_osdc_poolstat_active, poolstat_ops.size());
+
+ if (op->ontimeout && r != -ETIMEDOUT)
+ timer.cancel_event(op->ontimeout);
+
+ delete op;
+}
+
+void Objecter::get_fs_stats(boost::optional<int64_t> poolid,
+ decltype(StatfsOp::onfinish)&& onfinish)
+{
+ ldout(cct, 10) << "get_fs_stats" << dendl;
+ unique_lock l(rwlock);
+
+ auto op = new StatfsOp;
+ op->tid = ++last_tid;
+ op->data_pool = poolid;
+ op->onfinish = std::move(onfinish);
+ if (mon_timeout > timespan(0)) {
+ op->ontimeout = timer.add_event(mon_timeout,
+ [this, op]() {
+ statfs_op_cancel(op->tid,
+ -ETIMEDOUT); });
+ } else {
+ op->ontimeout = 0;
+ }
+ statfs_ops[op->tid] = op;
+
+ logger->set(l_osdc_statfs_active, statfs_ops.size());
+
+ _fs_stats_submit(op);
+}
+
+void Objecter::_fs_stats_submit(StatfsOp *op)
+{
+ // rwlock is locked unique
+
+ ldout(cct, 10) << "fs_stats_submit" << op->tid << dendl;
+ monc->send_mon_message(new MStatfs(monc->get_fsid(), op->tid,
+ op->data_pool,
+ last_seen_pgmap_version));
+ op->last_submit = ceph::coarse_mono_clock::now();
+
+ logger->inc(l_osdc_statfs_send);
+}
+
+void Objecter::handle_fs_stats_reply(MStatfsReply *m)
+{
+ unique_lock wl(rwlock);
+ if (!initialized) {
+ m->put();
+ return;
+ }
+
+ ldout(cct, 10) << "handle_fs_stats_reply " << *m << dendl;
+ ceph_tid_t tid = m->get_tid();
+
+ if (statfs_ops.count(tid)) {
+ StatfsOp *op = statfs_ops[tid];
+ ldout(cct, 10) << "have request " << tid << " at " << op << dendl;
+ if (m->h.version > last_seen_pgmap_version)
+ last_seen_pgmap_version = m->h.version;
+ op->onfinish->defer(std::move(op->onfinish), bs::error_code{}, m->h.st);
+ _finish_statfs_op(op, 0);
+ } else {
+ ldout(cct, 10) << "unknown request " << tid << dendl;
+ }
+ m->put();
+ ldout(cct, 10) << "done" << dendl;
+}
+
+int Objecter::statfs_op_cancel(ceph_tid_t tid, int r)
+{
+ ceph_assert(initialized);
+
+ unique_lock wl(rwlock);
+
+ auto it = statfs_ops.find(tid);
+ if (it == statfs_ops.end()) {
+ ldout(cct, 10) << __func__ << " tid " << tid << " dne" << dendl;
+ return -ENOENT;
+ }
+
+ ldout(cct, 10) << __func__ << " tid " << tid << dendl;
+
+ auto op = it->second;
+ if (op->onfinish)
+ op->onfinish->defer(std::move(op->onfinish), osdcode(r), ceph_statfs{});
+ _finish_statfs_op(op, r);
+ return 0;
+}
+
+void Objecter::_finish_statfs_op(StatfsOp *op, int r)
+{
+ // rwlock is locked unique
+
+ statfs_ops.erase(op->tid);
+ logger->set(l_osdc_statfs_active, statfs_ops.size());
+
+ if (op->ontimeout && r != -ETIMEDOUT)
+ timer.cancel_event(op->ontimeout);
+
+ delete op;
+}
+
+// scatter/gather
+
+void Objecter::_sg_read_finish(vector<ObjectExtent>& extents,
+ vector<cb::list>& resultbl,
+ cb::list *bl, Context *onfinish)
+{
+ // all done
+ ldout(cct, 15) << "_sg_read_finish" << dendl;
+
+ if (extents.size() > 1) {
+ Striper::StripedReadResult r;
+ auto bit = resultbl.begin();
+ for (auto eit = extents.begin();
+ eit != extents.end();
+ ++eit, ++bit) {
+ r.add_partial_result(cct, *bit, eit->buffer_extents);
+ }
+ bl->clear();
+ r.assemble_result(cct, *bl, false);
+ } else {
+ ldout(cct, 15) << " only one frag" << dendl;
+ *bl = std::move(resultbl[0]);
+ }
+
+ // done
+ uint64_t bytes_read = bl->length();
+ ldout(cct, 7) << "_sg_read_finish " << bytes_read << " bytes" << dendl;
+
+ if (onfinish) {
+ onfinish->complete(bytes_read);// > 0 ? bytes_read:m->get_result());
+ }
+}
+
+
+void Objecter::ms_handle_connect(Connection *con)
+{
+ ldout(cct, 10) << "ms_handle_connect " << con << dendl;
+ if (!initialized)
+ return;
+
+ if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON)
+ resend_mon_ops();
+}
+
+bool Objecter::ms_handle_reset(Connection *con)
+{
+ if (!initialized)
+ return false;
+ if (con->get_peer_type() == CEPH_ENTITY_TYPE_OSD) {
+ unique_lock wl(rwlock);
+
+ auto priv = con->get_priv();
+ auto session = static_cast<OSDSession*>(priv.get());
+ if (session) {
+ ldout(cct, 1) << "ms_handle_reset " << con << " session " << session
+ << " osd." << session->osd << dendl;
+ // the session maybe had been closed if new osdmap just handled
+ // says the osd down
+ if (!(initialized && osdmap->is_up(session->osd))) {
+ ldout(cct, 1) << "ms_handle_reset aborted,initialized=" << initialized << dendl;
+ wl.unlock();
+ return false;
+ }
+ map<uint64_t, LingerOp *> lresend;
+ unique_lock sl(session->lock);
+ _reopen_session(session);
+ _kick_requests(session, lresend);
+ sl.unlock();
+ _linger_ops_resend(lresend, wl);
+ wl.unlock();
+ maybe_request_map();
+ }
+ return true;
+ }
+ return false;
+}
+
+void Objecter::ms_handle_remote_reset(Connection *con)
+{
+ /*
+ * treat these the same.
+ */
+ ms_handle_reset(con);
+}
+
+bool Objecter::ms_handle_refused(Connection *con)
+{
+ // just log for now
+ if (osdmap && (con->get_peer_type() == CEPH_ENTITY_TYPE_OSD)) {
+ int osd = osdmap->identify_osd(con->get_peer_addr());
+ if (osd >= 0) {
+ ldout(cct, 1) << "ms_handle_refused on osd." << osd << dendl;
+ }
+ }
+ return false;
+}
+
+void Objecter::op_target_t::dump(Formatter *f) const
+{
+ f->dump_stream("pg") << pgid;
+ f->dump_int("osd", osd);
+ f->dump_stream("object_id") << base_oid;
+ f->dump_stream("object_locator") << base_oloc;
+ f->dump_stream("target_object_id") << target_oid;
+ f->dump_stream("target_object_locator") << target_oloc;
+ f->dump_int("paused", (int)paused);
+ f->dump_int("used_replica", (int)used_replica);
+ f->dump_int("precalc_pgid", (int)precalc_pgid);
+}
+
+void Objecter::_dump_active(OSDSession *s)
+{
+ for (auto p = s->ops.begin(); p != s->ops.end(); ++p) {
+ Op *op = p->second;
+ ldout(cct, 20) << op->tid << "\t" << op->target.pgid
+ << "\tosd." << (op->session ? op->session->osd : -1)
+ << "\t" << op->target.base_oid
+ << "\t" << op->ops << dendl;
+ }
+}
+
+void Objecter::_dump_active()
+{
+ ldout(cct, 20) << "dump_active .. " << num_homeless_ops << " homeless"
+ << dendl;
+ for (auto siter = osd_sessions.begin();
+ siter != osd_sessions.end(); ++siter) {
+ auto s = siter->second;
+ shared_lock sl(s->lock);
+ _dump_active(s);
+ sl.unlock();
+ }
+ _dump_active(homeless_session);
+}
+
+void Objecter::dump_active()
+{
+ shared_lock rl(rwlock);
+ _dump_active();
+ rl.unlock();
+}
+
+void Objecter::dump_requests(Formatter *fmt)
+{
+ // Read-lock on Objecter held here
+ fmt->open_object_section("requests");
+ dump_ops(fmt);
+ dump_linger_ops(fmt);
+ dump_pool_ops(fmt);
+ dump_pool_stat_ops(fmt);
+ dump_statfs_ops(fmt);
+ dump_command_ops(fmt);
+ fmt->close_section(); // requests object
+}
+
+void Objecter::_dump_ops(const OSDSession *s, Formatter *fmt)
+{
+ for (auto p = s->ops.begin(); p != s->ops.end(); ++p) {
+ Op *op = p->second;
+ auto age = std::chrono::duration<double>(ceph::coarse_mono_clock::now() - op->stamp);
+ fmt->open_object_section("op");
+ fmt->dump_unsigned("tid", op->tid);
+ op->target.dump(fmt);
+ fmt->dump_stream("last_sent") << op->stamp;
+ fmt->dump_float("age", age.count());
+ fmt->dump_int("attempts", op->attempts);
+ fmt->dump_stream("snapid") << op->snapid;
+ fmt->dump_stream("snap_context") << op->snapc;
+ fmt->dump_stream("mtime") << op->mtime;
+
+ fmt->open_array_section("osd_ops");
+ for (auto it = op->ops.begin(); it != op->ops.end(); ++it) {
+ fmt->dump_stream("osd_op") << *it;
+ }
+ fmt->close_section(); // osd_ops array
+
+ fmt->close_section(); // op object
+ }
+}
+
+void Objecter::dump_ops(Formatter *fmt)
+{
+ // Read-lock on Objecter held
+ fmt->open_array_section("ops");
+ for (auto siter = osd_sessions.begin();
+ siter != osd_sessions.end(); ++siter) {
+ OSDSession *s = siter->second;
+ shared_lock sl(s->lock);
+ _dump_ops(s, fmt);
+ sl.unlock();
+ }
+ _dump_ops(homeless_session, fmt);
+ fmt->close_section(); // ops array
+}
+
+void Objecter::_dump_linger_ops(const OSDSession *s, Formatter *fmt)
+{
+ for (auto p = s->linger_ops.begin(); p != s->linger_ops.end(); ++p) {
+ auto op = p->second;
+ fmt->open_object_section("linger_op");
+ fmt->dump_unsigned("linger_id", op->linger_id);
+ op->target.dump(fmt);
+ fmt->dump_stream("snapid") << op->snap;
+ fmt->dump_stream("registered") << op->registered;
+ fmt->close_section(); // linger_op object
+ }
+}
+
+void Objecter::dump_linger_ops(Formatter *fmt)
+{
+ // We have a read-lock on the objecter
+ fmt->open_array_section("linger_ops");
+ for (auto siter = osd_sessions.begin();
+ siter != osd_sessions.end(); ++siter) {
+ auto s = siter->second;
+ shared_lock sl(s->lock);
+ _dump_linger_ops(s, fmt);
+ sl.unlock();
+ }
+ _dump_linger_ops(homeless_session, fmt);
+ fmt->close_section(); // linger_ops array
+}
+
+void Objecter::_dump_command_ops(const OSDSession *s, Formatter *fmt)
+{
+ for (auto p = s->command_ops.begin(); p != s->command_ops.end(); ++p) {
+ auto op = p->second;
+ fmt->open_object_section("command_op");
+ fmt->dump_unsigned("command_id", op->tid);
+ fmt->dump_int("osd", op->session ? op->session->osd : -1);
+ fmt->open_array_section("command");
+ for (auto q = op->cmd.begin(); q != op->cmd.end(); ++q)
+ fmt->dump_string("word", *q);
+ fmt->close_section();
+ if (op->target_osd >= 0)
+ fmt->dump_int("target_osd", op->target_osd);
+ else
+ fmt->dump_stream("target_pg") << op->target_pg;
+ fmt->close_section(); // command_op object
+ }
+}
+
+void Objecter::dump_command_ops(Formatter *fmt)
+{
+ // We have a read-lock on the Objecter here
+ fmt->open_array_section("command_ops");
+ for (auto siter = osd_sessions.begin();
+ siter != osd_sessions.end(); ++siter) {
+ auto s = siter->second;
+ shared_lock sl(s->lock);
+ _dump_command_ops(s, fmt);
+ sl.unlock();
+ }
+ _dump_command_ops(homeless_session, fmt);
+ fmt->close_section(); // command_ops array
+}
+
+void Objecter::dump_pool_ops(Formatter *fmt) const
+{
+ fmt->open_array_section("pool_ops");
+ for (auto p = pool_ops.begin(); p != pool_ops.end(); ++p) {
+ auto op = p->second;
+ fmt->open_object_section("pool_op");
+ fmt->dump_unsigned("tid", op->tid);
+ fmt->dump_int("pool", op->pool);
+ fmt->dump_string("name", op->name);
+ fmt->dump_int("operation_type", op->pool_op);
+ fmt->dump_unsigned("crush_rule", op->crush_rule);
+ fmt->dump_stream("snapid") << op->snapid;
+ fmt->dump_stream("last_sent") << op->last_submit;
+ fmt->close_section(); // pool_op object
+ }
+ fmt->close_section(); // pool_ops array
+}
+
+void Objecter::dump_pool_stat_ops(Formatter *fmt) const
+{
+ fmt->open_array_section("pool_stat_ops");
+ for (auto p = poolstat_ops.begin();
+ p != poolstat_ops.end();
+ ++p) {
+ PoolStatOp *op = p->second;
+ fmt->open_object_section("pool_stat_op");
+ fmt->dump_unsigned("tid", op->tid);
+ fmt->dump_stream("last_sent") << op->last_submit;
+
+ fmt->open_array_section("pools");
+ for (const auto& it : op->pools) {
+ fmt->dump_string("pool", it);
+ }
+ fmt->close_section(); // pools array
+
+ fmt->close_section(); // pool_stat_op object
+ }
+ fmt->close_section(); // pool_stat_ops array
+}
+
+void Objecter::dump_statfs_ops(Formatter *fmt) const
+{
+ fmt->open_array_section("statfs_ops");
+ for (auto p = statfs_ops.begin(); p != statfs_ops.end(); ++p) {
+ auto op = p->second;
+ fmt->open_object_section("statfs_op");
+ fmt->dump_unsigned("tid", op->tid);
+ fmt->dump_stream("last_sent") << op->last_submit;
+ fmt->close_section(); // statfs_op object
+ }
+ fmt->close_section(); // statfs_ops array
+}
+
+Objecter::RequestStateHook::RequestStateHook(Objecter *objecter) :
+ m_objecter(objecter)
+{
+}
+
+int Objecter::RequestStateHook::call(std::string_view command,
+ const cmdmap_t& cmdmap,
+ Formatter *f,
+ std::ostream& ss,
+ cb::list& out)
+{
+ shared_lock rl(m_objecter->rwlock);
+ m_objecter->dump_requests(f);
+ return 0;
+}
+
+void Objecter::blocklist_self(bool set)
+{
+ ldout(cct, 10) << "blocklist_self " << (set ? "add" : "rm") << dendl;
+
+ vector<string> cmd;
+ cmd.push_back("{\"prefix\":\"osd blocklist\", ");
+ if (set)
+ cmd.push_back("\"blocklistop\":\"add\",");
+ else
+ cmd.push_back("\"blocklistop\":\"rm\",");
+ stringstream ss;
+ // this is somewhat imprecise in that we are blocklisting our first addr only
+ ss << messenger->get_myaddrs().front().get_legacy_str();
+ cmd.push_back("\"addr\":\"" + ss.str() + "\"");
+
+ auto m = new MMonCommand(monc->get_fsid());
+ m->cmd = cmd;
+
+ // NOTE: no fallback to legacy blacklist command implemented here
+ // since this is only used for test code.
+
+ monc->send_mon_message(m);
+}
+
+// commands
+
+void Objecter::handle_command_reply(MCommandReply *m)
+{
+ unique_lock wl(rwlock);
+ if (!initialized) {
+ m->put();
+ return;
+ }
+
+ ConnectionRef con = m->get_connection();
+ auto priv = con->get_priv();
+ auto s = static_cast<OSDSession*>(priv.get());
+ if (!s || s->con != con) {
+ ldout(cct, 7) << __func__ << " no session on con " << con << dendl;
+ m->put();
+ return;
+ }
+
+ shared_lock sl(s->lock);
+ auto p = s->command_ops.find(m->get_tid());
+ if (p == s->command_ops.end()) {
+ ldout(cct, 10) << "handle_command_reply tid " << m->get_tid()
+ << " not found" << dendl;
+ m->put();
+ sl.unlock();
+ return;
+ }
+
+ CommandOp *c = p->second;
+ if (!c->session ||
+ m->get_connection() != c->session->con) {
+ ldout(cct, 10) << "handle_command_reply tid " << m->get_tid()
+ << " got reply from wrong connection "
+ << m->get_connection() << " " << m->get_source_inst()
+ << dendl;
+ m->put();
+ sl.unlock();
+ return;
+ }
+
+ if (m->r == -EAGAIN) {
+ ldout(cct,10) << __func__ << " tid " << m->get_tid()
+ << " got EAGAIN, requesting map and resending" << dendl;
+ // NOTE: This might resend twice... once now, and once again when
+ // we get an updated osdmap and the PG is found to have moved.
+ _maybe_request_map();
+ _send_command(c);
+ m->put();
+ sl.unlock();
+ return;
+ }
+
+ sl.unlock();
+
+ unique_lock sul(s->lock);
+ _finish_command(c, m->r < 0 ? bs::error_code(-m->r, osd_category()) :
+ bs::error_code(), std::move(m->rs),
+ std::move(m->get_data()));
+ sul.unlock();
+
+ m->put();
+}
+
+Objecter::LingerOp::LingerOp(Objecter *o, uint64_t linger_id)
+ : objecter(o),
+ linger_id(linger_id),
+ watch_lock(ceph::make_shared_mutex(
+ fmt::format("LingerOp::watch_lock #{}", linger_id)))
+{}
+
+void Objecter::submit_command(CommandOp *c, ceph_tid_t *ptid)
+{
+ shunique_lock sul(rwlock, ceph::acquire_unique);
+
+ ceph_tid_t tid = ++last_tid;
+ ldout(cct, 10) << "_submit_command " << tid << " " << c->cmd << dendl;
+ c->tid = tid;
+
+ {
+ unique_lock hs_wl(homeless_session->lock);
+ _session_command_op_assign(homeless_session, c);
+ }
+
+ _calc_command_target(c, sul);
+ _assign_command_session(c, sul);
+ if (osd_timeout > timespan(0)) {
+ c->ontimeout = timer.add_event(osd_timeout,
+ [this, c, tid]() {
+ command_op_cancel(
+ c->session, tid,
+ osdc_errc::timed_out); });
+ }
+
+ if (!c->session->is_homeless()) {
+ _send_command(c);
+ } else {
+ _maybe_request_map();
+ }
+ if (c->map_check_error)
+ _send_command_map_check(c);
+ if (ptid)
+ *ptid = tid;
+
+ logger->inc(l_osdc_command_active);
+}
+
+int Objecter::_calc_command_target(CommandOp *c,
+ shunique_lock<ceph::shared_mutex>& sul)
+{
+ ceph_assert(sul.owns_lock() && sul.mutex() == &rwlock);
+
+ c->map_check_error = 0;
+
+ // ignore overlays, just like we do with pg ops
+ c->target.flags |= CEPH_OSD_FLAG_IGNORE_OVERLAY;
+
+ if (c->target_osd >= 0) {
+ if (!osdmap->exists(c->target_osd)) {
+ c->map_check_error = -ENOENT;
+ c->map_check_error_str = "osd dne";
+ c->target.osd = -1;
+ return RECALC_OP_TARGET_OSD_DNE;
+ }
+ if (osdmap->is_down(c->target_osd)) {
+ c->map_check_error = -ENXIO;
+ c->map_check_error_str = "osd down";
+ c->target.osd = -1;
+ return RECALC_OP_TARGET_OSD_DOWN;
+ }
+ c->target.osd = c->target_osd;
+ } else {
+ int ret = _calc_target(&(c->target), nullptr, true);
+ if (ret == RECALC_OP_TARGET_POOL_DNE) {
+ c->map_check_error = -ENOENT;
+ c->map_check_error_str = "pool dne";
+ c->target.osd = -1;
+ return ret;
+ } else if (ret == RECALC_OP_TARGET_OSD_DOWN) {
+ c->map_check_error = -ENXIO;
+ c->map_check_error_str = "osd down";
+ c->target.osd = -1;
+ return ret;
+ }
+ }
+
+ OSDSession *s;
+ int r = _get_session(c->target.osd, &s, sul);
+ ceph_assert(r != -EAGAIN); /* shouldn't happen as we're holding the write lock */
+
+ if (c->session != s) {
+ put_session(s);
+ return RECALC_OP_TARGET_NEED_RESEND;
+ }
+
+ put_session(s);
+
+ ldout(cct, 20) << "_recalc_command_target " << c->tid << " no change, "
+ << c->session << dendl;
+
+ return RECALC_OP_TARGET_NO_ACTION;
+}
+
+void Objecter::_assign_command_session(CommandOp *c,
+ shunique_lock<ceph::shared_mutex>& sul)
+{
+ ceph_assert(sul.owns_lock() && sul.mutex() == &rwlock);
+
+ OSDSession *s;
+ int r = _get_session(c->target.osd, &s, sul);
+ ceph_assert(r != -EAGAIN); /* shouldn't happen as we're holding the write lock */
+
+ if (c->session != s) {
+ if (c->session) {
+ OSDSession *cs = c->session;
+ unique_lock csl(cs->lock);
+ _session_command_op_remove(c->session, c);
+ csl.unlock();
+ }
+ unique_lock sl(s->lock);
+ _session_command_op_assign(s, c);
+ }
+
+ put_session(s);
+}
+
+void Objecter::_send_command(CommandOp *c)
+{
+ ldout(cct, 10) << "_send_command " << c->tid << dendl;
+ ceph_assert(c->session);
+ ceph_assert(c->session->con);
+ auto m = new MCommand(monc->monmap.fsid);
+ m->cmd = c->cmd;
+ m->set_data(c->inbl);
+ m->set_tid(c->tid);
+ c->session->con->send_message(m);
+ logger->inc(l_osdc_command_send);
+}
+
+int Objecter::command_op_cancel(OSDSession *s, ceph_tid_t tid,
+ bs::error_code ec)
+{
+ ceph_assert(initialized);
+
+ unique_lock wl(rwlock);
+
+ auto it = s->command_ops.find(tid);
+ if (it == s->command_ops.end()) {
+ ldout(cct, 10) << __func__ << " tid " << tid << " dne" << dendl;
+ return -ENOENT;
+ }
+
+ ldout(cct, 10) << __func__ << " tid " << tid << dendl;
+
+ CommandOp *op = it->second;
+ _command_cancel_map_check(op);
+ unique_lock sl(op->session->lock);
+ _finish_command(op, ec, {}, {});
+ sl.unlock();
+ return 0;
+}
+
+void Objecter::_finish_command(CommandOp *c, bs::error_code ec,
+ string&& rs, cb::list&& bl)
+{
+ // rwlock is locked unique
+ // session lock is locked
+
+ ldout(cct, 10) << "_finish_command " << c->tid << " = " << ec << " "
+ << rs << dendl;
+
+ if (c->onfinish)
+ c->onfinish->defer(std::move(c->onfinish), ec, std::move(rs), std::move(bl));
+
+ if (c->ontimeout && ec != bs::errc::timed_out)
+ timer.cancel_event(c->ontimeout);
+
+ _session_command_op_remove(c->session, c);
+
+ c->put();
+
+ logger->dec(l_osdc_command_active);
+}
+
+Objecter::OSDSession::~OSDSession()
+{
+ // Caller is responsible for re-assigning or
+ // destroying any ops that were assigned to us
+ ceph_assert(ops.empty());
+ ceph_assert(linger_ops.empty());
+ ceph_assert(command_ops.empty());
+}
+
+Objecter::Objecter(CephContext *cct,
+ Messenger *m, MonClient *mc,
+ boost::asio::io_context& service) :
+ Dispatcher(cct), messenger(m), monc(mc), service(service)
+{
+ mon_timeout = cct->_conf.get_val<std::chrono::seconds>("rados_mon_op_timeout");
+ osd_timeout = cct->_conf.get_val<std::chrono::seconds>("rados_osd_op_timeout");
+}
+
+Objecter::~Objecter()
+{
+ ceph_assert(homeless_session->get_nref() == 1);
+ ceph_assert(num_homeless_ops == 0);
+ homeless_session->put();
+
+ ceph_assert(osd_sessions.empty());
+ ceph_assert(poolstat_ops.empty());
+ ceph_assert(statfs_ops.empty());
+ ceph_assert(pool_ops.empty());
+ ceph_assert(waiting_for_map.empty());
+ ceph_assert(linger_ops.empty());
+ ceph_assert(check_latest_map_lingers.empty());
+ ceph_assert(check_latest_map_ops.empty());
+ ceph_assert(check_latest_map_commands.empty());
+
+ ceph_assert(!m_request_state_hook);
+ ceph_assert(!logger);
+}
+
+/**
+ * Wait until this OSD map epoch is received before
+ * sending any more operations to OSDs. Use this
+ * when it is known that the client can't trust
+ * anything from before this epoch (e.g. due to
+ * client blocklist at this epoch).
+ */
+void Objecter::set_epoch_barrier(epoch_t epoch)
+{
+ unique_lock wl(rwlock);
+
+ ldout(cct, 7) << __func__ << ": barrier " << epoch << " (was "
+ << epoch_barrier << ") current epoch " << osdmap->get_epoch()
+ << dendl;
+ if (epoch > epoch_barrier) {
+ epoch_barrier = epoch;
+ _maybe_request_map();
+ }
+}
+
+
+
+hobject_t Objecter::enumerate_objects_begin()
+{
+ return hobject_t();
+}
+
+hobject_t Objecter::enumerate_objects_end()
+{
+ return hobject_t::get_max();
+}
+
+template<typename T>
+struct EnumerationContext {
+ Objecter* objecter;
+ const hobject_t end;
+ const cb::list filter;
+ uint32_t max;
+ const object_locator_t oloc;
+ std::vector<T> ls;
+private:
+ fu2::unique_function<void(bs::error_code,
+ std::vector<T>,
+ hobject_t) &&> on_finish;
+public:
+ epoch_t epoch = 0;
+ int budget = -1;
+
+ EnumerationContext(Objecter* objecter,
+ hobject_t end, cb::list filter,
+ uint32_t max, object_locator_t oloc,
+ decltype(on_finish) on_finish)
+ : objecter(objecter), end(std::move(end)), filter(std::move(filter)),
+ max(max), oloc(std::move(oloc)), on_finish(std::move(on_finish)) {}
+
+ void operator()(bs::error_code ec,
+ std::vector<T> v,
+ hobject_t h) && {
+ if (budget >= 0) {
+ objecter->put_op_budget_bytes(budget);
+ budget = -1;
+ }
+
+ std::move(on_finish)(ec, std::move(v), std::move(h));
+ }
+};
+
+template<typename T>
+struct CB_EnumerateReply {
+ cb::list bl;
+
+ Objecter* objecter;
+ std::unique_ptr<EnumerationContext<T>> ctx;
+
+ CB_EnumerateReply(Objecter* objecter,
+ std::unique_ptr<EnumerationContext<T>>&& ctx) :
+ objecter(objecter), ctx(std::move(ctx)) {}
+
+ void operator()(bs::error_code ec) {
+ objecter->_enumerate_reply(std::move(bl), ec, std::move(ctx));
+ }
+};
+
+template<typename T>
+void Objecter::enumerate_objects(
+ int64_t pool_id,
+ std::string_view ns,
+ hobject_t start,
+ hobject_t end,
+ const uint32_t max,
+ const cb::list& filter_bl,
+ fu2::unique_function<void(bs::error_code,
+ std::vector<T>,
+ hobject_t) &&> on_finish) {
+ if (!end.is_max() && start > end) {
+ lderr(cct) << __func__ << ": start " << start << " > end " << end << dendl;
+ std::move(on_finish)(osdc_errc::precondition_violated, {}, {});
+ return;
+ }
+
+ if (max < 1) {
+ lderr(cct) << __func__ << ": result size may not be zero" << dendl;
+ std::move(on_finish)(osdc_errc::precondition_violated, {}, {});
+ return;
+ }
+
+ if (start.is_max()) {
+ std::move(on_finish)({}, {}, {});
+ return;
+ }
+
+ shared_lock rl(rwlock);
+ ceph_assert(osdmap->get_epoch());
+ if (!osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE)) {
+ rl.unlock();
+ lderr(cct) << __func__ << ": SORTBITWISE cluster flag not set" << dendl;
+ std::move(on_finish)(osdc_errc::not_supported, {}, {});
+ return;
+ }
+ const pg_pool_t* p = osdmap->get_pg_pool(pool_id);
+ if (!p) {
+ lderr(cct) << __func__ << ": pool " << pool_id << " DNE in osd epoch "
+ << osdmap->get_epoch() << dendl;
+ rl.unlock();
+ std::move(on_finish)(osdc_errc::pool_dne, {}, {});
+ return;
+ } else {
+ rl.unlock();
+ }
+
+ _issue_enumerate(start,
+ std::make_unique<EnumerationContext<T>>(
+ this, std::move(end), filter_bl,
+ max, object_locator_t{pool_id, ns},
+ std::move(on_finish)));
+}
+
+template
+void Objecter::enumerate_objects<librados::ListObjectImpl>(
+ int64_t pool_id,
+ std::string_view ns,
+ hobject_t start,
+ hobject_t end,
+ const uint32_t max,
+ const cb::list& filter_bl,
+ fu2::unique_function<void(bs::error_code,
+ std::vector<librados::ListObjectImpl>,
+ hobject_t) &&> on_finish);
+
+template
+void Objecter::enumerate_objects<neorados::Entry>(
+ int64_t pool_id,
+ std::string_view ns,
+ hobject_t start,
+ hobject_t end,
+ const uint32_t max,
+ const cb::list& filter_bl,
+ fu2::unique_function<void(bs::error_code,
+ std::vector<neorados::Entry>,
+ hobject_t) &&> on_finish);
+
+
+
+template<typename T>
+void Objecter::_issue_enumerate(hobject_t start,
+ std::unique_ptr<EnumerationContext<T>> ctx) {
+ ObjectOperation op;
+ auto c = ctx.get();
+ op.pg_nls(c->max, c->filter, start, osdmap->get_epoch());
+ auto on_ack = std::make_unique<CB_EnumerateReply<T>>(this, std::move(ctx));
+ // I hate having to do this. Try to find a cleaner way
+ // later.
+ auto epoch = &c->epoch;
+ auto budget = &c->budget;
+ auto pbl = &on_ack->bl;
+
+ // Issue. See you later in _enumerate_reply
+ pg_read(start.get_hash(),
+ c->oloc, op, pbl, 0,
+ Op::OpComp::create(service.get_executor(),
+ [c = std::move(on_ack)]
+ (bs::error_code ec) mutable {
+ (*c)(ec);
+ }), epoch, budget);
+}
+
+template
+void Objecter::_issue_enumerate<librados::ListObjectImpl>(
+ hobject_t start,
+ std::unique_ptr<EnumerationContext<librados::ListObjectImpl>> ctx);
+template
+void Objecter::_issue_enumerate<neorados::Entry>(
+ hobject_t start, std::unique_ptr<EnumerationContext<neorados::Entry>> ctx);
+
+template<typename T>
+void Objecter::_enumerate_reply(
+ cb::list&& bl,
+ bs::error_code ec,
+ std::unique_ptr<EnumerationContext<T>>&& ctx)
+{
+ if (ec) {
+ std::move(*ctx)(ec, {}, {});
+ return;
+ }
+
+ // Decode the results
+ auto iter = bl.cbegin();
+ pg_nls_response_template<T> response;
+
+ try {
+ response.decode(iter);
+ if (!iter.end()) {
+ // extra_info isn't used anywhere. We do this solely to preserve
+ // backward compatibility
+ cb::list legacy_extra_info;
+ decode(legacy_extra_info, iter);
+ }
+ } catch (const bs::system_error& e) {
+ std::move(*ctx)(e.code(), {}, {});
+ return;
+ }
+
+ shared_lock rl(rwlock);
+ auto pool = osdmap->get_pg_pool(ctx->oloc.get_pool());
+ rl.unlock();
+ if (!pool) {
+ // pool is gone, drop any results which are now meaningless.
+ std::move(*ctx)(osdc_errc::pool_dne, {}, {});
+ return;
+ }
+
+ hobject_t next;
+ if ((response.handle <= ctx->end)) {
+ next = response.handle;
+ } else {
+ next = ctx->end;
+
+ // drop anything after 'end'
+ while (!response.entries.empty()) {
+ uint32_t hash = response.entries.back().locator.empty() ?
+ pool->hash_key(response.entries.back().oid,
+ response.entries.back().nspace) :
+ pool->hash_key(response.entries.back().locator,
+ response.entries.back().nspace);
+ hobject_t last(response.entries.back().oid,
+ response.entries.back().locator,
+ CEPH_NOSNAP,
+ hash,
+ ctx->oloc.get_pool(),
+ response.entries.back().nspace);
+ if (last < ctx->end)
+ break;
+ response.entries.pop_back();
+ }
+ }
+
+ if (response.entries.size() <= ctx->max) {
+ ctx->max -= response.entries.size();
+ std::move(response.entries.begin(), response.entries.end(),
+ std::back_inserter(ctx->ls));
+ } else {
+ auto i = response.entries.begin();
+ while (ctx->max > 0) {
+ ctx->ls.push_back(std::move(*i));
+ --(ctx->max);
+ ++i;
+ }
+ uint32_t hash =
+ i->locator.empty() ?
+ pool->hash_key(i->oid, i->nspace) :
+ pool->hash_key(i->locator, i->nspace);
+
+ next = hobject_t{i->oid, i->locator,
+ CEPH_NOSNAP,
+ hash,
+ ctx->oloc.get_pool(),
+ i->nspace};
+ }
+
+ if (next == ctx->end || ctx->max == 0) {
+ std::move(*ctx)(ec, std::move(ctx->ls), std::move(next));
+ } else {
+ _issue_enumerate(next, std::move(ctx));
+ }
+}
+
+template
+void Objecter::_enumerate_reply<librados::ListObjectImpl>(
+ cb::list&& bl,
+ bs::error_code ec,
+ std::unique_ptr<EnumerationContext<librados::ListObjectImpl>>&& ctx);
+
+template
+void Objecter::_enumerate_reply<neorados::Entry>(
+ cb::list&& bl,
+ bs::error_code ec,
+ std::unique_ptr<EnumerationContext<neorados::Entry>>&& ctx);
+
+namespace {
+ using namespace librados;
+
+ template <typename T>
+ void do_decode(std::vector<T>& items, std::vector<cb::list>& bls)
+ {
+ for (auto bl : bls) {
+ auto p = bl.cbegin();
+ T t;
+ decode(t, p);
+ items.push_back(t);
+ }
+ }
+
+ struct C_ObjectOperation_scrub_ls : public Context {
+ cb::list bl;
+ uint32_t* interval;
+ std::vector<inconsistent_obj_t> *objects = nullptr;
+ std::vector<inconsistent_snapset_t> *snapsets = nullptr;
+ int* rval;
+
+ C_ObjectOperation_scrub_ls(uint32_t* interval,
+ std::vector<inconsistent_obj_t>* objects,
+ int* rval)
+ : interval(interval), objects(objects), rval(rval) {}
+ C_ObjectOperation_scrub_ls(uint32_t* interval,
+ std::vector<inconsistent_snapset_t>* snapsets,
+ int* rval)
+ : interval(interval), snapsets(snapsets), rval(rval) {}
+ void finish(int r) override {
+ if (r < 0 && r != -EAGAIN) {
+ if (rval)
+ *rval = r;
+ return;
+ }
+
+ if (rval)
+ *rval = 0;
+
+ try {
+ decode();
+ } catch (cb::error&) {
+ if (rval)
+ *rval = -EIO;
+ }
+ }
+ private:
+ void decode() {
+ scrub_ls_result_t result;
+ auto p = bl.cbegin();
+ result.decode(p);
+ *interval = result.interval;
+ if (objects) {
+ do_decode(*objects, result.vals);
+ } else {
+ do_decode(*snapsets, result.vals);
+ }
+ }
+ };
+
+ template <typename T>
+ void do_scrub_ls(::ObjectOperation* op,
+ const scrub_ls_arg_t& arg,
+ std::vector<T> *items,
+ uint32_t* interval,
+ int* rval)
+ {
+ OSDOp& osd_op = op->add_op(CEPH_OSD_OP_SCRUBLS);
+ op->flags |= CEPH_OSD_FLAG_PGOP;
+ ceph_assert(interval);
+ arg.encode(osd_op.indata);
+ unsigned p = op->ops.size() - 1;
+ auto h = new C_ObjectOperation_scrub_ls{interval, items, rval};
+ op->set_handler(h);
+ op->out_bl[p] = &h->bl;
+ op->out_rval[p] = rval;
+ }
+}
+
+void ::ObjectOperation::scrub_ls(const librados::object_id_t& start_after,
+ uint64_t max_to_get,
+ std::vector<librados::inconsistent_obj_t>* objects,
+ uint32_t* interval,
+ int* rval)
+{
+ scrub_ls_arg_t arg = {*interval, 0, start_after, max_to_get};
+ do_scrub_ls(this, arg, objects, interval, rval);
+}
+
+void ::ObjectOperation::scrub_ls(const librados::object_id_t& start_after,
+ uint64_t max_to_get,
+ std::vector<librados::inconsistent_snapset_t> *snapsets,
+ uint32_t *interval,
+ int *rval)
+{
+ scrub_ls_arg_t arg = {*interval, 1, start_after, max_to_get};
+ do_scrub_ls(this, arg, snapsets, interval, rval);
+}