Merge pull request ceph#60041 from xxhdx1985126/wip-68286

crimson/osd/pg_shard_manager: discard outdated operations when the corresponding pgs are already removed

Reviewed-by: Matan Breizman <[email protected]>

Author: Matan-B

src/crimson/osd/osd_operation.h

@@ -217,6 +217,9 @@ class TrackableOperationT : public OperationT<T> {
 
 public:
   static constexpr bool is_trackable = true;
+  virtual bool requires_pg() const {
+    return true;
+  }
 };
 
 template <class T>

src/crimson/osd/osd_operations/client_request.h

@@ -42,6 +42,10 @@ class ClientRequest final : public PhasedOperationT<ClientRequest>,
   unsigned instance_id = 0;
 
 public:
+  epoch_t get_epoch_sent_at() const {
+    return m->get_map_epoch();
+  }
+
   /**
    * instance_handle_t
    *

src/crimson/osd/osd_operations/logmissing_request.h

@@ -36,6 +36,9 @@ class LogMissingRequest final : public PhasedOperationT<LogMissingRequest> {
   }
   PipelineHandle &get_handle() { return handle; }
   epoch_t get_epoch() const { return req->get_min_epoch(); }
+  epoch_t get_epoch_sent_at() const {
+    return req->get_map_epoch();
+  }
 
   ConnectionPipeline &get_connection_pipeline();
 

src/crimson/osd/osd_operations/logmissing_request_reply.h

@@ -36,6 +36,9 @@ class LogMissingRequestReply final : public PhasedOperationT<LogMissingRequestRe
   }
   PipelineHandle &get_handle() { return handle; }
   epoch_t get_epoch() const { return req->get_min_epoch(); }
+  epoch_t get_epoch_sent_at() const {
+    return req->get_map_epoch();
+  }
 
   ConnectionPipeline &get_connection_pipeline();
 

src/crimson/osd/osd_operations/peering_event.h

@@ -44,6 +44,10 @@ class PeeringEvent : public PhasedOperationT<T> {
   float delay = 0;
   PGPeeringEvent evt;
 
+  epoch_t get_epoch_sent_at() const {
+    return evt.get_epoch_sent();
+  }
+
   const pg_shard_t get_from() const {
     return from;
   }
@@ -84,6 +88,10 @@ class PeeringEvent : public PhasedOperationT<T> {
     evt(std::forward<Args>(args)...)
   {}
 
+  bool requires_pg() const final {
+    return evt.requires_pg;
+  }
+
   void print(std::ostream &) const final;
   void dump_detail(ceph::Formatter* f) const final;
   seastar::future<> with_pg(

src/crimson/osd/osd_operations/pg_advance_map.h

@@ -50,6 +50,10 @@ class PGAdvanceMap : public PhasedOperationT<PGAdvanceMap> {
     PGPeeringPipeline::Process::BlockingEvent
   > tracking_events;
 
+  epoch_t get_epoch_sent_at() const {
+    return to;
+  }
+
 private:
   PGPeeringPipeline &peering_pp(PG &pg);
 };

src/crimson/osd/osd_operations/recovery_subrequest.h

@@ -39,6 +39,9 @@ class RecoverySubRequest final : public PhasedOperationT<RecoverySubRequest> {
   }
   PipelineHandle &get_handle() { return handle; }
   epoch_t get_epoch() const { return m->get_min_epoch(); }
+  epoch_t get_epoch_sent_at() const {
+    return m->get_map_epoch();
+  }
 
   ConnectionPipeline &get_connection_pipeline();
 

src/crimson/osd/osd_operations/replicated_request.h

@@ -36,6 +36,9 @@ class RepRequest final : public PhasedOperationT<RepRequest> {
   }
   PipelineHandle &get_handle() { return handle; }
   epoch_t get_epoch() const { return req->get_min_epoch(); }
+  epoch_t get_epoch_sent_at() const {
+    return req->get_map_epoch();
+  }
 
   ConnectionPipeline &get_connection_pipeline();
 

src/crimson/osd/osd_operations/scrub_events.h

@@ -27,11 +27,11 @@ class RemoteScrubEventBaseT : public PhasedOperationT<T> {
   crimson::net::ConnectionRef l_conn;
   crimson::net::ConnectionXcoreRef r_conn;
 
-  epoch_t epoch;
   spg_t pgid;
 
 protected:
   using interruptor = InterruptibleOperation::interruptor;
+  epoch_t epoch;
 
   template <typename U=void>
   using ifut = InterruptibleOperation::interruptible_future<U>;
@@ -40,7 +40,7 @@ class RemoteScrubEventBaseT : public PhasedOperationT<T> {
 public:
   RemoteScrubEventBaseT(
     crimson::net::ConnectionRef conn, epoch_t epoch, spg_t pgid)
-    : l_conn(std::move(conn)), epoch(epoch), pgid(pgid) {}
+    : l_conn(std::move(conn)), pgid(pgid), epoch(epoch) {}
 
   PGPeeringPipeline &get_peering_pipeline(PG &pg);
 
@@ -117,6 +117,10 @@ class ScrubRequested final : public RemoteScrubEventBaseT<ScrubRequested> {
     : RemoteScrubEventBaseT<ScrubRequested>(std::forward<Args>(base_args)...),
       deep(deep) {}
 
+  epoch_t get_epoch_sent_at() const {
+    return epoch;
+  }
+
   void print(std::ostream &out) const final {
     out << "(deep=" << deep << ")";
   }
@@ -141,6 +145,10 @@ class ScrubMessage final : public RemoteScrubEventBaseT<ScrubMessage> {
     ceph_assert(scrub::PGScrubber::is_scrub_message(*m));
   }
 
+  epoch_t get_epoch_sent_at() const {
+    return epoch;
+  }
+
   void print(std::ostream &out) const final {
     out << "(m=" << *m << ")";
   }

src/crimson/osd/pg_shard_manager.h

@@ -256,18 +256,40 @@ class PGShardManager {
     auto &opref = *op;
     return opref.template with_blocking_event<
       PGMap::PGCreationBlockingEvent
-    >([&target_shard_services, &opref](auto &&trigger) {
-      return target_shard_services.wait_for_pg(
-        std::move(trigger), opref.get_pgid());
-    }).safe_then([&logger, &target_shard_services, &opref](Ref<PG> pgref) {
-      logger.debug("{}: have_pg", opref);
-      return opref.with_pg(target_shard_services, pgref);
-    }).handle_error(
-      crimson::ct_error::ecanceled::handle([&logger, &opref](auto) {
-        logger.debug("{}: pg creation canceled, dropping", opref);
-        return seastar::now();
-      })
-    ).then([op=std::move(op)] {});
+    >([&target_shard_services, &opref, &logger](auto &&trigger) mutable {
+      auto pg = target_shard_services.get_pg(opref.get_pgid());
+      auto fut = ShardServices::wait_for_pg_ertr::make_ready_future<Ref<PG>>(pg);
+      if (!pg) {
+	if (opref.requires_pg()) {
+	  auto osdmap = target_shard_services.get_map();
+	  if (!osdmap->is_up_acting_osd_shard(
+		opref.get_pgid(), target_shard_services.local_state.whoami)) {
+	    logger.debug(
+	      "pg {} for {} is no longer here, discarding",
+	      opref.get_pgid(), opref);
+	    opref.get_handle().exit();
+	    auto _fut = seastar::now();
+	    if (osdmap->get_epoch() > opref.get_epoch_sent_at()) {
+	      _fut = target_shard_services.send_incremental_map(
+		std::ref(opref.get_foreign_connection()),
+		opref.get_epoch_sent_at() + 1);
+	    }
+	    return _fut;
+	  }
+	}
+	fut = target_shard_services.wait_for_pg(
+	  std::move(trigger), opref.get_pgid());
+      }
+      return fut.safe_then([&logger, &target_shard_services, &opref](Ref<PG> pgref) {
+	logger.debug("{}: have_pg", opref);
+	return opref.with_pg(target_shard_services, pgref);
+      }).handle_error(
+	crimson::ct_error::ecanceled::handle([&logger, &opref](auto) {
+	  logger.debug("{}: pg creation canceled, dropping", opref);
+	  return seastar::now();
+	})
+      );
+    }).then([op=std::move(op)] {});
   }
 
   seastar::future<> load_pgs(crimson::os::FuturizedStore& store);