[coro_io][ibverbs] save memory usage (#964)

save memory usage

Author: poor-circle

include/ylt/coro_io/ibverbs/ib_socket.hpp

@@ -180,9 +180,8 @@ struct ib_socket_shared_state_t {
         if (wc.wr_id == 0) {  // recv
           if (recv_cb_) {
             assert(recv_result.empty());
-            bool could_insert = (recv_queue_.size() <= recv_buffer_cnt_);
             recv_buf_ = recv_queue_.pop();
-            if (could_insert) {
+            if (recv_queue_.empty()) {
               if (recv_queue_.push(ib_buffer_pool_->get_buffer(), this)) {
                 std::error_code ec;
                 close(ec);
@@ -191,13 +190,13 @@ struct ib_socket_shared_state_t {
             resume(std::pair{ec, (std::size_t)wc.byte_len}, recv_cb_);
           }
           else {
-            if (!recv_queue_.full()) {
+            recv_result.push(std::pair{ec, (std::size_t)wc.byte_len});
+            if (!recv_queue_.full() && recv_result.size()==recv_queue_.size()) {
               if (recv_queue_.push(ib_buffer_pool_->get_buffer(), this)) {
                 std::error_code ec;
                 close(ec);
               }
             }
-            recv_result.push(std::pair{ec, (std::size_t)wc.byte_len});
           }
         }
         else {

src/coro_io/tests/ibverbs/test_io.cpp

@@ -1,6 +1,7 @@
 #include <atomic>
 #include <chrono>
 #include <cmath>
+#include <cstdint>
 #include <ostream>
 #include <string>
 #include <string_view>
@@ -46,6 +47,8 @@ config_t config;
 
 std::atomic<uint64_t> cnt[2];
 
+std::atomic<uint64_t> connect_cnt;
+
 using namespace std::chrono_literals;
 using namespace std::literals;
 async_simple::coro::Lazy<std::error_code> echo_connect(
@@ -185,12 +188,15 @@ async_simple::coro::Lazy<std::error_code> echo_accept() {
 
     ELOG_INFO << "start new connection";
     auto executor = soc.get_executor();
+    ++connect_cnt;
     if (config.test_type == 0) {
       echo_connect(std::move(soc)).start([](auto &&) {
+        --connect_cnt;
       });
     }
     else {
       echo_connect_read_some(std::move(soc)).start([](auto &&) {
+        --connect_cnt;
       });
     }
   }
@@ -297,12 +303,15 @@ async_simple::coro::Lazy<std::error_code> echo_connect() {
   s.resize(config.request_size, 'A');
   uint64_t &sz = *(uint64_t *)s.data();
   sz = config.request_size - 8;
+  ++connect_cnt;
   if (config.test_type == 0) {
-    co_return co_await echo_client(soc, s);
+    ec = co_await echo_client(soc, s);
   }
   else {
-    co_return co_await echo_client_read_some(soc, s);
+    ec = co_await echo_client_read_some(soc, s);
   }
+  --connect_cnt;
+  co_return ec;
 }
 
 int main() {
@@ -345,7 +354,7 @@ int main() {
   for (int i = 0; i < config.test_time; ++i) {
     std::this_thread::sleep_for(std::chrono::seconds{1});
     auto c = cnt_p->exchange(0);
-    std::cout << "Throughput:" << 8.0 * c / 1000'000'000 << " Gb/s"
+    std::cout << "Throughput:" << 8.0 * c / 1000'000'000 << " Gb/s, alive connection:" << connect_cnt
               << std::endl;
   }
   return 0;