From 6bb4f00a9d006f99b21104bf46b78dff269a116f Mon Sep 17 00:00:00 2001
From: chunmei <chunmei.liu@intel.com>
Date: Fri, 31 Mar 2023 04:11:27 +0000
Subject: [PATCH] crimson/osd: relax requirement that PGShardManager methods be
 invoked on PRIMARY_CORE

Proxy uses of osd_singleton_state to PRIMARY_CORE as needed.

Signed-off-by: chunmei <chunmei.liu@intel.com>
---
 src/crimson/osd/pg_shard_manager.h | 30 +++++++++++++-----------------
 1 file changed, 13 insertions(+), 17 deletions(-)

diff --git a/src/crimson/osd/pg_shard_manager.h b/src/crimson/osd/pg_shard_manager.h
index 8333b1f483b64..f4dac6471f39e 100644
--- a/src/crimson/osd/pg_shard_manager.h
+++ b/src/crimson/osd/pg_shard_manager.h
@@ -64,11 +64,9 @@ public:
     return osd_singleton_state.local();
   }
   auto &get_shard_services() {
-    ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
     return shard_services.local();
   }
   auto &get_shard_services() const {
-    ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
     return shard_services.local();
   }
   auto &get_local_state() { return get_shard_services().local_state; }
@@ -140,7 +138,6 @@ public:
 
   template <typename F>
   auto with_remote_shard_state(core_id_t core, F &&f) {
-    ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
     return shard_services.invoke_on(
       core, [f=std::move(f)](auto &target_shard_services) mutable {
 	return std::invoke(
@@ -154,7 +151,6 @@ public:
       core_id_t core,
       typename T::IRef &&op,
       F &&f) {
-    ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
     if (seastar::this_shard_id() == core) {
       auto &target_shard_services = shard_services.local();
       return std::invoke(
@@ -186,7 +182,6 @@ public:
     typename T::IRef op
   ) {
     ceph_assert(op->use_count() == 1);
-    ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
     auto &logger = crimson::get_logger(ceph_subsys_osd);
     static_assert(T::can_create());
     logger.debug("{}: can_create", *op);
@@ -231,7 +226,6 @@ public:
     typename T::IRef op
   ) {
     ceph_assert(op->use_count() == 1);
-    ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
     auto &logger = crimson::get_logger(ceph_subsys_osd);
     static_assert(!T::can_create());
     logger.debug("{}: !can_create", *op);
@@ -337,7 +331,6 @@ public:
 
   template <typename T, typename... Args>
   auto start_pg_operation(Args&&... args) {
-    ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
     auto op = get_local_state().registry.create_operation<T>(
       std::forward<Args>(args)...);
     auto &logger = crimson::get_logger(ceph_subsys_osd);
@@ -351,35 +344,38 @@ public:
     auto fut = opref.template enter_stage<>(
       opref.get_connection_pipeline().await_active
     ).then([this, &opref, &logger] {
-      ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
       logger.debug("{}: start_pg_operation in await_active stage", opref);
-      return get_osd_singleton_state().osd_state.when_active();
+      return osd_singleton_state.invoke_on(PRIMARY_CORE, []
+      (auto& primary_osd_singleton_state) {
+        return primary_osd_singleton_state.osd_state.when_active();
+      });
     }).then([&logger, &opref] {
-      ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
       logger.debug("{}: start_pg_operation active, entering await_map", opref);
       return opref.template enter_stage<>(
 	opref.get_connection_pipeline().await_map);
     }).then([this, &logger, &opref] {
-      ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
       logger.debug("{}: start_pg_operation await_map stage", opref);
       using OSDMapBlockingEvent =
 	OSD_OSDMapGate::OSDMapBlocker::BlockingEvent;
       return opref.template with_blocking_event<OSDMapBlockingEvent>(
 	[this, &opref](auto &&trigger) {
 	  std::ignore = this;
-	  return get_osd_singleton_state().osdmap_gate.wait_for_map(
-	    std::move(trigger),
-	    opref.get_epoch(),
-	    &get_shard_services());
+          return osd_singleton_state.invoke_on(PRIMARY_CORE,
+            [trigger = std::move(trigger), epoch = opref.get_epoch(),
+             &shard_services = shard_services]
+            (auto& primary_osd_singleton_state) mutable {
+	    return primary_osd_singleton_state.osdmap_gate.wait_for_map(
+	      std::move(trigger),
+	      epoch,
+	      &shard_services.local());
 	});
+      });
     }).then([&logger, &opref](auto epoch) {
-      ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
       logger.debug("{}: got map {}, entering get_pg", opref, epoch);
       return opref.template enter_stage<>(
 	opref.get_connection_pipeline().get_pg);
     }).then([this, &logger, &opref, op=std::move(op)]() mutable {
       logger.debug("{}: in get_pg core {}", opref, seastar::this_shard_id());
-      ceph_assert(seastar::this_shard_id() == PRIMARY_CORE);
       logger.debug("{}: in get_pg", opref);
       if constexpr (T::can_create()) {
 	logger.debug("{}: can_create", opref);
-- 
2.39.5