fix rdma buffer pool (#958)

* fix buffer pool

* fix max send sge

Author: poor-circle

include/ylt/coro_io/ibverbs/ib_buffer.hpp

@@ -145,14 +145,15 @@ 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_.max_buffer_count) {
+      if (free_buffers_.size() * pool_config_.buffer_size <
+          pool_config_.max_memory_usage) {
         ELOG_TRACE << "collect free buffer{data:" << buffer->addr << ",len"
                    << buffer->length << "} enqueue";
         enqueue(std::move(buffer));
       }
       else {
         ELOG_TRACE << "out of max connection limit <<"
-                   << pool_config_.max_buffer_count
+                   << pool_config_.max_memory_usage
                    << "buffer{data:" << buffer->addr << ",len" << buffer->length
                    << "} wont be collect";
       }
@@ -164,6 +165,9 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
   };
 
  public:
+  std::size_t max_buffer_size() const noexcept {
+    return this->pool_config_.buffer_size;
+  }
   void collect_free(ib_buffer_t buffer) {
     if (buffer) {
       buffer.change_owner(weak_from_this());
@@ -183,7 +187,7 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
       if (pool_config_.max_memory_usage <
           pool_config_.buffer_size +
               total_memory_.load(std::memory_order_acquire)) [[unlikely]] {
-        ELOG_TRACE << "Memory out of pool limit";
+        ELOG_WARN << "Memory out of pool limit";
         // return std::move(buffer);
       }
       auto ptr = malloc(pool_config_.buffer_size);
@@ -212,9 +216,8 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
     return buffer;
   }
   struct config_t {
-    size_t buffer_size = 8 * 1024 * 1024;  // 16MB
-    size_t max_buffer_count = 64;
-    size_t max_memory_usage = 512 * 1024 * 1024;  // 512MB
+    size_t buffer_size = 4 * 1024 * 1024;                   // 4MB
+    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};
   };

include/ylt/coro_io/ibverbs/ib_io.hpp

@@ -143,6 +143,10 @@ async_simple::coro::
   std::vector<ib_buffer_t> tmp_buffers;
   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}};
+  }
 
   auto len = copy(sge_list, socket_buffer);
   assert(len == io_size);

include/ylt/coro_io/ibverbs/ib_socket.hpp

@@ -53,6 +53,10 @@ struct buffer_queue {
   std::error_code post_recv_real(ibv_sge buffer, shared_state_t* state);
   std::error_code push(ib_buffer_t buf, shared_state_t* state) {
     ELOG_TRACE << "push. now buffer size:" << size();
+    if (buf->length == 0) {
+      ELOG_INFO << "Out of ib_buffer_pool limit." << size();
+      return std::make_error_code(std::errc::no_buffer_space);
+    }
     auto ec = post_recv_real(buf.subview(), state);
     if (!ec) {
       front = (front + 1) % queue.size();
@@ -241,14 +245,12 @@ inline std::error_code buffer_queue::post_recv_real(ibv_sge buffer,
 #ifdef YLT_ENABLE_IBV
 struct ibverbs_config {
   uint32_t cq_size = 1024;
-  uint32_t request_buffer_size = 8 * 1024 * 1024;
+  uint32_t request_buffer_size = 4 * 1024 * 1024;
   uint32_t recv_buffer_cnt = 2;
-  std::chrono::milliseconds tcp_handshake_timeout =
-      std::chrono::milliseconds{1000};
   ibv_qp_type qp_type = IBV_QPT_RC;
-  ibv_qp_cap cap = {.max_send_wr = 8,
-                    .max_recv_wr = 8,
-                    .max_send_sge = 6,
+  ibv_qp_cap cap = {.max_send_wr = 32,
+                    .max_recv_wr = 32,
+                    .max_send_sge = 1,
                     .max_recv_sge = 1,
                     .max_inline_data = 0};
   std::shared_ptr<coro_io::ib_device_t> device;
@@ -388,6 +390,11 @@ class ib_socket_t {
     buffer_size_ = std::min<uint32_t>(peer_info.buffer_size,
                                       get_config().request_buffer_size);
     ELOGV(INFO, "Final buffer size = %d", buffer_size_);
+    if (buffer_size_ > buffer_pool()->max_buffer_size()) {
+      ELOGV(ERROR, "Buffer size larger than buffer_pool limit: %d",
+            buffer_size_);
+      co_return std::make_error_code(std::errc::no_buffer_space);
+    }
     try {
       init_qp();
       modify_qp_to_init();
@@ -492,6 +499,11 @@ class ib_socket_t {
       buffer_size_ = std::min<uint32_t>(peer_info.buffer_size,
                                         get_config().request_buffer_size);
       ELOGV(INFO, "Final buffer size = %d", buffer_size_);
+      if (buffer_size_ > buffer_pool()->max_buffer_size()) {
+        ELOGV(ERROR, "Buffer size larger than buffer_pool limit: %d",
+              buffer_size_);
+        co_return std::make_error_code(std::errc::no_buffer_space);
+      }
       modify_qp_to_rtr(peer_info.qp_num, peer_info.lid,
                        (uint8_t*)peer_info.gid);
 

src/coro_io/tests/ibverbs/test_io.cpp

@@ -29,8 +29,8 @@
 #include "ylt/struct_json/json_reader.h"
 #include "ylt/struct_json/json_writer.h"
 struct config_t {
-  std::size_t buffer_size = 8 * 1024 * 1024;
-  std::size_t request_size = 8 * 1024 * 1024;
+  std::size_t buffer_size = 4 * 1024 * 1024;
+  std::size_t request_size = 4 * 1024 * 1024;
   int concurrency = 2;
   int test_type = 0;
   int enable_log = 0;

src/coro_io/tests/ibverbs/test_io2.cpp

@@ -25,7 +25,7 @@
 #include "ylt/easylog/record.hpp"
 #include "ylt/struct_pack/util.h"
 
-std::size_t buffer_size = 8 * 1024 * 1024;
+std::size_t buffer_size = 4 * 1024 * 1024;
 int concurrency = 10;
 std::atomic<int> port;