[coro_io][rdma] enhance rdma implement (#968)

* speed up write

* remove tmp

* add max_buffer

* fix

* speed up add buffer

* change buffer to 2m

* fix crash

* fix

* fix

* test inline data

* support inline send

* fix

* fix

* rmda support notify peer when close

* fix soc

* fix crash

* fix test

* fix format

* fix code

* add doc

* fix rdma

* fix

* fix for lifetime

---------

Co-authored-by: qicosmos <qicosmos@linux.alibaba.com>

Author: poor-circle

cmake/find_ibverbs.cmake

@@ -1,5 +1,5 @@
-check_library_exists(ibverbs ibv_create_qp "" HAVE_IBVERBS_IBV)
-if(HAVE_IBVERBS_IBV)
+check_library_exists(ibverbs ibv_create_qp "" YLT_HAVE_IBVERBS)
+if(YLT_HAVE_IBVERBS)
   set(YLT_ENABLE_IBV ON)
   set(IBVERBS_LIBRARY ibverbs)
   message(STATUS "have libibverbs")

include/ylt/coro_io/ibverbs/ib_buffer.hpp

@@ -80,13 +80,14 @@ struct ib_buffer_t {
   ~ib_buffer_t();
 };
 
+std::shared_ptr<ib_device_t> g_ib_device(ib_config_t conf = {});
 class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
  private:
   friend struct ib_buffer_t;
   struct private_construct_token {};
   static async_simple::coro::Lazy<void> collect_idle_timeout_client(
       std::weak_ptr<ib_buffer_pool_t> self_weak,
-      std::chrono::milliseconds sleep_time, std::size_t clear_cnt) {
+      std::chrono::milliseconds sleep_time) {
     std::shared_ptr<ib_buffer_pool_t> self = self_weak.lock();
     if (self == nullptr) {
       co_return;
@@ -101,7 +102,7 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
       while (true) {
         ELOG_TRACE << "start collect timeout buffer of pool{" << self.get()
                    << "}, now client count: " << self->free_buffers_.size();
-        std::size_t is_all_cleared = self->free_buffers_.clear_old(clear_cnt);
+        std::size_t is_all_cleared = self->free_buffers_.clear_old(1000);
         ELOG_TRACE << "finish collect timeout buffer of pool{" << self.get()
                    << "}, now client cnt: " << self->free_buffers_.size();
         if (is_all_cleared != 0) [[unlikely]] {
@@ -134,8 +135,7 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
                    << "}";
         collect_idle_timeout_client(this->weak_from_this(),
                                     (std::max)(pool_config_.idle_timeout,
-                                               std::chrono::milliseconds{50}),
-                                    pool_config_.idle_queue_per_max_clear_count)
+                                               std::chrono::milliseconds{50}))
             .directlyStart(
                 [](auto&&) {
                 },
@@ -210,15 +210,15 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
     return buffer;
   }
   struct config_t {
-    size_t buffer_size = 4 * 1024 * 1024;                   // 4MB
+    size_t buffer_size = 2 * 1024 * 1024;                   // 2MB
     uint64_t max_memory_usage = 4ull * 1024 * 1024 * 1024;  // 4GB
-    size_t idle_queue_per_max_clear_count = 1000;
     std::chrono::milliseconds idle_timeout = std::chrono::milliseconds{5000};
   };
   ib_buffer_pool_t(private_construct_token t,
                    std::shared_ptr<ib_device_t> device,
                    const config_t& pool_config)
-      : device_(std::move(device)), pool_config_(std::move(pool_config)) {}
+      : device_(device ? std::move(device) : coro_io::g_ib_device()),
+        pool_config_(std::move(pool_config)) {}
   static std::shared_ptr<ib_buffer_pool_t> create(
       std::shared_ptr<ib_device_t> device, const config_t& pool_config) {
     return std::make_shared<ib_buffer_pool_t>(private_construct_token{},
@@ -245,17 +245,29 @@ inline ib_buffer_t ib_buffer_t::regist(ib_buffer_pool_t& pool,
                                        int ib_flags) {
   auto mr = ibv_reg_mr(dev->pd(), ptr, size, ib_flags);
   if (mr != nullptr) [[unlikely]] {
-    ELOG_INFO << "ibv_reg_mr regist: " << mr << " with pd:" << dev->pd();
+    ELOG_DEBUG << "ibv_reg_mr regist: " << mr << " with pd:" << dev->pd();
     return ib_buffer_t{mr, std::move(dev), pool.weak_from_this()};
   }
   else {
     throw std::make_error_code(std::errc{errno});
   }
 };
+
+inline std::shared_ptr<ib_device_t> g_ib_device(ib_config_t conf) {
+  static auto dev = std::make_shared<ib_device_t>(conf);
+  return dev;
+}
+
+inline std::shared_ptr<ib_buffer_pool_t> g_ib_buffer_pool(
+    const ib_buffer_pool_t::config_t& pool_config = {}) {
+  static auto pool = ib_buffer_pool_t::create(g_ib_device(), pool_config);
+  return pool;
+}
+
 inline ib_buffer_t::ib_buffer_t(ibv_mr* mr, std::shared_ptr<ib_device_t> dev,
                                 std::weak_ptr<ib_buffer_pool_t> owner_pool)
     : mr_(mr),
-      dev_(std::move(dev)),
+      dev_(dev ? std::move(dev) : coro_io::g_ib_device()),
       owner_pool_(std::move(owner_pool)),
       has_memory_ownership_(false) {
   if (auto ptr = owner_pool_.lock(); ptr) {
@@ -299,14 +311,4 @@ inline void ib_buffer_t::change_owner(
   }
 }
 
-inline std::shared_ptr<ib_device_t> g_ib_device(ib_config_t conf = {}) {
-  static auto dev = std::make_shared<ib_device_t>(conf);
-  return dev;
-}
-
-inline std::shared_ptr<ib_buffer_pool_t> g_ib_buffer_pool(
-    const ib_buffer_pool_t::config_t& pool_config = {}) {
-  static auto pool = ib_buffer_pool_t::create(g_ib_device(), pool_config);
-  return pool;
-}
 }  // namespace coro_io

include/ylt/coro_io/ibverbs/ib_device.hpp

@@ -211,6 +211,11 @@ class ib_device_t {
   asio::ip::address gid_address() const noexcept { return gid_address_; }
   const ibv_port_attr& attr() const noexcept { return attr_; }
 
+  bool is_support_inline_data() const noexcept { return support_inline_data_; }
+  void set_support_inline_data(bool flag) noexcept {
+    support_inline_data_ = flag;
+  }
+
  private:
   int ipv6_addr_v4mapped(const struct in6_addr* a) {
     return ((a->s6_addr32[0] | a->s6_addr32[1]) |
@@ -243,6 +248,7 @@ class ib_device_t {
   std::string name_;
   std::unique_ptr<ibv_pd, ib_deleter> pd_;
   std::unique_ptr<ibv_context, ib_deleter> ctx_;
+  std::atomic<bool> support_inline_data_ = true;
   ibv_port_attr attr_;
   ibv_gid gid_;
   asio::ip::address gid_address_;

include/ylt/coro_io/ibverbs/ib_io.hpp

@@ -8,6 +8,7 @@
 #include <exception>
 #include <future>
 #include <memory>
+#include <optional>
 #include <span>
 #include <string_view>
 #include <system_error>
@@ -19,6 +20,7 @@
 #include "async_simple/Promise.h"
 #include "async_simple/Signal.h"
 #include "async_simple/coro/Collect.h"
+#include "async_simple/coro/FutureAwaiter.h"
 #include "async_simple/coro/Lazy.h"
 #include "ib_buffer.hpp"
 #include "ib_socket.hpp"
@@ -30,13 +32,12 @@
 #include "ylt/struct_pack/reflection.hpp"
 namespace coro_io {
 
-// unlike tcp socket, client won't connnected util server first read ib_socket.
 inline async_simple::coro::Lazy<std::error_code> async_accept(
     asio::ip::tcp::acceptor& acceptor, coro_io::ib_socket_t& ib_socket) {
-  auto soc = std::make_unique<asio::ip::tcp::socket>(ib_socket.get_executor());
+  asio::ip::tcp::socket soc(ib_socket.get_executor());
   auto ec = co_await async_io<std::error_code>(
       [&](auto cb) {
-        acceptor.async_accept(*soc, cb);
+        acceptor.async_accept(soc, cb);
       },
       acceptor);
 
@@ -51,24 +52,13 @@ inline async_simple::coro::Lazy<std::error_code> async_accept(
 inline async_simple::coro::Lazy<std::error_code> async_connect(
     coro_io::ib_socket_t& ib_socket, const std::string& host,
     const std::string& port) {
-  asio::ip::tcp::socket soc{ib_socket.get_executor()};
-  auto ec =
-      co_await async_connect(ib_socket.get_coro_executor(), soc, host, port);
-  if (ec) [[unlikely]] {
-    co_return std::move(ec);
-  }
-  co_return co_await ib_socket.connect(soc);
+  return ib_socket.connect(host, port);
 }
 
 template <typename EndPointSeq>
 inline async_simple::coro::Lazy<std::error_code> async_connect(
     coro_io::ib_socket_t& ib_socket, const EndPointSeq& endpoint) noexcept {
-  asio::ip::tcp::socket soc{ib_socket.get_executor()};
-  auto ec = co_await async_connect(soc, endpoint);
-  if (ec) [[unlikely]] {
-    co_return std::move(ec);
-  }
-  co_return co_await ib_socket.connect(soc);
+  return ib_socket.connect(endpoint);
 }
 
 namespace detail {
@@ -138,32 +128,64 @@ async_simple::coro::
 async_simple::coro::
     Lazy<std::pair<std::error_code, std::size_t>> inline async_send_impl(
         coro_io::ib_socket_t& ib_socket, std::span<ibv_sge> sge_list,
-        std::size_t io_size) {
-  std::vector<ib_buffer_t> tmp_buffers;
+        std::size_t io_size,
+        std::optional<
+            async_simple::Future<std::pair<std::error_code, std::size_t>>>&
+            prev_op) {
+  if (io_size == 0) [[unlikely]] {
+    co_return std::pair{std::error_code{}, 0};
+  }
+  ib_buffer_t buffer;
   ibv_sge socket_buffer;
-  socket_buffer = ib_socket.get_send_buffer();
-  if (socket_buffer.length == 0) [[unlikely]] {
-    co_return std::pair{std::make_error_code(std::errc::no_buffer_space),
-                        std::size_t{0}};
+  std::span<ibv_sge> list;
+  if (ib_socket.get_config().cap.max_inline_data >= io_size) {
+    if (sge_list.size() <= ib_socket.get_config().cap.max_send_sge) {
+      list = sge_list;
+    }
+  }
+  else {
+    buffer = ib_socket.buffer_pool()->get_buffer();
+    if (!buffer || buffer->length < io_size) [[unlikely]] {
+      co_return std::pair{std::make_error_code(std::errc::no_buffer_space),
+                          std::size_t{0}};
+    }
+    socket_buffer = buffer.subview();
+    auto len = copy(sge_list, socket_buffer);
+    assert(len == io_size);
+    socket_buffer.length = io_size;
+    list = {&socket_buffer, 1};
   }
 
-  auto len = copy(sge_list, socket_buffer);
-  assert(len == io_size);
-
-  std::span<ibv_sge> io_buffer;
-  io_buffer = {&socket_buffer, 1};
-  socket_buffer.length = io_size;
-  auto result =
-      co_await coro_io::async_io<std::pair<std::error_code, std::size_t>>(
-          [&](auto&& cb) {
-            ib_socket.post_send(io_buffer, std::move(cb));
-          },
-          ib_socket);
-  if (result.first) [[unlikely]] {
-    co_return std::pair{result.first, result.second};
+  async_simple::Promise<std::pair<std::error_code, std::size_t>> promise;
+  std::pair<std::error_code, std::size_t> result{};
+  if (prev_op) {
+    result = co_await std::move(*prev_op);
+  }
+  prev_op = promise.getFuture();
+  auto slot = co_await async_simple::coro::CurrentSlot{};
+  auto work = coro_io::async_io<std::pair<std::error_code, std::size_t>>(
+      [&ib_socket, list](auto&& cb) mutable {
+        ib_socket.post_send(list, std::move(cb));
+      },
+      ib_socket);
+  auto cb = [p = std::move(promise), io_size = io_size,
+             buffer = std::move(buffer)](auto&& result) mutable {
+    if (buffer) {
+      std::move(buffer).collect();
+    }
+    if (!result.hasError()) {
+      result.value().second = io_size;
+    }
+    p.setValue(std::move(result));
+  };
+  if (slot) {
+    std::move(work).setLazyLocal(slot->signal()).start(std::move(cb));
+  }
+  else {
+    std::move(work).start(std::move(cb));
   }
 
-  co_return std::pair{result.first, io_size};
+  co_return std::pair{result.first, result.second};
 }
 
 template <typename T>
@@ -258,6 +280,8 @@ async_io_split(coro_io::ib_socket_t& ib_socket, Buffer&& raw_buffer,
 
   std::size_t block_size;
   uint32_t now_split_size = 0;
+  std::optional<async_simple::Future<std::pair<std::error_code, std::size_t>>>
+      future;
   for (auto& sge : sge_span) {
     for (std::size_t i = 0; i < sge.length; i += block_size) {
       block_size =
@@ -284,47 +308,45 @@ async_io_split(coro_io::ib_socket_t& ib_socket, Buffer&& raw_buffer,
         }
         else {
           std::tie(ec, len) = co_await async_send_impl(
-              ib_socket, split_sge_block, now_split_size);
+              ib_socket, split_sge_block, now_split_size, future);
         }
-
         io_completed_size += len;
         ELOG_TRACE << "has completed size:" << io_completed_size;
         if (ec) {
           co_return std::pair{ec, io_completed_size};
         }
 
-        if (len == 0 || len > now_split_size) [[unlikely]] {
-          ELOG_ERROR << "read size error, it shouldn't be:" << len;
-          co_return std::pair{std::make_error_code(std::errc::io_error),
-                              io_completed_size};
-        }
-
-        if (read_some) {
-          co_return std::pair{ec, io_completed_size};
-        }
-
-        if (len < now_split_size) [[unlikely]] {
-          reset_buffer(split_sge_block, len);
+        if constexpr (io == ib_socket_t::io_type::recv) {
+          if (read_some) {
+            co_return std::pair{ec, io_completed_size};
+          }
+          if (len < now_split_size) [[unlikely]] {
+            reset_buffer(split_sge_block, len);
+          }
+          else {
+            split_sge_block.clear();
+          }
+          now_split_size -= len;
         }
         else {
           split_sge_block.clear();
+          now_split_size = 0;
         }
-        now_split_size -= len;
       }
     }
   }
 
   std::error_code ec;
   std::size_t len = 0;
   while (now_split_size > 0) {
-    reset_buffer(split_sge_block, len);
     if constexpr (io == ib_socket_t::io_type::recv) {
+      reset_buffer(split_sge_block, len);
       std::tie(ec, len) =
           co_await async_recv_impl(ib_socket, split_sge_block, now_split_size);
     }
     else {
-      std::tie(ec, len) =
-          co_await async_send_impl(ib_socket, split_sge_block, now_split_size);
+      std::tie(ec, len) = co_await async_send_impl(ib_socket, split_sge_block,
+                                                   now_split_size, future);
     }
 
     io_completed_size += len;
@@ -333,17 +355,24 @@ async_io_split(coro_io::ib_socket_t& ib_socket, Buffer&& raw_buffer,
       co_return std::pair{ec, io_completed_size};
     }
 
-    if (len == 0 || len > now_split_size) [[unlikely]] {
-      ELOG_ERROR << ((io == ib_socket_t::io_type::recv) ? "recv " : "send")
-                 << "size error, it shouldn't be:" << len;
-      co_return std::pair{std::make_error_code(std::errc::io_error),
-                          io_completed_size};
+    if constexpr (io == ib_socket_t::io_type::recv) {
+      if (read_some) {
+        co_return std::pair{ec, io_completed_size};
+      }
+      now_split_size -= len;
     }
-
-    if (read_some) {
-      co_return std::pair{ec, io_completed_size};
+    else {
+      now_split_size = 0;
+    }
+  }
+  if constexpr (io == ib_socket_t::io_type::send) {
+    if (future) {
+      std::tie(ec, len) = co_await std::move(*future);
+      io_completed_size += len;
+      if (ec) [[unlikely]] {
+        co_return std::pair{ec, io_completed_size};
+      }
     }
-    now_split_size -= len;
   }
   co_return std::pair{ec, io_completed_size};
 }