crimson/osd: execute PGListener async operations in order

- Adds ShardServices::singleton_orderer_t mechanism to ensure that
  OSDSingleton calls are completed in order.
- Updates ShardServices accessors invoked from PeeringListener handlers
  to use orderer.
- Updates PGListener handlers and complete_rctx to use orderer.

Fixes: https://tracker.ceph.com/issues/66316
Signed-off-by: Samuel Just <[email protected]>

Author: athanatos

src/crimson/osd/pg.h

@@ -166,17 +166,37 @@ class PG : public boost::intrusive_ref_counter<
     return std::size(snap_trimq);
   }
 
+  /**
+   * complete_rctx
+   *
+   * complete_rctx is responsible for submitting writes and messages
+   * resulting from processing a PeeringState event as well as resolving
+   * any asyncronous actions initiated by the PeeringState::Listener
+   * callbacks below.  The caller is responsible for calling complete_rctx
+   * and waiting for the future to resolve before exiting the
+   * PGPeeringPipeline::process stage (see osd_operations/peering_event.h).
+   *
+   * orderer below ensures that operations submitted on the OSD-wide
+   * OSDSingleton instance are completed in the order initiated.  This is
+   * specifically important for operations on the local and remote async
+   * reserver instances, as well as setting and clearing pg_temp mapping
+   * requests.
+   */
+  ShardServices::singleton_orderer_t orderer;
   seastar::future<> complete_rctx(PeeringCtx &&rctx) {
-    return seastar::when_all_succeed(
-      get_need_up_thru()
-      ? shard_services.send_alive(
-	get_same_interval_since())
-      : seastar::now(),
+    shard_services.send_pg_temp(orderer);
+    if (get_need_up_thru()) {
+      shard_services.send_alive(orderer, get_same_interval_since());
+    }
+
+    ShardServices::singleton_orderer_t o;
+    std::swap(o, orderer);
+    return seastar::when_all(
       shard_services.dispatch_context(
 	get_collection_ref(),
 	std::move(rctx)),
-      shard_services.send_pg_temp()
-    ).then([](auto){});
+      shard_services.run_orderer(std::move(o))
+    ).then([](auto) {});
   }
 
   void send_cluster_message(
@@ -186,13 +206,21 @@ class PG : public boost::intrusive_ref_counter<
     SUBDEBUGDPP(
       osd, "message {} to {} share_map_update {}",
       *this, *m, osd, share_map_update);
-    (void)shard_services.send_to_osd(osd, std::move(m), epoch);
+    /* We don't bother to queue this one in the orderer because capturing the
+     * message ref in std::function is problematic as it isn't copyable.  This
+     * is solvable, but it's not quite worth the effort at the moment as we
+     * aren't worried about ordering of message send events except between
+     * messages to the same target within an interval, which doesn't really
+     * happen while processing a single event.  It'll probably be worth
+     * generalizing the orderer structure to fix this in the future, probably
+     * by using std::move_only_function once widely available. */
+    std::ignore = shard_services.send_to_osd(osd, std::move(m), epoch);
   }
 
   void send_pg_created(pg_t pgid) final {
     LOG_PREFIX(PG::send_pg_created);
     SUBDEBUGDPP(osd, "pgid {}", *this, pgid);
-    (void)shard_services.send_pg_created(pgid);
+    shard_services.send_pg_created(orderer, pgid);
   }
 
   bool try_flush_or_schedule_async() final;
@@ -237,37 +265,35 @@ class PG : public boost::intrusive_ref_counter<
     SUBDEBUGDPP(
       osd, "priority {} on_grant {} on_preempt {}",
       *this, on_grant->get_desc(), on_preempt->get_desc());
