[coro_rpc][rdma]coro_rpc_rdma (#955)

Author: qicosmos

CMakeLists.txt

@@ -22,6 +22,7 @@ if(CMAKE_PROJECT_NAME STREQUAL "yaLanTingLibs") # if ylt is top-level project
         include_directories(src/include)
 
         include(cmake/find_openssl.cmake)
+        include(cmake/find_ibverbs.cmake)
         include(cmake/utils.cmake)
         include(cmake/build.cmake)
         include(cmake/develop.cmake)

cmake/config.cmake

@@ -11,6 +11,17 @@ if (YLT_ENABLE_SSL)
         target_link_libraries(yalantinglibs INTERFACE OpenSSL::SSL OpenSSL::Crypto)
     endif ()
 endif ()
+option(YLT_ENABLE_IBV "Enable ibverbs support" OFF)
+if (YLT_ENABLE_IBV)
+    message(STATUS "Enable ibverbs support")
+    if(CMAKE_PROJECT_NAME STREQUAL "yaLanTingLibs")
+        add_compile_definitions("YLT_ENABLE_IBV")
+        link_libraries(-libverbs)
+    else ()
+        target_compile_definitions(yalantinglibs INTERFACE "YLT_ENABLE_IBV")
+        target_link_libraries(yalantinglibs -libverbs)
+    endif ()
+endif ()
 
 option(YLT_ENABLE_PMR "Enable pmr support" OFF)
 message(STATUS "YLT_ENABLE_PMR: ${YLT_ENABLE_PMR}")

cmake/find_ibverbs.cmake

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

include/ylt/coro_io/coro_io.hpp

@@ -23,17 +23,18 @@
 
 #include <atomic>
 #include <cstddef>
+#include <cstdint>
 #include <memory>
+#include <ostream>
 #include <system_error>
 #include <thread>
 #include <type_traits>
 #include <utility>
 
 #include "asio/dispatch.hpp"
 #include "asio/io_context.hpp"
+#include "asio/ip/address.hpp"
 #include "async_simple/Signal.h"
-#include "async_simple/coro/FutureAwaiter.h"
-#include "async_simple/coro/SpinLock.h"
 #include "ylt/easylog.hpp"
 #include "ylt/util/type_traits.h"
 
@@ -49,8 +50,6 @@
 #include <asio/read_until.hpp>
 #include <asio/write.hpp>
 #include <asio/write_at.hpp>
-#include <chrono>
-#include <deque>
 
 #include "io_context_pool.hpp"
 #if __has_include("ylt/util/type_traits.h")
@@ -61,7 +60,6 @@
 #ifdef __linux__
 #include <sys/sendfile.h>
 #endif
-
 namespace coro_io {
 template <typename T>
 constexpr inline bool is_lazy_v =
@@ -136,6 +134,8 @@ class callback_awaitor_base {
     }
     void resume() const { obj->coro_.resume(); }
 
+    auto handler() const { return (std::size_t)obj; }
+
    private:
     Derived *obj;
   };
@@ -185,6 +185,30 @@ struct post_helper {
   Func func;
 };
 
+template <typename R, typename Func, typename Executor>
+struct dispatch_helper {
+  void operator()(auto handler) {
+    asio::dispatch(e, [this, handler]() {
+      try {
+        if constexpr (std::is_same_v<R, async_simple::Try<void>>) {
+          func();
+          handler.resume();
+        }
+        else {
+          auto r = func();
+          handler.set_value_then_resume(std::move(r));
+        }
+      } catch (const std::exception &e) {
+        R er;
+        er.setException(std::current_exception());
+        handler.set_value_then_resume(std::move(er));
+      }
+    });
+  }
+  Executor e;
+  Func func;
+};
+
 template <typename Func, typename Executor>
 inline async_simple::coro::Lazy<
     async_simple::Try<typename util::function_traits<Func>::return_type>>
@@ -205,6 +229,17 @@ post(Func func,
   return post(std::move(func), e->get_asio_executor());
 }
 
+template <typename Func, typename Executor>
+inline async_simple::coro::Lazy<
+    async_simple::Try<typename util::function_traits<Func>::return_type>>
+dispatch(Func func, Executor executor) {
+  using R =
+      async_simple::Try<typename util::function_traits<Func>::return_type>;
+  callback_awaitor<R> awaitor;
+  dispatch_helper<R, Func, Executor> helper{executor, std::move(func)};
+  co_return co_await awaitor.await_resume(helper);
+}
+
 namespace detail {
 
 template <typename T>
@@ -261,7 +296,7 @@ inline async_simple::coro::Lazy<ret_type> async_io(IO_func io_func,
               }
             }
           });
