[coro_io] client pool support limit max live time of client (#1018)

* [coro_io] client pool support limit max live time of client

* fix

Author: poor-circle

include/ylt/coro_io/client_pool.hpp

@@ -307,6 +307,15 @@ class client_pool : public std::enable_shared_from_this<
   }
 
   void collect_free_client(std::unique_ptr<client_t> client) {
+    if (pool_config_.max_connection_life_time < std::chrono::seconds::max()) {
+      auto tp = client->get_create_time_point();
+      if (std::chrono::steady_clock::now() - tp >
+          pool_config_.max_connection_life_time) [[unlikely]] {
+        ELOG_INFO << "client{" << client.get()
+                  << "} live too long, we won't collect it";
+        return;
+      }
+    }
     if (!client->has_closed()) {
       if (free_clients_.size() < pool_config_.max_connection) {
         ELOG_TRACE << "collect free client{" << client.get() << "} enqueue";
@@ -361,6 +370,7 @@ class client_pool : public std::enable_shared_from_this<
         30000}; /* zero means wont detect */
     typename client_t::config client_config;
     std::chrono::seconds dns_cache_update_duration{5 * 60};  // 5mins
+    std::chrono::seconds max_connection_life_time = std::chrono::seconds::max();
   };
 
  private:

include/ylt/coro_io/ibverbs/ib_buffer.hpp

@@ -164,12 +164,12 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
         std::size_t clear_cnt = self->free_buffers_.clear_old(1000);
         self->modify_memory_usage(-1 * (ssize_t)clear_cnt *
                                   (ssize_t)self->buffer_size());
-        ELOG_WARN << "finish ib_buffer timeout free of pool{" << self.get()
-                  << "}, now ib_buffer cnt: " << self->free_buffers_.size()
-                  << " mem usage:"
-                  << (int64_t)(std::round(self->memory_usage() /
-                                          (1.0 * 1024 * 1024)))
-                  << " MB";
+        ELOG_TRACE << "finish ib_buffer timeout free of pool{" << self.get()
+                   << "}, now ib_buffer cnt: " << self->free_buffers_.size()
+                   << " mem usage:"
+                   << (int64_t)(std::round(self->memory_usage() /
+                                           (1.0 * 1024 * 1024)))
+                   << " MB";
         if (clear_cnt != 0) {
           try {
             co_await async_simple::coro::Yield{};

include/ylt/coro_io/ibverbs/ib_socket.hpp

@@ -293,7 +293,7 @@ struct ib_socket_shared_state_t
       }
       ec = make_error_code(wc.status);
       if (wc.status != IBV_WC_SUCCESS) {
-        ELOG_WARN << "rdma failed with error code: " << wc.status;
+        ELOG_WARN << "rdma failed with error: " << ec.message();
       }
       else {
         ELOG_TRACE << "rdma op success, id:" << (callback_t*)wc.wr_id

include/ylt/coro_rpc/impl/coro_rpc_client.hpp

@@ -288,6 +288,7 @@ class coro_rpc_client {
   }
 #endif
   [[nodiscard]] bool init_config(const config &conf) {
+    create_tp_ = std::chrono::steady_clock::now();
     config_ = conf;
     control_->socket_wrapper_.set_local_ip(config_.local_ip);
     return std::visit(
@@ -297,6 +298,7 @@ class coro_rpc_client {
         conf.socket_config);
   };
 
+  auto get_create_time_point() const noexcept { return create_tp_; }
   /*!
    * Check the client closed or not
    *
@@ -1291,5 +1293,6 @@ class coro_rpc_client {
   asio::ssl::context ssl_ctx_{asio::ssl::context::sslv23};
   bool ssl_init_ret_ = true;
 #endif
+  std::chrono::time_point<std::chrono::steady_clock> create_tp_;
 };
 }  // namespace coro_rpc

include/ylt/standalone/cinatra/coro_http_client.hpp

@@ -2,6 +2,7 @@
 #include <atomic>
 #include <cassert>
 #include <charconv>
+#include <chrono>
 #include <cstddef>
 #include <filesystem>
 #include <fstream>
@@ -142,7 +143,8 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
         timer_(&executor_wrapper_),
         socket_(std::make_shared<socket_t>(executor)),
         head_buf_(socket_->head_buf_),
-        chunked_buf_(socket_->chunked_buf_) {}
+        chunked_buf_(socket_->chunked_buf_),
+        create_tp_(std::chrono::steady_clock::now()) {}
 
   coro_http_client(
       coro_io::ExecutorWrapper<> *executor = coro_io::get_global_executor())
@@ -182,6 +184,10 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
 
   ~coro_http_client() { close(); }
 
+  auto get_create_time_point() const noexcept {
+    return std::chrono::steady_clock::now();
+  }
+
   void close() {
     if (socket_ == nullptr || socket_->has_closed_)
       return;
@@ -2463,6 +2469,7 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
       std::chrono::seconds(30);
   std::chrono::steady_clock::duration req_timeout_duration_ =
       std::chrono::seconds(60);
+  std::chrono::steady_clock::time_point create_tp_;
   bool enable_tcp_no_delay_ = true;
   std::string resp_chunk_str_;
   std::span<char> out_buf_;

src/coro_io/tests/test_client_pool.cpp

@@ -344,4 +344,19 @@ TEST_CASE("test client pools dns don't refresh") {
     CHECK(eps.get() == eps_init.get());
     CHECK(eps->empty());
   }());
+}
+
+TEST_CASE("test client pools client pool") {
+  async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
+    auto pool = coro_io::client_pool<coro_http::coro_http_client>::create(
+        "http://www.baidu.com",
+        coro_io::client_pool<coro_http::coro_http_client>::pool_config{
+            .max_connection_life_time = 0s});
+    co_await pool->send_request(
+        [](coro_http::coro_http_client &cli) -> Lazy<void> {
+          cli.close();
+          co_return;
+        });
+    CHECK(pool->free_client_count() == 0);
+  }());
 }