-    // TODO -- we probably want to add a mechanism for blocking on this
-    // after handling the peering event
-    std::ignore = shard_services.local_request_reservation(
+    shard_services.local_request_reservation(
+      orderer,
       pgid,
       on_grant ? make_lambda_context([this, on_grant=std::move(on_grant)] (int) {
 	start_peering_event_operation(std::move(*on_grant));
       }) : nullptr,
       priority,
       on_preempt ? make_lambda_context(
 	[this, on_preempt=std::move(on_preempt)] (int) {
-	start_peering_event_operation(std::move(*on_preempt));
-      }) : nullptr);
+	  start_peering_event_operation(std::move(*on_preempt));
+	}) : nullptr
+    );
   }
 
   void update_local_background_io_priority(
     unsigned priority) final {
     LOG_PREFIX(PG::update_local_background_io_priority);
     SUBDEBUGDPP(osd, "priority {}", *this, priority);
-    // TODO -- we probably want to add a mechanism for blocking on this
-    // after handling the peering event
-    std::ignore = shard_services.local_update_priority(
+    shard_services.local_update_priority(
+      orderer,
       pgid,
       priority);
   }
 
   void cancel_local_background_io_reservation() final {
     LOG_PREFIX(PG::cancel_local_background_io_reservation);
     SUBDEBUGDPP(osd, "", *this);
-    // TODO -- we probably want to add a mechanism for blocking on this
-    // after handling the peering event
-    std::ignore = shard_services.local_cancel_reservation(
+    shard_services.local_cancel_reservation(
+      orderer,
       pgid);
   }
 
@@ -279,27 +305,24 @@ class PG : public boost::intrusive_ref_counter<
     SUBDEBUGDPP(
       osd, "priority {} on_grant {} on_preempt {}",
       *this, on_grant->get_desc(), on_preempt->get_desc());
-    // TODO -- we probably want to add a mechanism for blocking on this
-    // after handling the peering event
-    std::ignore = shard_services.remote_request_reservation(
+    shard_services.remote_request_reservation(
+      orderer,
       pgid,
       on_grant ? make_lambda_context([this, on_grant=std::move(on_grant)] (int) {
 	start_peering_event_operation(std::move(*on_grant));
       }) : nullptr,
       priority,
       on_preempt ? make_lambda_context(
 	[this, on_preempt=std::move(on_preempt)] (int) {
-	start_peering_event_operation(std::move(*on_preempt));
-      }) : nullptr);
+	  start_peering_event_operation(std::move(*on_preempt));
+      }) : nullptr
+    );
   }
 
   void cancel_remote_recovery_reservation() final {
     LOG_PREFIX(PG::cancel_remote_recovery_reservation);
     SUBDEBUGDPP(osd, "", *this);
-    // TODO -- we probably want to add a mechanism for blocking on this
-    // after handling the peering event
-    std::ignore =  shard_services.remote_cancel_reservation(
-      pgid);
+    shard_services.remote_cancel_reservation(orderer, pgid);
   }
 
   void schedule_event_on_commit(
@@ -326,16 +349,12 @@ class PG : public boost::intrusive_ref_counter<
   void queue_want_pg_temp(const std::vector<int> &wanted) final {
     LOG_PREFIX(PG::queue_want_pg_temp);
     SUBDEBUGDPP(osd, "wanted {}", *this, wanted);
-    // TODO -- we probably want to add a mechanism for blocking on this
-    // after handling the peering event
-    std::ignore = shard_services.queue_want_pg_temp(pgid.pgid, wanted);
+    shard_services.queue_want_pg_temp(orderer, pgid.pgid, wanted);
   }
   void clear_want_pg_temp() final {
     LOG_PREFIX(PG::clear_want_pg_temp);
     SUBDEBUGDPP(osd, "", *this);
-    // TODO -- we probably want to add a mechanism for blocking on this
-    // after handling the peering event
-    std::ignore = shard_services.remove_want_pg_temp(pgid.pgid);
+    shard_services.remove_want_pg_temp(orderer, pgid.pgid);
   }
   void check_recovery_sources(const OSDMapRef& newmap) final {
     // Not needed yet