From: Radosław Zarzyński <rzarzyns@redhat.com>
Date: Wed, 12 Jul 2023 19:35:37 +0000 (+0200)
Subject: crimson/osd: bring ECRepRequest
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=2697e32050539bf9e736949e335e163035a144e2;p=ceph-ci.git

crimson/osd: bring ECRepRequest

Signed-off-by: Radosław Zarzyński <rzarzyns@redhat.com>
---

diff --git a/src/crimson/osd/CMakeLists.txt b/src/crimson/osd/CMakeLists.txt
index 7cecc53c36e..00d75da778a 100644
--- a/src/crimson/osd/CMakeLists.txt
+++ b/src/crimson/osd/CMakeLists.txt
@@ -18,6 +18,7 @@ add_executable(crimson-osd
   object_metadata_helper.cc
   ops_executer.cc
   osd_operation.cc
+  osd_operations/ecrep_request.cc
   osd_operations/client_request.cc
   osd_operations/internal_client_request.cc
   osd_operations/peering_event.cc
diff --git a/src/crimson/osd/osd.cc b/src/crimson/osd/osd.cc
index b914b3fc42b..7b957028330 100644
--- a/src/crimson/osd/osd.cc
+++ b/src/crimson/osd/osd.cc
@@ -54,6 +54,7 @@
 #include "crimson/osd/pg_backend.h"
 #include "crimson/osd/pg_meta.h"
 #include "crimson/osd/osd_operations/client_request.h"
+#include "crimson/osd/osd_operations/ecrep_request.h"
 #include "crimson/osd/osd_operations/peering_event.h"
 #include "crimson/osd/osd_operations/pgpct_request.h"
 #include "crimson/osd/osd_operations/pg_advance_map.h"
@@ -1006,6 +1007,20 @@ OSD::do_ms_dispatch(
   case MSG_OSD_PG_PCT:
     return handle_pg_pct(conn, boost::static_pointer_cast<
       MOSDPGPCT>(m));
+  case MSG_OSD_EC_WRITE:
+    [[fallthrough]];
+  case MSG_OSD_EC_WRITE_REPLY:
+    [[fallthrough]];
+  case MSG_OSD_EC_READ:
+    [[fallthrough]];
+  case MSG_OSD_EC_READ_REPLY:
+    return handle_some_ec_messages(conn, m);
+#if 0
+  case MSG_OSD_PG_PUSH:
+    [[fallthrough]];
+  case MSG_OSD_PG_PUSH_REPLY:
+    return handle_ec_messages(conn, m);
+#endif
   default:
     return std::nullopt;
   }
@@ -1552,6 +1567,19 @@ bool OSD::should_restart() const
   }
 }
 