-      if (hasCanceled) {
+      if (hasCanceled) [[unlikely]] {
         asio::dispatch(executor, [handler]() {
           handler.set_value(
               std::make_error_code(std::errc::operation_canceled));
@@ -324,7 +359,8 @@ inline async_simple::coro::Lazy<std::error_code> async_accept(
     asio::ip::tcp::acceptor &acceptor, asio::ip::tcp::socket &socket) noexcept {
   return async_io<std::error_code>(
       [&](auto &&cb) {
-        acceptor.async_accept(socket, std::move(cb));
+        ELOG_INFO << "call asio acceptor.async_accept";
+        acceptor.async_accept(socket, cb);
       },
       acceptor);
 }
@@ -415,37 +451,56 @@ template <typename executor_t>
 inline async_simple::coro::Lazy<std::error_code> async_connect(
     executor_t *executor, asio::ip::tcp::socket &socket,
     const std::string &host, const std::string &port) noexcept {
-  asio::ip::tcp::resolver resolver(executor->get_asio_executor());
-  auto result = co_await async_io<
-      std::pair<std::error_code, asio::ip::tcp::resolver::iterator>>(
-      [&](auto &&cb) {
-        resolver.async_resolve(host, port, std::move(cb));
-      },
-      resolver);
-
-  if (result.first) {
-    co_return result.first;
+  std::error_code ec;
+  auto address = asio::ip::make_address(host, ec);
+  std::pair<std::error_code, asio::ip::tcp::resolver::iterator> result;
+  if (ec) {
+    asio::ip::tcp::resolver resolver(executor->get_asio_executor());
+    result = co_await async_io<
+        std::pair<std::error_code, asio::ip::tcp::resolver::iterator>>(
+        [&](auto &&cb) {
+          ELOG_INFO << "call asio resolver.async_resolve";
+          resolver.async_resolve(host, port, std::move(cb));
+          ELOG_INFO << "call asio resolver.async_resolve over, waiting cb";
+        },
+        resolver);
+    ELOG_INFO << "call asio resolver.async_resolve cbover";
+    if (result.first) {
+      co_return result.first;
+    }
+    co_return co_await async_io<std::error_code>(
+        [&](auto &&cb) {
+          ELOG_INFO << "call asio socket.async_connect";
+          asio::async_connect(socket, result.second, std::move(cb));
+        },
+        socket);
   }
-  result = co_await async_io<
-      std::pair<std::error_code, asio::ip::tcp::resolver::iterator>>(
+  else {
+    ELOG_INFO << "direct call without resolve";
+    uint16_t port_v;
+    auto result =
+        std::from_chars(port.data(), port.data() + port.size(), port_v);
+    if (result.ec != std::errc{}) {
+      co_return std::make_error_code(result.ec);
+    }
+    asio::ip::tcp::endpoint ep{address, port_v};
+    co_return co_await async_io<std::error_code>(
+        [&](auto &&cb) {
+          ELOG_INFO << "call asio socket.async_connect";
+          asio::async_connect(socket, std::span{&ep, 1}, std::move(cb));
+        },
+        socket);
+  }
+}
+
+template <typename EndPointSeq>
+inline async_simple::coro::Lazy<std::error_code> async_connect(
+    asio::ip::tcp::socket &socket, const EndPointSeq &endpoint) noexcept {
+  auto result = co_await async_io<std::error_code>(
       [&](auto &&cb) {
-        asio::async_connect(socket, result.second, std::move(cb));
+        asio::async_connect(socket, endpoint, std::move(cb));
       },
       socket);
-  co_return result.first;
-}
-
-template <typename executor_t, typename EndPointSeq>
-inline async_simple::coro::Lazy<
-    std::pair<std::error_code, asio::ip::tcp::endpoint>>
-async_connect(executor_t *executor, asio::ip::tcp::socket &socket,
-              const EndPointSeq &endpoint) noexcept {
-  auto result =
-      co_await async_io<std::pair<std::error_code, asio::ip::tcp::endpoint>>(
-          [&](auto &&cb) {
-            asio::async_connect(socket, endpoint, std::move(cb));
-          },
-          socket);
   co_return result;
 }
 
@@ -463,6 +518,20 @@ async_resolve(executor_t *executor, asio::ip::tcp::socket &socket,
       resolver);
 }
 
+template <typename executor_t>
+inline async_simple::coro::Lazy<
+    std::pair<std::error_code, asio::ip::tcp::resolver::iterator>>
+async_resolve(executor_t *executor, const std::string &host,
+              const std::string &port) noexcept {
+  asio::ip::tcp::resolver resolver(executor->get_asio_executor());
+  co_return co_await async_io<
+      std::pair<std::error_code, asio::ip::tcp::resolver::iterator>>(
+      [&](auto &&cb) {
+        resolver.async_resolve(host, port, std::move(cb));
+      },
+      resolver);
+}
+
 template <typename Socket>
 inline async_simple::coro::Lazy<void> async_close(Socket &socket) noexcept {
   callback_awaitor<void> awaitor;
@@ -479,13 +548,37 @@ inline async_simple::coro::Lazy<void> async_close(Socket &socket) noexcept {
 
 #if defined(YLT_ENABLE_SSL) || defined(CINATRA_ENABLE_SSL)
 inline async_simple::coro::Lazy<std::error_code> async_handshake(
-    auto &ssl_stream, asio::ssl::stream_base::handshake_type type) noexcept {
+    auto &&ssl_stream, asio::ssl::stream_base::handshake_type type) noexcept {
   return async_io<std::error_code>(
       [&, type](auto &&cb) {
         ssl_stream->async_handshake(type, std::move(cb));
       },
       *ssl_stream);
 }
+template <typename executor_t>
+inline async_simple::coro::Lazy<std::error_code> async_connect(
+    executor_t *executor, asio::ssl::stream<asio::ip::tcp::socket &> &socket,
+    const std::string &host, const std::string &port) noexcept {
+  auto ec = co_await async_connect(executor, socket, host, port);
+  if (ec) [[unlikely]] {
+    co_return ec;
+  }
+  ec = co_await coro_io::async_handshake(&socket,
+                                         asio::ssl::stream_base::client);
+  co_return ec;
+}
+template <typename EndPointSeq>
+inline async_simple::coro::Lazy<std::error_code> async_connect(
+    asio::ssl::stream<asio::ip::tcp::socket &> &socket,
+    const EndPointSeq &endpoint) noexcept {
+  auto ec = co_await async_connect(socket.next_layer(), endpoint);
+  if (ec) [[unlikely]] {
+    co_return ec;
+  }
+  ec = co_await coro_io::async_handshake(&socket,
+                                         asio::ssl::stream_base::client);
+  co_return ec;
+}
 #endif
 class period_timer : public asio::steady_timer {
  public:
@@ -689,34 +782,15 @@ async_sendfile(asio::ip::tcp::socket &socket, int fd, off_t offset,
 }
 #endif
 
-struct socket_wrapper_t {
-  socket_wrapper_t(asio::ip::tcp::socket &&soc,
-                   coro_io::ExecutorWrapper<> *executor)
-      : socket_(std::make_unique<asio::ip::tcp::socket>(std::move(soc))),
-        executor_(executor) {}
-
- private:
-  std::unique_ptr<asio::ip::tcp::socket> socket_;
-  coro_io::ExecutorWrapper<> *executor_;
-#ifdef YLT_ENABLE_SSL
-  std::unique_ptr<asio::ssl::stream<asio::ip::tcp::socket &>> ssl_stream_;
-#endif
- public:
-  bool use_ssl() const noexcept {
-#ifdef YLT_ENABLE_SSL
-    return ssl_stream_ != nullptr;
-#else
-    return false;
-#endif
-  }
-  auto get_executor() const noexcept { return executor_; }
-  std::unique_ptr<asio::ip::tcp::socket> &socket() noexcept { return socket_; }
-#ifdef YLT_ENABLE_SSL
-  std::unique_ptr<asio::ssl::stream<asio::ip::tcp::socket &>>
-      &ssl_stream() noexcept {
-    return ssl_stream_;
-  }
-#endif
+enum protocal { tcp, tcp_with_ssl, rdma };
+struct endpoint {
+  asio::ip::address address;
+  asio::ip::port_type port;
+  protocal proto;
 };
 
+inline std::ostream &operator<<(std::ostream &stream, const endpoint &ep) {
+  return stream << ep.address.to_string() << ":" << ep.port;
+}
+
 }  // namespace coro_io