[coro_rpc][feat and fix]Main support local ip (#965)

Author: qicosmos

include/ylt/coro_io/socket_wrapper.hpp

@@ -22,11 +22,22 @@
 namespace coro_io {
 struct socket_wrapper_t {
   // construct by listen tcp
-  socket_wrapper_t() {};
-  socket_wrapper_t(coro_io::ExecutorWrapper<> *executor)
-      : socket_(std::make_unique<asio::ip::tcp::socket>(
-            executor->get_asio_executor())),
-        executor_(executor) {
+  socket_wrapper_t(){};
+  socket_wrapper_t(coro_io::ExecutorWrapper<> *executor,
+                   const std::string &local_ip = "")
+      : executor_(executor) {
+    if (local_ip.empty()) {
+      socket_ = std::make_unique<asio::ip::tcp::socket>(
+          executor->get_asio_executor());
+    }
+    else {
+      asio::error_code ec;
+      socket_ = std::make_unique<asio::ip::tcp::socket>(
+          executor->get_asio_executor(),
+          asio::ip::tcp::endpoint(asio::ip::address::from_string(local_ip, ec),
+                                  0));
+    }
+
     init_client(true);
   };
   socket_wrapper_t(asio::ip::tcp::socket &&soc,
@@ -167,8 +178,8 @@ struct socket_wrapper_t {
     ssl_stream_ = std::make_unique<asio::ssl::stream<asio::ip::tcp::socket &>>(
         *socket_, ssl_ctx);
   }
-  std::unique_ptr<asio::ssl::stream<asio::ip::tcp::socket &>> &
-  ssl_stream() noexcept {
+  std::unique_ptr<asio::ssl::stream<asio::ip::tcp::socket &>>
+      &ssl_stream() noexcept {
     return ssl_stream_;
   }
   using tcp_socket_with_ssl_t = asio::ssl::stream<asio::ip::tcp::socket &>;

include/ylt/coro_rpc/impl/coro_rpc_client.hpp

@@ -196,6 +196,7 @@ class coro_rpc_client {
     std::chrono::milliseconds request_timeout_duration;
     std::string host;
     std::string port;
+    std::string local_ip;
     std::variant<tcp_config
 #ifdef YLT_ENABLE_SSL
                  ,
@@ -214,6 +215,7 @@ class coro_rpc_client {
           host(),
           port(),
           socket_config(tcp_config{}) {}
+    config(const std::string &loc_ip) : config() { local_ip = loc_ip; }
     config(config &&) = default;
     config(const config &) = default;
     config &operator=(const config &) = default;
@@ -240,16 +242,19 @@ class coro_rpc_client {
    */
   coro_rpc_client(
       coro_io::ExecutorWrapper<> *executor = coro_io::get_global_executor(),
-      config conf = config{})
-      : control_(
-            std::make_shared<control_t>(executor->get_asio_executor(), false)),
+      config conf = {})
+      : control_(std::make_shared<control_t>(executor->get_asio_executor(),
+                                             false, conf.local_ip)),
         timer_(std::make_unique<coro_io::period_timer>(
             executor->get_asio_executor())) {
     if (!init_config(config{})) [[unlikely]] {
       close();
     }
   }
 
+  coro_rpc_client(const std::string &local_ip)
+      : coro_rpc_client(coro_io::get_global_executor(), config(local_ip)) {}
+
   std::string_view get_host() const { return config_.host; }
 
   std::string_view get_port() const { return config_.port; }
@@ -855,11 +860,12 @@ class coro_rpc_client {
     std::unordered_map<uint32_t, handler_t> response_handler_table_;
     resp_body resp_buffer_;
     std::atomic<uint32_t> recving_cnt_ = 0;
-    control_t(asio::io_context::executor_type executor, bool is_timeout)
+    control_t(asio::io_context::executor_type executor, bool is_timeout,
+              const std::string &local_ip = "")
         : is_timeout_(is_timeout),
           has_closed_(false),
           executor_(executor),
-          socket_wrapper_(&executor_) {}
+          socket_wrapper_(&executor_, local_ip) {}
   };
 
   static void close_socket_async(
@@ -958,8 +964,11 @@ class coro_rpc_client {
           header, std::string_view{buffer, buffer + sizeof(buffer)});
       assert(!ec);
       if (ret.first) {
-        ELOG_ERROR << "read rpc head failed, error msg:" << ret.first.message()
-                   << ". close the socket.value=" << ret.first.value();
+        if (ret.first != asio::error::eof) {
+          ELOG_ERROR << "read rpc head failed, error msg:"
+                     << ret.first.message()
+                     << ". close the socket.value=" << ret.first.value();
+        }
         break;
       }
       auto iter = controller->response_handler_table_.find(header.seq_num);

include/ylt/standalone/cinatra/coro_http_server.hpp

@@ -670,7 +670,7 @@ class coro_http_server {
                            std::string_view head_msg) {
     auto conn = accept_impl(std::move(soc), true);
     conn->add_head(head_msg);
-    start_one(conn).via(conn->get_executor()).detach();
+    start_one(conn, true).via(conn->get_executor()).detach();
   }
 
  private:
@@ -759,6 +759,11 @@ class coro_http_server {
     co_await conn->start();
   }
 
+  async_simple::coro::Lazy<void> start_one(
+      std::shared_ptr<coro_http_connection> conn, bool has_shake) noexcept {
+    co_await conn->start(has_shake);
+  }
+
   void close_acceptor() {
     asio::dispatch(acceptor_.get_executor(), [this]() {
       asio::error_code ec;

src/coro_rpc/examples/base_examples/coro_rpc_lib/coro_rpc.cpp

@@ -1,5 +1,6 @@
 #include "coro_rpc.h"
 
+#include <asio/ip/host_name.hpp>
 #include <ylt/coro_io/load_balancer.hpp>
 #include <ylt/coro_rpc/coro_rpc_client.hpp>
 #include <ylt/coro_rpc/coro_rpc_server.hpp>
@@ -83,17 +84,37 @@ void stop_rpc_server(void *server) {
 }
 
 // rpc client
-void *create_client_pool(char *addr, int req_timeout_sec, bool enable_ib) {
+void *create_client_pool(char *addr, client_config conf) {
   std::vector<std::string_view> hosts{std::string_view(addr)};
   coro_io::client_pool<coro_rpc::coro_rpc_client>::pool_config pool_conf{};
 #ifdef YLT_ENABLE_IBV
-  if (enable_ib) {
+  if (conf.enable_ib) {
     coro_io::ibverbs_config ib_conf{};
     pool_conf.client_config.socket_config = ib_conf;
   }
 #endif
-  pool_conf.client_config.request_timeout_duration =
-      std::chrono::seconds{req_timeout_sec};
+  if (conf.connect_timeout_sec != 0) {
+    pool_conf.client_config.connect_timeout_duration =
+        std::chrono::seconds{conf.connect_timeout_sec};
+  }
+
+  if (conf.req_timeout_sec != 0) {
+    pool_conf.client_config.request_timeout_duration =
+        std::chrono::seconds{conf.req_timeout_sec};
+  }
+
+  if (conf.local_ip == nullptr) {
+    pool_conf.client_config.local_ip = "localhost";
+  }
+  else {
+    pool_conf.client_config.local_ip = conf.local_ip;
+  }
+
+  ELOG_INFO << "client config connect timeout seconds: "
+            << conf.connect_timeout_sec
+            << ", request timeout seconds: " << conf.req_timeout_sec
+            << ", local_ip: " << pool_conf.client_config.local_ip
+            << ", enable ibverbs: " << conf.enable_ib;
 
   auto ld = coro_io::load_balancer<coro_rpc::coro_rpc_client>::create(
       hosts, {pool_conf});
@@ -158,3 +179,22 @@ void init_rpc_log(char *log_filename, int log_level, uint64_t max_file_size,
 }
 
 void flush_rpc_log() { easylog::flush(); }
+
+char *get_first_local_ip() {
+  std::string local_ip = "localhost";
+  using asio::ip::tcp;
+  tcp::resolver resolver(coro_io::get_global_executor()->get_asio_executor());
+  tcp::resolver::query query(asio::ip::host_name(), "");
+  tcp::resolver::iterator iter = resolver.resolve(query);
+  tcp::resolver::iterator end;  // End marker.
+  while (iter != end) {
+    tcp::endpoint ep = *iter++;
+    auto addr = ep.address();
+    if (addr.is_v4()) {
+      local_ip = addr.to_string();
+      break;
+    }
+  }
+
+  return create_copy_cstr(local_ip);
+}

src/coro_rpc/examples/base_examples/coro_rpc_lib/coro_rpc.h

@@ -12,6 +12,13 @@ typedef struct {
   uint64_t len;
 } rpc_result;
 
+typedef struct {
+  int connect_timeout_sec;
+  int req_timeout_sec;
+  char *local_ip;
+  bool enable_ib;
+} client_config;
+
 // ip:port
 extern void load_service(void *ctx, uint64_t req_id);
 extern void *response_msg(void *ctx, char *msg, uint64_t size);
@@ -20,12 +27,10 @@ extern rpc_result wait_response_finish(void *p);
 extern void *start_rpc_server(char *addr, int parallel, bool enable_ib);
 extern void stop_rpc_server(void *server);
 
-extern void *create_client_pool(char *addr, int req_timeout_sec,
-                                bool enable_ib);
+extern void *create_client_pool(char *addr, client_config conf);
 extern void free_client_pool(void *pool);
 extern rpc_result load(void *pool, uint64_t req_id, char *dest,
                        uint64_t dest_len);
-
 /*
 enum log_level {
   NONE = 0,
@@ -44,6 +49,8 @@ extern void init_rpc_log(char *log_filename, int log_level,
                          bool async);
 extern void flush_rpc_log();
 
+extern char *get_first_local_ip();
+
 #ifdef __cplusplus
 }
 #endif

src/coro_rpc/examples/base_examples/go_example/test_rpc.go

@@ -9,7 +9,7 @@ import(
 /*
 #cgo CXXFLAGS: -std=c++20
 #cgo CFLAGS: -I../coro_rpc_lib
-#cgo LDFLAGS: -L./ -lcoro_rpc -lm -lstdc++ -libverbs
+#cgo LDFLAGS: -L./ -lcoro_rpc -lm -lstdc++
 #include "coro_rpc.h"
 */
 import "C"
@@ -54,7 +54,8 @@ func load_service(ctx unsafe.Pointer, req_id C.uint64_t) {
 
 func test_client(host string, len int) {
   peer := C.CString(host)
-  pool := C.create_client_pool(peer, C.int(30), C.bool(true))
+  conf := C.client_config{connect_timeout_sec:15, req_timeout_sec:30, local_ip:C.get_first_local_ip(), enable_ib:C.bool(false)}
+  pool := C.create_client_pool(peer, conf)
 
   outbuf := make([]byte, len)
   for i := 0; i < 3; i++ {

src/coro_rpc/tests/test_coro_rpc_client.cpp

@@ -17,6 +17,7 @@
 #include <async_simple/coro/SyncAwait.h>
 
 #include <asio/io_context.hpp>
+#include <asio/ip/host_name.hpp>
 #include <chrono>
 #include <cstddef>
 #include <memory>
@@ -170,6 +171,49 @@ TEST_CASE("testing client") {
   thd.join();
 }
 
+std::string get_first_local_ip() {
+  using asio::ip::tcp;
+  tcp::resolver resolver(coro_io::get_global_executor()->get_asio_executor());
+  tcp::resolver::query query(asio::ip::host_name(), "");
+  tcp::resolver::iterator iter = resolver.resolve(query);
+  tcp::resolver::iterator end;  // End marker.
+  while (iter != end) {
+    tcp::endpoint ep = *iter++;
+    auto addr = ep.address();
+    if (addr.is_v4()) {
+      return addr.to_string();
+    }
+  }
+
+  return "localhost";
+}
+
+TEST_CASE("testing client with local ip") {
+  coro_rpc_server server(1, 8901);
+  server.register_handler<hello>();
+  auto res = server.async_start();
+  REQUIRE_MESSAGE(!res.hasResult(), "server start failed");
+
+  std::string local_ip = get_first_local_ip();
+  ELOG_INFO << "local ip: " << local_ip;
+  coro_rpc_client client(local_ip);
+  auto ec = client.sync_connect("127.0.0.1", "8901");
+  REQUIRE_MESSAGE(!ec, ec.message());
+
+  auto ret = client.sync_call<hello>();
+  CHECK(ret.value() == "hello"s);
+
+  coro_rpc_server server1(1, 8902, local_ip);
+  server1.register_handler<hello>();
+  res = server1.async_start();
+  REQUIRE_MESSAGE(!res.hasResult(), "server start failed");
+  ec = client.sync_connect(local_ip, "8902");
+  REQUIRE_MESSAGE(!ec, ec.message());
+
+  ret = client.sync_call<hello>();
+  CHECK(ret.value() == "hello"s);
+}
+
 TEST_CASE("testing client with inject server") {
   g_action = {};
   std::string port = std::to_string(coro_rpc_server_port);

src/coro_rpc/tests/test_coro_rpc_server.cpp

@@ -557,7 +557,7 @@ TEST_CASE("testing coro rpc ssl subserver") {
   CHECK_MESSAGE(!result.net_err, result.net_err.message());
   result = syncAwait(cli.async_get("/index.html"));
   CHECK_MESSAGE(result.status == (int)coro_http::status_type::ok,
-                result.status);
+                result.net_err.message());
   CHECK_MESSAGE(result.resp_body == http_body, result.resp_body);
 }
 #endif