[coro_io] fix ib_socket log & ib_buffer_pool (#962)

fix buffer pool

Author: poor-circle

include/ylt/coro_io/coro_io.hpp

@@ -782,10 +782,10 @@ async_sendfile(asio::ip::tcp::socket &socket, int fd, off_t offset,
 }
 #endif
 
-enum protocal { tcp, tcp_with_ssl, rdma };
 struct endpoint {
+  enum protocal { tcp, rdma };
   asio::ip::address address;
-  asio::ip::port_type port;
+  uint32_t port;
   protocal proto;
 };
 

include/ylt/coro_io/ibverbs/ib_buffer.hpp

@@ -55,10 +55,11 @@ struct ib_buffer_t {
                                            IBV_ACCESS_REMOTE_READ |
                                            IBV_ACCESS_REMOTE_WRITE) {
     auto mr = ibv_reg_mr(dev->pd(), ptr, size, ib_flags);
-    ELOG_INFO << "ibv_reg_mr regist: " << mr << " with pd:" << dev->pd();
+    ELOG_TRACE << "ibv_reg_mr regist: " << mr << " with pd:" << dev->pd();
     if (mr != nullptr) [[unlikely]] {
-      ELOG_INFO << "regist sge.lkey: " << mr->lkey << ", sge.addr: " << mr->addr
-                << ", sge.length: " << mr->length;
+      ELOG_TRACE << "regist sge.lkey: " << mr->lkey
+                 << ", sge.addr: " << mr->addr
+                 << ", sge.length: " << mr->length;
       return ib_buffer_t{mr, std::move(dev)};
     }
     else {
@@ -144,7 +145,6 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
   }
   void collect_free_inner(ib_buffer_t buffer) {
     if (buffer) {
-      ELOG_TRACE << "collecting:" << (void*)buffer->addr;
       if (free_buffers_.size() * pool_config_.buffer_size <
           pool_config_.max_memory_usage) {
         ELOG_TRACE << "collect free buffer{data:" << buffer->addr << ",len"
@@ -158,9 +158,6 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
                    << "} wont be collect";
       }
     }
-    else {
-      ELOG_TRACE << "collecting nullptr";
-    }
     return;
   };
 
@@ -173,9 +170,6 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
       buffer.change_owner(weak_from_this());
       collect_free_inner(std::move(buffer));
     }
-    else {
-      ELOG_TRACE << "collecting nullptr";
-    }
     return;
   };
   ib_buffer_t get_buffer() {
@@ -188,11 +182,11 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
           pool_config_.buffer_size +
               total_memory_.load(std::memory_order_acquire)) [[unlikely]] {
         ELOG_WARN << "Memory out of pool limit";
-        // return std::move(buffer);
+        return std::move(buffer);
       }
       auto ptr = malloc(pool_config_.buffer_size);
       if (ptr == nullptr) {
-        ELOG_TRACE << "Allocate failed.";
+        ELOG_ERROR << "ib_buffer_pool allocate failed.";
         throw std::bad_alloc();
       }
       buffer =
@@ -284,9 +278,7 @@ inline ib_buffer_t::~ib_buffer_t() {
 }
 
 inline void ib_buffer_t::collect() && {
-  ELOG_TRACE << "collecting buffer";
   if (auto pool = owner_pool_.lock(); pool) {
-    ELOG_TRACE << "has pool. collecting buffer continue";
     pool->collect_free_inner(std::move(*this));
   }
 }

include/ylt/coro_io/ibverbs/ib_device.hpp

@@ -8,6 +8,7 @@
 #include <string>
 #include <system_error>
 
+#include "asio/ip/address.hpp"
 #include "ylt/easylog.hpp"
 
 namespace coro_io {
@@ -101,14 +102,25 @@ struct ib_deleter {
     }
   }
   void operator()(ibv_mr* ptr) const noexcept {
-    ELOG_INFO << "ibv_reg_mr unregist: " << ptr;
     if (auto ret = ibv_dereg_mr(ptr); ret) [[unlikely]] {
       ELOG_ERROR << "ibv_dereg_mr failed: "
                  << std::make_error_code(std::errc{ret}).message();
     }
   }
 };
 
+namespace detail {
+inline std::string gid_to_string(uint8_t (&a)[16]) noexcept {
+  std::string ret;
+  ret.resize(40);
+  sprintf(ret.data(),
+          "%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%"
+          "02x%02x",
+          a[0], a[1], a[2], a[3], a[4], a[5], a[6], a[7], a[8], a[9], a[10],
+          a[11], a[12], a[13], a[14], a[15]);
+  return ret;
+}
+
 inline std::string mtu_str(ibv_mtu mtu) {
   std::string str;
   switch (mtu) {
@@ -132,7 +144,7 @@ inline std::string mtu_str(ibv_mtu mtu) {
   }
   return str;
 }
-
+}  // namespace detail
 class ib_device_t {
  public:
   ib_device_t(const ib_config_t& conf) {
@@ -150,13 +162,11 @@ class ib_device_t {
       throw std::system_error(ec);
     }
 
-    ELOG_INFO << "Active MTU: " << mtu_str(attr_.active_mtu) << ", "
-              << "Max MTU: " << mtu_str(attr_.max_mtu);
+    ELOG_TRACE << "Active MTU: " << detail::mtu_str(attr_.active_mtu) << ", "
+               << "Max MTU: " << detail::mtu_str(attr_.max_mtu);
 
     find_best_gid_index();
 
-    ELOG_INFO << "IBDevice " << name_ << ", best gid index " << gid_index_;
-
     if (gid_index_ >= 0) {
       if (auto ec = ibv_query_gid(ctx_.get(), conf.port, gid_index_, &gid_);
           ec) {
@@ -166,6 +176,15 @@ class ib_device_t {
                    << ", error msg: " << err_code.message();
         throw std::system_error(err_code);
       }
+      std::error_code err_code;
+      gid_address_ =
+          asio::ip::make_address(detail::gid_to_string(gid_.raw), err_code);
+      if (err_code) {
+        ELOG_ERROR << "IBDevice failed to convert gid to ip address of device "
+                   << name_ << " by gid_index:" << gid_index_
+                   << ", error msg: " << err_code.message();
+        throw std::system_error(err_code);
+      }
     }
     else {
       ELOG_ERROR << "gid index should greater than zero, now is: "
@@ -189,6 +208,7 @@ class ib_device_t {
   ibv_pd* pd() const noexcept { return pd_.get(); }
   int gid_index() const noexcept { return gid_index_; }
   const ibv_gid& gid() const noexcept { return gid_; }
+  asio::ip::address gid_address() const noexcept { return gid_address_; }
   const ibv_port_attr& attr() const noexcept { return attr_; }
 
  private:
@@ -225,6 +245,7 @@ class ib_device_t {
   std::unique_ptr<ibv_context, ib_deleter> ctx_;
   ibv_port_attr attr_;
   ibv_gid gid_;
+  asio::ip::address gid_address_;
   ibv_device* device_;
   int gid_index_;
 

include/ylt/coro_io/ibverbs/ib_io.hpp

@@ -43,7 +43,6 @@ inline async_simple::coro::Lazy<std::error_code> async_accept(
   if (ec) [[unlikely]] {
     co_return std::move(ec);
   }
-  // TODO: SSL?
   auto ret = co_await ib_socket.accept(std::move(soc));
   ELOGV(INFO, "accept over:%s", ret.message().data());
   co_return ret;