Merge pull request ceph#54304 from cyx1231st/wip-crimson-make-crosscore-send-ordered

crimson/osd: support to send messages concurrently with exclusive phases

Reviewed-by: Samuel Just <[email protected]>
Reviewed-by: chunmei-liu <[email protected]>

Author: Matan-B

src/crimson/common/smp_helpers.h

@@ -3,10 +3,16 @@
 
 #pragma once
 
+#include <concepts>
 #include <limits>
+#include <optional>
+#include <type_traits>
+#include <vector>
 
+#include <seastar/core/shared_future.hh>
 #include <seastar/core/smp.hh>
 
+#include "common/likely.h"
 #include "crimson/common/errorator.h"
 #include "crimson/common/utility.h"
 
@@ -89,4 +95,142 @@ 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
+};
+
+/**
+ * smp_crosscore_ordering_t
+ *
+ * To preserve the event order from source to target core(s).
+ */
+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 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)
+    : out_seqs(seastar::smp::count, 0),
+      in_controls(seastar::smp::count) {}
+
+  ~smp_crosscore_ordering_t() = default;
+
+  /*
+   * Called by the original core to get the ordering sequence
+   */
+
+  seq_t prepare_submit() requires IS_ONE {
+    return do_prepare_submit(out_seqs);
+  }
+
+  seq_t prepare_submit(core_id_t target_core) requires IS_ONE_N {
+    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
+   */
+
+  seq_t get_in_seq() const requires IS_ONE {
+    return in_controls.seq;
+  }
+
+  seq_t get_in_seq() const requires IS_ONE_N {
+    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);
+  }
+
+  bool proceed_or_wait(seq_t seq) requires IS_ONE_N {
+    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);
+  }
+
+  seastar::future<> wait(seq_t seq) requires IS_ONE_N {
+    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;
+    std::optional<seastar::shared_promise<>> pr_wait;
+
+    bool proceed_or_wait(seq_t in_seq) {
+      if (in_seq == seq + 1) {
+        ++seq;
+        reset_wait();
+        return true;
+      } else {
+        return false;
+      }
+    }
+
+    seastar::future<> wait(seq_t in_seq) {
+      assert(in_seq != seq + 1);
+      if (!pr_wait.has_value()) {
+        pr_wait = seastar::shared_promise<>();
+      }
+      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) {
+    return ++out_seq;
+  }
+
+  std::conditional_t<
+    IS_ONE,
+    seq_t, std::vector<seq_t>
+  > out_seqs;
+
+  std::conditional_t<
+    IS_ONE,
+    in_control_t, std::vector<in_control_t>
+  > in_controls;
+};
+
+} // namespace crimson

src/crimson/net/Connection.h

@@ -81,19 +81,44 @@ 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.
+   *
+   * The returned future will be resolved only after the message is enqueued
+   * remotely.
    */
-  virtual seastar::future<> send(MessageURef msg) = 0;
+  virtual seastar::future<> send(
+      MessageURef msg) = 0;
+
+  /**
+   * send_with_throttling
+   *
+   * Send a message over a connection that has completed its handshake.
+   *
+   * May be invoked from any core, and the send order will be preserved upon
+   * the call.
+   *
+   * TODO:
+   *
+   * The returned future is reserved for throttling.
+   *
+   * Gating is needed for graceful shutdown, to wait until the message is
+   * enqueued remotely.
+   */
+  seastar::future<> send_with_throttling(
+      MessageURef msg /* , seastar::gate & */) {
+    std::ignore = send(std::move(msg));
+    return seastar::now();
+  }
 
   /**
    * send_keepalive
    *
    * 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/Fwd.h

@@ -21,7 +21,7 @@
 #include <seastar/core/sharded.hh>
 
 #include "msg/Connection.h"
-#include "msg/MessageRef.h"
+#include "msg/Message.h"
 #include "msg/msg_types.h"
 
 #include "crimson/common/errorator.h"

src/crimson/net/ProtocolV2.cc

@@ -2073,7 +2073,7 @@ void ProtocolV2::trigger_replacing(bool reconnect,
 // READY state
 
 seastar::future<> ProtocolV2::notify_out_fault(
-    crosscore_t::seq_t cc_seq,
+    cc_seq_t cc_seq,
     const char *where,
     std::exception_ptr eptr,
     io_handler_state _io_states)
@@ -2121,7 +2121,7 @@ void ProtocolV2::execute_standby()
 }
 
 seastar::future<> ProtocolV2::notify_out(
-    crosscore_t::seq_t cc_seq)
+    cc_seq_t cc_seq)
 {
   assert(seastar::this_shard_id() == conn.get_messenger_shard_id());
   if (!crosscore.proceed_or_wait(cc_seq)) {
@@ -2210,7 +2210,7 @@ void ProtocolV2::execute_server_wait()
 // CLOSING state
 
 seastar::future<> ProtocolV2::notify_mark_down(
-    crosscore_t::seq_t cc_seq)
+    cc_seq_t cc_seq)
 {
   assert(seastar::this_shard_id() == conn.get_messenger_shard_id());
   if (!crosscore.proceed_or_wait(cc_seq)) {

src/crimson/net/ProtocolV2.h

@@ -29,16 +29,16 @@ class ProtocolV2 final : public HandshakeListener {
  */
 private:
   seastar::future<> notify_out(
-      crosscore_t::seq_t cc_seq) final;
+      cc_seq_t cc_seq) final;
 
   seastar::future<> notify_out_fault(
-      crosscore_t::seq_t cc_seq,
+      cc_seq_t cc_seq,
       const char *where,
       std::exception_ptr,
       io_handler_state) final;
 
   seastar::future<> notify_mark_down(
-      crosscore_t::seq_t cc_seq) final;
+      cc_seq_t cc_seq) final;
 
 /*
 * as ProtocolV2 to be called by SocketConnection
@@ -251,7 +251,7 @@ class ProtocolV2 final : public HandshakeListener {
   // asynchronously populated from io_handler
   io_handler_state io_states;
 
-  crosscore_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;