crimson: refactor RepRequest to use start_pg_operation

Signed-off-by: Samuel Just <sjust@redhat.com>

diff --git a/src/crimson/osd/osd.cc b/src/crimson/osd/osd.cc
index 2bd2fe8408cb3556920fe0f9f4a0afe4802f03e2..34c894dd64fda82822143e2b2f3b85e3a0d55ee2 100644 (file)
--- a/src/crimson/osd/osd.cc
+++ b/src/crimson/osd/osd.cc
@@ -1235,8 +1235,7 @@ seastar::future<> OSD::handle_rep_op(crimson::net::ConnectionRef conn,
                                     Ref<MOSDRepOp> m)
 {
   m->finish_decode();
-  (void) shard_services.start_operation<RepRequest>(
-    *this,
+  std::ignore = start_pg_operation<RepRequest>(
     std::move(conn),
     std::move(m));
   return seastar::now();

diff --git a/src/crimson/osd/osd_operations/replicated_request.cc b/src/crimson/osd/osd_operations/replicated_request.cc
index 3a8c3f9a117b9fe2a0f966c2a9b826a50be8a77f..d93e4035ed0f5d0ce572c966e746cf2068bbf614 100644 (file)
--- a/src/crimson/osd/osd_operations/replicated_request.cc
+++ b/src/crimson/osd/osd_operations/replicated_request.cc
@@ -4,7 +4,6 @@
 #include "replicated_request.h"
 
 #include "common/Formatter.h"
-#include "messages/MOSDRepOp.h"
 
 #include "crimson/osd/osd.h"
 #include "crimson/osd/osd_connection_priv.h"
@@ -19,11 +18,9 @@ namespace {
 
 namespace crimson::osd {
 
-RepRequest::RepRequest(OSD &osd,
-                      crimson::net::ConnectionRef&& conn,
+RepRequest::RepRequest(crimson::net::ConnectionRef&& conn,
                       Ref<MOSDRepOp> &&req)
-  : osd{osd},
-    conn{std::move(conn)},
+  : conn{std::move(conn)},
     req{req}
 {}
 
@@ -47,7 +44,7 @@ void RepRequest::dump_detail(Formatter *f) const
   f->close_section();
 }
 
-ConnectionPipeline &RepRequest::cp()
+ConnectionPipeline &RepRequest::get_connection_pipeline()
 {
   return get_osd_priv(conn.get()).replicated_request_conn_pipeline;
 }
@@ -57,28 +54,15 @@ RepRequest::PGPipeline &RepRequest::pp(PG &pg)
   return pg.replicated_request_pg_pipeline;
 }
 
-seastar::future<> RepRequest::start()
+seastar::future<> RepRequest::with_pg(
+  ShardServices &shard_services, Ref<PG> pg)
 {
-  logger().debug("{} start", *this);
-  IRef ref = this;
+  logger().debug("{}: RepRequest::with_pg", *this);
 
-  return enter_stage<>(cp().await_map).then([this] {
-    return with_blocking_event<OSD_OSDMapGate::OSDMapBlocker::BlockingEvent>(
-      [this] (auto&& trigger) {
-        return osd.osdmap_gate.wait_for_map(std::move(trigger),
-                                           req->get_min_epoch());
-      });
-  }).then([this](epoch_t epoch) {
-    return enter_stage<>(cp().get_pg);
-  }).then([this] {
-    return with_blocking_event<PGMap::PGCreationBlockingEvent>(
-      [this] (auto&& trigger) {
-        return osd.wait_for_pg(std::move(trigger), req->get_spg());
-      });
-  }).then([this, ref=std::move(ref)](Ref<PG> pg) {
-    return interruptor::with_interruption([this, ref, pg] {
-       return pg->handle_rep_op(std::move(req));
-      }, [](std::exception_ptr) { return seastar::now(); }, pg);
-  });
+  IRef ref = this;
+  return interruptor::with_interruption([this, pg] {
+    return pg->handle_rep_op(std::move(req));
+  }, [ref](std::exception_ptr) { return seastar::now(); }, pg);
 }
+
 }

diff --git a/src/crimson/osd/osd_operations/replicated_request.h b/src/crimson/osd/osd_operations/replicated_request.h
index 39c6d6a2473753eeb8f2f80dbd356c38a1f7a622..f84b107721ac5e5fe8c1b5fc221fadf35e35a37a 100644 (file)
--- a/src/crimson/osd/osd_operations/replicated_request.h
+++ b/src/crimson/osd/osd_operations/replicated_request.h
@@ -8,8 +8,7 @@
 #include "crimson/osd/osd_operation.h"
 #include "crimson/osd/pg_map.h"
 #include "crimson/common/type_helpers.h"
-
-class MOSDRepOp;
+#include "messages/MOSDRepOp.h"
 
 namespace ceph {
   class Formatter;
@@ -17,6 +16,8 @@ namespace ceph {
 
 namespace crimson::osd {
 
+class ShardServices;
+
 class OSD;
 class PG;
 
@@ -32,11 +33,21 @@ public:
     friend RepRequest;
   };
   static constexpr OperationTypeCode type = OperationTypeCode::replicated_request;
-  RepRequest(OSD&, crimson::net::ConnectionRef&&, Ref<MOSDRepOp>&&);
+  RepRequest(crimson::net::ConnectionRef&&, Ref<MOSDRepOp>&&);
 
   void print(std::ostream &) const final;
   void dump_detail(ceph::Formatter* f) const final;
-  seastar::future<> start();
+
+  static constexpr bool can_create() { return false; }
+  spg_t get_pgid() const {
+    return req->get_spg();
+  }
+  ConnectionPipeline &get_connection_pipeline();
+  PipelineHandle &get_handle() { return handle; }
+  epoch_t get_epoch() const { return req->get_min_epoch(); }
+
+  seastar::future<> with_pg(
+    ShardServices &shard_services, Ref<PG> pg);
 
   std::tuple<
     ConnectionPipeline::AwaitActive::BlockingEvent,
@@ -47,10 +58,8 @@ public:
   > tracking_events;
 
 private:
-  ConnectionPipeline &cp();
   PGPipeline &pp(PG &pg);
 
-  OSD &osd;
   crimson::net::ConnectionRef conn;
   Ref<MOSDRepOp> req;
 };