crimson/net: preserve the ordering upon the calls to Connection::send()/keepalive()

Signed-off-by: Yingxin Cheng <[email protected]>

Author: cyx1231st

src/crimson/common/smp_helpers.h

@@ -98,6 +98,7 @@ auto sharded_map_seq(T &t, F &&f) {
 enum class crosscore_type_t {
   ONE,   // from 1 to 1 core
   ONE_N, // from 1 to n cores
+  N_ONE, // from n to 1 core
 };
 
 /**
@@ -109,15 +110,16 @@ template <crosscore_type_t CTypeValue>
 class smp_crosscore_ordering_t {
   static constexpr bool IS_ONE = (CTypeValue == crosscore_type_t::ONE);
   static constexpr bool IS_ONE_N = (CTypeValue == crosscore_type_t::ONE_N);
-  static_assert(IS_ONE || IS_ONE_N);
+  static constexpr bool IS_N_ONE = (CTypeValue == crosscore_type_t::N_ONE);
+  static_assert(IS_ONE || IS_ONE_N || IS_N_ONE);
 
 public:
   using seq_t = uint64_t;
 
   smp_crosscore_ordering_t() requires IS_ONE
     : out_seqs(0) { }
 
-  smp_crosscore_ordering_t() requires IS_ONE_N
+  smp_crosscore_ordering_t() requires (!IS_ONE)
     : out_seqs(seastar::smp::count, 0),
       in_controls(seastar::smp::count) {}
 
@@ -135,6 +137,10 @@ class smp_crosscore_ordering_t {
     return do_prepare_submit(out_seqs[target_core]);
   }
 
+  seq_t prepare_submit() requires IS_N_ONE {
+    return do_prepare_submit(out_seqs[seastar::this_shard_id()]);
+  }
+
   /*
    * Called by the target core to preserve the ordering
    */
@@ -147,6 +153,10 @@ class smp_crosscore_ordering_t {
     return in_controls[seastar::this_shard_id()].seq;
   }
 
+  seq_t get_in_seq(core_id_t source_core) const requires IS_N_ONE {
+    return in_controls[source_core].seq;
+  }
+
   bool proceed_or_wait(seq_t seq) requires IS_ONE {
     return in_controls.proceed_or_wait(seq);
   }
@@ -155,6 +165,10 @@ class smp_crosscore_ordering_t {
     return in_controls[seastar::this_shard_id()].proceed_or_wait(seq);
   }
 
+  bool proceed_or_wait(seq_t seq, core_id_t source_core) requires IS_N_ONE {
+    return in_controls[source_core].proceed_or_wait(seq);
+  }
+
   seastar::future<> wait(seq_t seq) requires IS_ONE {
     return in_controls.wait(seq);
   }
@@ -163,6 +177,16 @@ class smp_crosscore_ordering_t {
     return in_controls[seastar::this_shard_id()].wait(seq);
   }
 
+  seastar::future<> wait(seq_t seq, core_id_t source_core) requires IS_N_ONE {
+    return in_controls[source_core].wait(seq);
+  }
+
+  void reset_wait() requires IS_N_ONE {
+    for (auto &in_control : in_controls) {
+      in_control.reset_wait();
+    }
+  }
+
 private:
   struct in_control_t {
     seq_t seq = 0;
@@ -171,10 +195,7 @@ class smp_crosscore_ordering_t {
     bool proceed_or_wait(seq_t in_seq) {
       if (in_seq == seq + 1) {
         ++seq;
-        if (unlikely(pr_wait.has_value())) {
-          pr_wait->set_value();
-          pr_wait = std::nullopt;
-        }
+        reset_wait();
         return true;
       } else {
         return false;
@@ -188,6 +209,13 @@ class smp_crosscore_ordering_t {
       }
       return pr_wait->get_shared_future();
     }
+
+    void reset_wait() {
+      if (unlikely(pr_wait.has_value())) {
+        pr_wait->set_value();
+        pr_wait = std::nullopt;
+      }
+    }
   };
 
   seq_t do_prepare_submit(seq_t &out_seq) {

src/crimson/net/Connection.h

@@ -81,8 +81,8 @@ class Connection : public seastar::enable_shared_from_this<Connection> {
    *
    * Send a message over a connection that has completed its handshake.
    *
-   * May be invoked from any core, but that requires to chain the returned
-   * future to preserve ordering.
+   * May be invoked from any core, and the send order will be preserved upon
+   * the call.
    */
   virtual seastar::future<> send(MessageURef msg) = 0;
 
@@ -92,8 +92,8 @@ class Connection : public seastar::enable_shared_from_this<Connection> {
    * Send a keepalive message over a connection that has completed its
    * handshake.
    *
-   * May be invoked from any core, but that requires to chain the returned
-   * future to preserve ordering.
+   * May be invoked from any core, and the send order will be preserved upon
+   * the call.
    */
   virtual seastar::future<> send_keepalive() = 0;
 

src/crimson/net/ProtocolV2.h

@@ -251,7 +251,7 @@ class ProtocolV2 final : public HandshakeListener {
   // asynchronously populated from io_handler
   io_handler_state io_states;
 
-  crosscore_ordering_t crosscore;
+  proto_crosscore_ordering_t crosscore;
 
   bool has_socket = false;
 

src/crimson/net/SocketConnection.cc

@@ -79,16 +79,13 @@ bool SocketConnection::peer_wins() const
   return (messenger.get_myaddr() > peer_addr || policy.server);
 }
 
-seastar::future<> SocketConnection::send(MessageURef _msg)
+seastar::future<> SocketConnection::send(MessageURef msg)
 {
-  // may be invoked from any core
-  MessageFRef msg = seastar::make_foreign(std::move(_msg));
   return io_handler->send(std::move(msg));
 }
 
 seastar::future<> SocketConnection::send_keepalive()
 {
-  // may be invoked from any core
   return io_handler->send_keepalive();
 }
 

src/crimson/net/SocketConnection.h

@@ -54,7 +54,7 @@ class ConnectionHandler {
 
   virtual bool is_connected() const = 0;
 
-  virtual seastar::future<> send(MessageFRef) = 0;
+  virtual seastar::future<> send(MessageURef) = 0;
 
   virtual seastar::future<> send_keepalive() = 0;