+template <class MessageRefT>
+seastar::future<>
+OSD::handle_some_ec_messages(crimson::net::ConnectionRef conn, MessageRefT&& m)
+{
+  m->decode_payload();
+  //m->set_features(conn->get_features());
+  (void) pg_shard_manager.start_pg_operation<ECRepRequest>(
+    std::move(conn),
+    std::forward<MessageRefT>(m));
+  return seastar::now();
+}
+
+
 seastar::future<> OSD::restart()
 {
   beacon_timer.cancel();
diff --git a/src/crimson/osd/osd.h b/src/crimson/osd/osd.h
index 3ec1a3636c9..e26af63b170 100644
--- a/src/crimson/osd/osd.h
+++ b/src/crimson/osd/osd.h
@@ -238,6 +238,10 @@ private:
   seastar::future<> handle_pg_pct(
     crimson::net::ConnectionRef conn,
     Ref<MOSDPGPCT> m);
+  template <class MessageRefT>
+  seastar::future<> handle_some_ec_messages(
+    crimson::net::ConnectionRef conn,
+    MessageRefT&& m);
 
   std::vector<DaemonHealthMetric> get_health_metrics();
 
diff --git a/src/crimson/osd/osd_operation.h b/src/crimson/osd/osd_operation.h
index a5b0d7815f9..8ceaddd2c51 100644
--- a/src/crimson/osd/osd_operation.h
+++ b/src/crimson/osd/osd_operation.h
@@ -106,6 +106,7 @@ enum class OperationTypeCode {
   scrub_reserve_range,
   scrub_scan,
   pgpct_request,
+  ecrep_request,
   last_op
 };
 
@@ -131,6 +132,7 @@ static constexpr const char* const OP_NAMES[] = {
   "scrub_reserve_range",
   "scrub_scan",
   "pgpct_request",
+  "ecrep_request",
 };
 
 // prevent the addition of OperationTypeCode-s with no matching OP_NAMES entry:
diff --git a/src/crimson/osd/osd_operation_external_tracking.h b/src/crimson/osd/osd_operation_external_tracking.h
index 622254e26fc..b91edb78288 100644
--- a/src/crimson/osd/osd_operation_external_tracking.h
+++ b/src/crimson/osd/osd_operation_external_tracking.h
@@ -7,6 +7,7 @@
 #include "crimson/osd/osdmap_gate.h"
 #include "crimson/osd/osd_operations/background_recovery.h"
 #include "crimson/osd/osd_operations/client_request.h"
+#include "crimson/osd/osd_operations/ecrep_request.h"
 #include "crimson/osd/osd_operations/peering_event.h"
 #include "crimson/osd/osd_operations/pgpct_request.h"
 #include "crimson/osd/osd_operations/pg_advance_map.h"
@@ -293,6 +294,13 @@ struct HistoricBackend
 
 namespace crimson {
 
+template <>
+struct EventBackendRegistry<osd::ECRepRequest> {
+  static std::tuple<> get_backends() {
+    return {/* no extenral backends */};
+  }
+};
+
 template <>
 struct EventBackendRegistry<osd::ClientRequest> {
   static std::tuple<osd::LttngBackend, osd::HistoricBackend> get_backends() {
diff --git a/src/crimson/osd/osd_operations/ecrep_request.cc b/src/crimson/osd/osd_operations/ecrep_request.cc
new file mode 100644
index 00000000000..4e9ed4808f8
--- /dev/null
+++ b/src/crimson/osd/osd_operations/ecrep_request.cc
@@ -0,0 +1,74 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "ecrep_request.h"
+
+#include "common/Formatter.h"
+
+#include "crimson/osd/osd.h"
+#include "crimson/osd/osd_connection_priv.h"
+#include "crimson/osd/osd_operation_external_tracking.h"
+#include "crimson/osd/pg.h"
+
+namespace {
+  seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_osd);
+  }
+}
+
+namespace crimson::osd {
+
+void ECRepRequest::print(std::ostream& os) const
+{
+  os << "ECRepRequest("
+     << ")";
+}
+
+void ECRepRequest::dump_detail(Formatter *f) const
+{
+#if 0
+  f->open_object_section("ECRepRequest");
+  f->dump_stream("req_tid") << req->get_tid();
+  f->dump_stream("pgid") << get_pgid();
+  f->dump_unsigned("map_epoch", req->get_map_epoch());
+  f->dump_unsigned("min_epoch", req->get_min_epoch());
+  f->close_section();
+#endif
+}
+
+ConnectionPipeline &ECRepRequest::get_connection_pipeline()
+{
+  return get_osd_priv(&get_local_connection()
+         ).replicated_request_conn_pipeline;
+}
+
+PerShardPipeline &ECRepRequest::get_pershard_pipeline(
+  ShardServices &shard_services)
+{
+  return shard_services.get_replicated_request_pipeline();
+}
+
+// from https://en.cppreference.com/w/cpp/utility/variant/visit
+// helper type for the visitor #4
+template<class... Ts>
+struct overloaded : Ts... { using Ts::operator()...; };
+// explicit deduction guide (not needed as of C++20)
+template<class... Ts>
+overloaded(Ts...) -> overloaded<Ts...>;
+
+seastar::future<> ECRepRequest::with_pg(
+  ShardServices &shard_services, Ref<PG> pg)
+{
+  logger().debug("{}: ECRepRequest::with_pg", *this);
+
+  IRef ref = this;
+  return interruptor::with_interruption([this, pg] {
+    return interruptor::now();
+  }, [ref, this](std::exception_ptr) {
+    logger().debug("{}: ECRepRequest::exception handling", *this);
+    return seastar::now();
+  }, pg, pg->get_osdmap_epoch());
+}
+
+}
+
diff --git a/src/crimson/osd/osd_operations/ecrep_request.h b/src/crimson/osd/osd_operations/ecrep_request.h
new file mode 100644
index 00000000000..7192f22b35b
--- /dev/null
+++ b/src/crimson/osd/osd_operations/ecrep_request.h
@@ -0,0 +1,121 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <variant>
+
+#include "crimson/net/Connection.h"
+#include "crimson/osd/osdmap_gate.h"
+#include "crimson/osd/osd_operation.h"
+#include "crimson/osd/osd_operations/client_request.h"
+#include "crimson/osd/pg_map.h"
+#include "crimson/common/type_helpers.h"
+
+
+namespace ceph {
+  class Formatter;
+}
+
+namespace crimson::osd {
+
+class ShardServices;
+
+class OSD;
+class PG;
+
+class ECRepRequest final : public PhasedOperationT<ECRepRequest> {
+public:
+  static constexpr OperationTypeCode type = OperationTypeCode::ecrep_request;
+
+  template <class MessageRefT>
+  ECRepRequest(crimson::net::ConnectionRef&& conn,
+               MessageRefT &&req)
+    : l_conn{std::move(conn)},
+      req{std::forward<MessageRefT>(req)}
+  {}
+
+  void print(std::ostream &) const final;
+  void dump_detail(ceph::Formatter* f) const final;
+
+  static constexpr bool can_create() { return false; }
+  spg_t get_pgid() const {
+    return std::visit([] (const auto& concrete_req) {
+      return concrete_req->get_spg();
+    }, req);
+  }
+  epoch_t get_epoch() const {
+    return std::visit([] (const auto& concrete_req) {
+      return concrete_req->get_min_epoch();
+    }, req);
+  }
+
+  epoch_t get_epoch_sent_at() const {
+    return std::visit([] (const auto& concrete_req) {
+      return concrete_req->get_map_epoch();
+    }, req);
+  }
+
+  PipelineHandle &get_handle() { return handle; }
+
+  ConnectionPipeline &get_connection_pipeline();
+
+  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));
+  }
+
+  seastar::future<> with_pg(
+    ShardServices &shard_services, Ref<PG> pg);
+
+  std::tuple<
+    StartEvent,
+    ConnectionPipeline::AwaitActive::BlockingEvent,
+    ConnectionPipeline::AwaitMap::BlockingEvent,
+    ConnectionPipeline::GetPGMapping::BlockingEvent,
+    PerShardPipeline::CreateOrWaitPG::BlockingEvent,
+    PGMap::PGCreationBlockingEvent,
+    OSD_OSDMapGate::OSDMapBlocker::BlockingEvent
+  > tracking_events;
+
+private:
+  crimson::net::ConnectionRef l_conn;
+  crimson::net::ConnectionXcoreRef r_conn;
+  // must be after `conn` to ensure the ConnectionPipeline's is alive
+  PipelineHandle handle;
+  std::variant<
+    Ref<MOSDECSubOpWrite>,
+    Ref<MOSDECSubOpWriteReply>,
+    Ref<MOSDECSubOpRead>,
+    Ref<MOSDECSubOpReadReply>
+  > req;
+};
+
+}
+
+#if FMT_VERSION >= 90000
+template <> struct fmt::formatter<crimson::osd::ECRepRequest> : fmt::ostream_formatter {};
+#endif
diff --git a/src/crimson/osd/pg.h b/src/crimson/osd/pg.h
index c840648975d..a298618eb63 100644
--- a/src/crimson/osd/pg.h
+++ b/src/crimson/osd/pg.h
@@ -978,6 +978,7 @@ private:
   PglogBasedRecovery* pglog_based_recovery_op = nullptr;
 
   friend std::ostream& operator<<(std::ostream&, const PG& pg);
+  friend class ECRepRequest;
   friend class ClientRequest;
   friend struct CommonClientRequest;
   friend class PGAdvanceMap;