Merge pull request ceph#63061 from Matan-B/wip-71137-tentacle

tentacle: crimson: osd_operation cleanups and fix for MOSDRepOpReply ordering

Reviewed-by: Aishwarya Mathuria <[email protected]>

Author: Matan-B

src/crimson/osd/CMakeLists.txt

@@ -19,11 +19,11 @@ add_executable(crimson-osd
   ops_executer.cc
   osd_operation.cc
   osd_operations/client_request.cc
-  osd_operations/client_request_common.cc
   osd_operations/internal_client_request.cc
   osd_operations/peering_event.cc
   osd_operations/pg_advance_map.cc
   osd_operations/replicated_request.cc
+  osd_operations/replicated_request_reply.cc
   osd_operations/logmissing_request.cc
   osd_operations/logmissing_request_reply.cc
   osd_operations/background_recovery.cc

src/crimson/osd/osd.cc

@@ -58,6 +58,7 @@
 #include "crimson/osd/osd_operations/pg_advance_map.h"
 #include "crimson/osd/osd_operations/recovery_subrequest.h"
 #include "crimson/osd/osd_operations/replicated_request.h"
+#include "crimson/osd/osd_operations/replicated_request_reply.h"
 #include "crimson/osd/osd_operations/scrub_events.h"
 #include "crimson/osd/osd_operation_external_tracking.h"
 #include "crimson/crush/CrushLocation.h"
@@ -1401,56 +1402,45 @@ seastar::future<> OSD::handle_update_log_missing(
   crimson::net::ConnectionRef conn,
   Ref<MOSDPGUpdateLogMissing> m)
 {
-  return pg_shard_manager.start_pg_operation<LogMissingRequest>(
+  return pg_shard_manager.start_pg_operation_active<LogMissingRequest>(
     std::move(conn),
-    std::move(m)).second;
+    std::move(m));
 }
 
 seastar::future<> OSD::handle_update_log_missing_reply(
   crimson::net::ConnectionRef conn,
   Ref<MOSDPGUpdateLogMissingReply> m)
 {
-  return pg_shard_manager.start_pg_operation<LogMissingRequestReply>(
+  return pg_shard_manager.start_pg_operation_active<LogMissingRequestReply>(
     std::move(conn),
-    std::move(m)).second;
+    std::move(m));
 }
 
 seastar::future<> OSD::handle_pg_pct(
   crimson::net::ConnectionRef conn,
   Ref<MOSDPGPCT> m)
 {
-  return pg_shard_manager.start_pg_operation<PGPCTRequest>(
+  return pg_shard_manager.start_pg_operation_active<PGPCTRequest>(
     std::move(conn),
-    std::move(m)).second;
+    std::move(m));
 }
 
 seastar::future<> OSD::handle_rep_op(
   crimson::net::ConnectionRef conn,
   Ref<MOSDRepOp> m)
 {
-  m->finish_decode();
-  return pg_shard_manager.start_pg_operation<RepRequest>(
+  return pg_shard_manager.start_pg_operation_active<RepRequest>(
     std::move(conn),
-    std::move(m)).second;
+    std::move(m));
 }
 
 seastar::future<> OSD::handle_rep_op_reply(
   crimson::net::ConnectionRef conn,
   Ref<MOSDRepOpReply> m)
 {
-  LOG_PREFIX(OSD::handle_rep_op_reply);
-  spg_t pgid = m->get_spg();
-  return pg_shard_manager.with_pg(
-    pgid,
-    [FNAME, m=std::move(m)](auto &&pg) {
-      if (pg) {
-	m->finish_decode();
-	pg->handle_rep_op_reply(*m);
-      } else {
-	WARN("stale reply: {}", *m);
-      }
-      return seastar::now();
-    });
+  return pg_shard_manager.start_pg_operation_active<ReplicatedRequestReply>(
+    std::move(conn),
+    std::move(m));
 }
 
 seastar::future<> OSD::handle_scrub_command(
@@ -1474,9 +1464,9 @@ seastar::future<> OSD::handle_scrub_message(
   crimson::net::ConnectionRef conn,
   Ref<MOSDFastDispatchOp> m)
 {
-  return pg_shard_manager.start_pg_operation<
+  return pg_shard_manager.start_pg_operation_active<
     crimson::osd::ScrubMessage
-    >(m, conn, m->get_min_epoch(), m->get_spg()).second;
+    >(m, conn, m->get_min_epoch(), m->get_spg());
 }
 
 seastar::future<> OSD::handle_mark_me_down(
@@ -1494,8 +1484,8 @@ seastar::future<> OSD::handle_recovery_subreq(
   crimson::net::ConnectionRef conn,
   Ref<MOSDFastDispatchOp> m)
 {
-  return pg_shard_manager.start_pg_operation<RecoverySubRequest>(
-    conn, std::move(m)).second;
+  return pg_shard_manager.start_pg_operation_active<RecoverySubRequest>(
+    conn, std::move(m));
 }
 
 vector<DaemonHealthMetric> OSD::get_health_metrics()

src/crimson/osd/osd_operation.h

@@ -4,6 +4,7 @@
 #pragma once
 
 #include "crimson/common/operation.h"
+#include "crimson/net/Connection.h"
 #include "crimson/osd/pg_interval_interrupt_condition.h"
 #include "crimson/osd/scheduler/scheduler.h"
 #include "osd/osd_types.h"
@@ -89,6 +90,7 @@ enum class OperationTypeCode {
   pg_advance_map,
   pg_creation,
   replicated_request,
+  replicated_request_reply,
   background_recovery,
   background_recovery_sub,
   internal_client_request,
@@ -113,6 +115,7 @@ static constexpr const char* const OP_NAMES[] = {
   "pg_advance_map",
   "pg_creation",
   "replicated_request",
+  "replicated_request_reply",
   "background_recovery",
   "background_recovery_sub",
   "internal_client_request",
@@ -165,6 +168,61 @@ struct OperationT : InterruptibleOperation {
   virtual void dump_detail(ceph::Formatter *f) const = 0;
 };
 
+class RemoteOperation {
+  crimson::net::ConnectionRef l_conn;
+  crimson::net::ConnectionXcoreRef r_conn;
+
+public:
+  RemoteOperation(crimson::net::ConnectionRef &&conn)
+    : l_conn(std::move(conn)) {}
+
+  crimson::net::Connection &get_local_connection() {
+    assert(l_conn);
+    assert(!r_conn);
+    return *l_conn;
+  };
+
+  crimson::net::Connection &get_foreign_connection() {
+    assert(r_conn);
+    assert(!l_conn);
+    return *r_conn;
+  };
+
+  crimson::net::ConnectionFFRef prepare_remote_submission() {
+    assert(l_conn);
+    assert(!r_conn);
+    auto ret = seastar::make_foreign(std::move(l_conn));
+    l_conn.reset();
+    return ret;
+  }
+
+  void finish_remote_submission(crimson::net::ConnectionFFRef conn) {
+    assert(conn);
+    assert(!l_conn);
+    assert(!r_conn);
+    r_conn = make_local_shared_foreign(std::move(conn));
+  }
+
+  crimson::net::Connection &get_connection() const {
+    if (l_conn) {
+      return *l_conn;
+    } else {
+      assert(r_conn);
+      return *r_conn;
+    }
+  }
+
+  /**
+   * get_remote_connection
+   *
+   * Return a reference to the remote connection to allow caller to
+   * perform a copy only as needed.
+   */
+  crimson::net::ConnectionXcoreRef &get_remote_connection() {
+    return r_conn;
+  }
+};
+
 template <class T>
 class TrackableOperationT : public OperationT<T> {
   T* that() {

src/crimson/osd/osd_operations/client_request.cc

@@ -60,8 +60,8 @@ void ClientRequest::complete_request(PG &pg)
 ClientRequest::ClientRequest(
   ShardServices &_shard_services, crimson::net::ConnectionRef conn,
   Ref<MOSDOp> &&m)
-  : shard_services(&_shard_services),
-    l_conn(std::move(conn)),
+  : RemoteOperation(std::move(conn)),
+    shard_services(&_shard_services),
     m(std::move(m)),
     begin_time(std::chrono::steady_clock::now()),
     instance_handle(new instance_handle_t)
@@ -321,7 +321,7 @@ ClientRequest::recover_missing_snaps(
   co_await std::move(resolve_oids);
 
   for (auto &oid : ret) {
-    auto unfound = co_await do_recover_missing(pg, oid, m->get_reqid());
+    auto unfound = co_await pg->do_recover_missing(oid, m->get_reqid());
     if (unfound) {
       DEBUGDPP("{} unfound, hang it for now", *pg, oid);
       co_await interruptor::make_interruptible(
@@ -347,8 +347,8 @@ ClientRequest::process_op(
       "Skipping recover_missings on non primary pg for soid {}",
       *pg, m->get_hobj());
   } else {
-    auto unfound = co_await do_recover_missing(
-      pg, m->get_hobj().get_head(), m->get_reqid());
+    auto unfound = co_await pg->do_recover_missing(
+      m->get_hobj().get_head(), m->get_reqid());
     if (unfound) {
       DEBUGDPP("{} unfound, hang it for now", *pg, m->get_hobj().get_head());
       co_await interruptor::make_interruptible(
@@ -486,7 +486,7 @@ ClientRequest::do_process(
     co_return;
   }
 
-  OpsExecuter ox(pg, obc, op_info, *m, r_conn, snapc);
+  OpsExecuter ox(pg, obc, op_info, *m, get_remote_connection(), snapc);
   auto ret = co_await pg->run_executer(
     ox, obc, op_info, m->ops
   ).si_then([]() -> std::optional<std::error_code> {

src/crimson/osd/osd_operations/client_request.h

@@ -15,7 +15,6 @@
 #include "crimson/osd/object_context_loader.h"
 #include "crimson/osd/osdmap_gate.h"
 #include "crimson/osd/osd_operation.h"
-#include "crimson/osd/osd_operations/client_request_common.h"
 #include "crimson/osd/pg_activation_blocker.h"
 #include "crimson/osd/pg_map.h"
 #include "crimson/osd/scrub/pg_scrubber.h"
@@ -28,14 +27,13 @@ class PG;
 class OSD;
 class ShardServices;
 
-class ClientRequest final : public PhasedOperationT<ClientRequest>,
-                            private CommonClientRequest {
+class ClientRequest final
+  : public PhasedOperationT<ClientRequest>,
+    public RemoteOperation
+{
   // Initially set to primary core, updated to pg core after with_pg()
   ShardServices *shard_services = nullptr;
 
-  crimson::net::ConnectionRef l_conn;
-  crimson::net::ConnectionXcoreRef r_conn;
-
   // must be after conn due to ConnectionPipeline's life-time
   Ref<MOSDOp> m;
   OpInfo op_info;
@@ -56,9 +54,6 @@ class ClientRequest final : public PhasedOperationT<ClientRequest>,
     static_assert(std::is_same_v<T, MOSDOp>);
     return m.get();
   }
-  const crimson::net::ConnectionRef &get_connection() const {
-    return l_conn;
-  }
 
   /**
    * instance_handle_t
@@ -233,33 +228,6 @@ class ClientRequest final : public PhasedOperationT<ClientRequest>,
 
   PerShardPipeline &get_pershard_pipeline(ShardServices &);
 
-  crimson::net::Connection &get_local_connection() {
-    assert(l_conn);
-    assert(!r_conn);
-    return *l_conn;
-  };
-
-  crimson::net::Connection &get_foreign_connection() {
-    assert(r_conn);
-    assert(!l_conn);
-    return *r_conn;
-  };
-
-  crimson::net::ConnectionFFRef prepare_remote_submission() {
-    assert(l_conn);
-    assert(!r_conn);
-    auto ret = seastar::make_foreign(std::move(l_conn));
-    l_conn.reset();
-    return ret;
-  }
-
-  void finish_remote_submission(crimson::net::ConnectionFFRef conn) {
-    assert(conn);
-    assert(!l_conn);
-    assert(!r_conn);
-    r_conn = make_local_shared_foreign(std::move(conn));
-  }
-
   interruptible_future<> with_pg_process_interruptible(
     Ref<PG> pgref, const unsigned instance_id, instance_handle_t &ihref);