crimson/net: support cross-core blocking for interceptor

Signed-off-by: Yingxin Cheng <yingxin.cheng@intel.com>
(cherry picked from commit bcf20d6771f5adf608fcf44b47073c133c6b0b35)

diff --git a/src/crimson/net/Interceptor.h b/src/crimson/net/Interceptor.h
index 921175cbf9417104e170a01ab843ef661cca6fd7..35b74e24369c4d6aa9db657ff052b5c8c6412b59 100644 (file)
--- a/src/crimson/net/Interceptor.h
+++ b/src/crimson/net/Interceptor.h
@@ -45,16 +45,21 @@ enum class bp_action_t {
 class socket_blocker {
   std::optional<seastar::abort_source> p_blocked;
   std::optional<seastar::abort_source> p_unblocked;
+  const seastar::shard_id primary_sid;
 
  public:
+  socket_blocker() : primary_sid{seastar::this_shard_id()} {}
+
   seastar::future<> wait_blocked() {
+    ceph_assert(seastar::this_shard_id() == primary_sid);
     ceph_assert(!p_blocked);
     if (p_unblocked) {
       return seastar::make_ready_future<>();
     } else {
       p_blocked = seastar::abort_source();
-      return seastar::sleep_abortable(std::chrono::seconds(10),
-                                     *p_blocked).then([] {
+      return seastar::sleep_abortable(
+        std::chrono::seconds(10), *p_blocked
+      ).then([] {
         throw std::runtime_error(
             "Timeout (10s) in socket_blocker::wait_blocked()");
       }).handle_exception_type([] (const seastar::sleep_aborted& e) {
@@ -64,21 +69,25 @@ class socket_blocker {
   }
 
   seastar::future<> block() {
-    if (p_blocked) {
-      p_blocked->request_abort();
-      p_blocked = std::nullopt;
-    }
-    ceph_assert(!p_unblocked);
-    p_unblocked = seastar::abort_source();
-    return seastar::sleep_abortable(std::chrono::seconds(10),
-                                   *p_unblocked).then([] {
-      ceph_abort("Timeout (10s) in socket_blocker::block()");
-    }).handle_exception_type([] (const seastar::sleep_aborted& e) {
-      // wait done!
+    return seastar::smp::submit_to(primary_sid, [this] {
+      if (p_blocked) {
+        p_blocked->request_abort();
+        p_blocked = std::nullopt;
+      }
+      ceph_assert(!p_unblocked);
+      p_unblocked = seastar::abort_source();
+      return seastar::sleep_abortable(
+        std::chrono::seconds(10), *p_unblocked
+      ).then([] {
+        ceph_abort("Timeout (10s) in socket_blocker::block()");
+      }).handle_exception_type([] (const seastar::sleep_aborted& e) {
+        // wait done!
+      });
     });
   }
 
   void unblock() {
+    ceph_assert(seastar::this_shard_id() == primary_sid);
     ceph_assert(!p_blocked);
     ceph_assert(p_unblocked);
     p_unblocked->request_abort();