crimson/osd/pg: Let PGListener use start_peering_event_operation

PG::start_peering_event_operation is a template function while
PGRecovery::pg is of PGRecoveryListener* type. We can't expose a template
function through the PGRecoveryListener interface since it must be
also virtual.
Instead, introduce start_peering_event_operation_listener which will act
as a wrapper to PG::start_peering_event_operation for PGRecovery to use
freely.

Signed-off-by: Matan Breizman <[email protected]>
(cherry picked from commit c94e604)

Author: Matan-B

src/crimson/osd/pg_recovery.cc

@@ -623,47 +623,36 @@ void PGRecovery::on_pg_clean()
   backfill_state.reset();
 }
 
+// PGListener wrapper to PG::start_peering_event_operation
+template <typename T>
+void PGRecovery::start_peering_event_operation_listener(T &&evt, float delay) {
+  (void) pg->schedule_event_after(PGPeeringEventRef(
+    std::make_shared<PGPeeringEvent>(
+      pg->get_osdmap_epoch(),
+      pg->get_osdmap_epoch(),
+      std::forward<T>(evt))), delay);
+}
+
 void PGRecovery::backfilled()
 {
   LOG_PREFIX(PGRecovery::backfilled);
   DEBUGDPP("", pg->get_pgid());
-  using LocalPeeringEvent = crimson::osd::LocalPeeringEvent;
-  std::ignore = pg->get_shard_services().start_operation<LocalPeeringEvent>(
-    static_cast<crimson::osd::PG*>(pg),
-    pg->get_pg_whoami(),
-    pg->get_pgid(),
-    pg->get_osdmap_epoch(),
-    pg->get_osdmap_epoch(),
-    PeeringState::Backfilled{});
+  start_peering_event_operation_listener(PeeringState::Backfilled());
 }
 
 void PGRecovery::request_backfill()
 {
   LOG_PREFIX(PGRecovery::request_backfill);
   DEBUGDPP("", pg->get_pgid());
-  using LocalPeeringEvent = crimson::osd::LocalPeeringEvent;
-  std::ignore = pg->get_shard_services().start_operation<LocalPeeringEvent>(
-    static_cast<crimson::osd::PG*>(pg),
-    pg->get_pg_whoami(),
-    pg->get_pgid(),
-    pg->get_osdmap_epoch(),
-    pg->get_osdmap_epoch(),
-    PeeringState::RequestBackfill{});
+  start_peering_event_operation_listener(PeeringState::RequestBackfill());
 }
 
 
 void PGRecovery::all_replicas_recovered()
 {
   LOG_PREFIX(PGRecovery::all_replicas_recovered);
   DEBUGDPP("", pg->get_pgid());
-  using LocalPeeringEvent = crimson::osd::LocalPeeringEvent;
-  std::ignore = pg->get_shard_services().start_operation<LocalPeeringEvent>(
-    static_cast<crimson::osd::PG*>(pg),
-    pg->get_pg_whoami(),
-    pg->get_pgid(),
-    pg->get_osdmap_epoch(),
-    pg->get_osdmap_epoch(),
-    PeeringState::AllReplicasRecovered{});
+  start_peering_event_operation_listener(PeeringState::AllReplicasRecovered());
 }
 
 void PGRecovery::backfill_suspended()

src/crimson/osd/pg_recovery.h

@@ -138,7 +138,8 @@ class PGRecovery : public crimson::osd::BackfillState::BackfillListener {
     const hobject_t& new_last_backfill) final;
   bool budget_available() const final;
 
-  // TODO: move to start_peering_event_operation
+  template <typename T>
+  void start_peering_event_operation_listener(T &&evt, float delay = 0);
   void backfilled() final;
   void request_backfill();
   void all_replicas_recovered();

src/crimson/osd/pg_recovery_listener.h

@@ -43,4 +43,5 @@ class PGRecoveryListener {
   virtual void set_pglog_based_recovery_op(
     crimson::osd::PglogBasedRecovery *op) = 0;
   virtual void reset_pglog_based_recovery_op() = 0;
+  virtual void schedule_event_after(PGPeeringEventRef evt, float delay) = 0;
 };