crimson/osd/osd_operations: Add RepRequest::with_pg() to the pipeline

Wait on a PG to advance to the request's map epoch.

Signed-off-by: Matan Breizman <mbreizma@redhat.com>

src/crimson/osd/osd_operations/client_request.h

@@ -41,6 +41,7 @@ class ClientRequest final : public PhasedOperationT<ClientRequest>,
 
 public:
   class PGPipeline : public CommonPGPipeline {
+    public:
     struct AwaitMap : OrderedExclusivePhaseT<AwaitMap> {
       static constexpr auto type_name = "ClientRequest::PGPipeline::await_map";
     } await_map;

src/crimson/osd/osd_operations/replicated_request.cc

@@ -58,11 +58,23 @@ seastar::future<> RepRequest::with_pg(
   ShardServices &shard_services, Ref<PG> pg)
 {
   logger().debug("{}: RepRequest::with_pg", *this);
-
   IRef ref = this;
   return interruptor::with_interruption([this, pg] {
-    return pg->handle_rep_op(req);
-  }, [ref](std::exception_ptr) { return seastar::now(); }, pg);
+    logger().debug("{}: pg present", *this);
+    return this->template enter_stage<interruptor>(pp(*pg).await_map
+    ).then_interruptible([this, pg] {
+      return this->template with_blocking_event<
+        PG_OSDMapGate::OSDMapBlocker::BlockingEvent
+      >([this, pg](auto &&trigger) {
+        return pg->osdmap_gate.wait_for_map(
+          std::move(trigger), req->min_epoch);
+      });
+    }).then_interruptible([this, pg] (auto) {
+      return pg->handle_rep_op(req);
+    });
+  }, [ref](std::exception_ptr) {
+    return seastar::now();
+  }, pg);
 }
 
 }

src/crimson/osd/osd_operations/replicated_request.h

@@ -46,6 +46,8 @@ public:
     ConnectionPipeline::AwaitActive::BlockingEvent,
     ConnectionPipeline::AwaitMap::BlockingEvent,
     ConnectionPipeline::GetPG::BlockingEvent,
+    ClientRequest::PGPipeline::AwaitMap::BlockingEvent,
+    PG_OSDMapGate::OSDMapBlocker::BlockingEvent,
     PGMap::PGCreationBlockingEvent,
     OSD_OSDMapGate::OSDMapBlocker::BlockingEvent
   > tracking_events;