feat(http,oss): expose max idle timeout for pooled connections

ecosystem/oss.cpp

@@ -547,6 +547,7 @@ OssClient::OssClient(const ClientOptions& options, Authenticator* authenticator)
 
   m_client = photon::net::http::new_http_client();
   m_client->timeout(m_oss_options.request_timeout_us);
+  m_client->set_connection_idle_expiration(m_oss_options.connection_idle_expiration_us);
   m_client->set_user_agent(m_oss_options.user_agent);
   if (!m_oss_options.proxy.empty()) m_client->set_proxy(m_oss_options.proxy);
 

ecosystem/oss.h

@@ -59,6 +59,9 @@ struct ClientOptions {
   // the virtual-hosted style http(s)://bucket.endpoint/object
   bool path_style = false;
   uint64_t request_timeout_us = 60ull * 1000 * 1000;
+  // Max idle time in microseconds before the underlying HTTP client closes a
+  // pooled connection; -1ULL keeps it until the peer closes it.
+  uint64_t connection_idle_expiration_us = -1ULL;
   int retry_times = 2;
 
   // When the request timeouts or encounters 5xx error, we will

net/http/client.cpp

@@ -189,7 +189,7 @@ class ClientImpl : public Client {
     }
 
     IDialer* make_dialer() {   // on the current vCPU, for this client
-        return new_http_dialer(m_tls_ctx, m_bind_ips);
+        return new_http_dialer(m_tls_ctx, m_bind_ips, m_connection_idle_expiration);
     }
 
     IDialer* acquire_dialer() {

net/http/client.h

@@ -195,6 +195,14 @@ class Client : public Object {
     void timeout_ms(uint64_t tmo) { timeout(tmo * 1000ULL); }
     void timeout_s(uint64_t tmo) { timeout(tmo * 1000ULL * 1000ULL); }
 
+    // Max idle time in microseconds before a pooled connection is closed.
+    // -1ULL (default) keeps idle connections until the peer closes them; 0 is
+    // treated as -1ULL. Captured by each vCPU's built-in dialer on its first
+    // dial; later changes do not update an existing dialer. No effect on a
+    // dialer set by set_dialer().
+    void set_connection_idle_expiration(uint64_t timeout_us) { m_connection_idle_expiration = timeout_us; }
+    uint64_t connection_idle_expiration() const { return m_connection_idle_expiration; }
+
     // Inject a dialer to take over connection establishment (and pooling, if
     // any), replacing the built-in per-vCPU pooled dialer. Not owned; must
     // outlive the client, and must be safe for concurrent use across vCPUs.
@@ -239,6 +247,7 @@ class Client : public Object {
     ProxyAuthenticator m_proxy_authenticator;
     std::string m_user_agent;
     uint64_t m_timeout = -1ULL;
+    uint64_t m_connection_idle_expiration = -1ULL;
     bool m_proxy = false;
     std::vector<IPAddr> m_bind_ips;
     IDialer* m_dialer = nullptr;

net/http/dialer.cpp

@@ -554,7 +554,8 @@ static void add_layer(std::unique_ptr<IDialer>& chain, Args&&... args) {
 }
 
 IDialer* new_http_dialer(TLSContext* context,
-                         const std::vector<IPAddr>& bind_ips) {
+                         const std::vector<IPAddr>& bind_ips,
+                         uint64_t expiration) {
     std::unique_ptr<TLSContext> owned_context;
     if (!context) {
         owned_context.reset(new_tls_context(nullptr, nullptr, nullptr));
@@ -572,7 +573,7 @@ IDialer* new_http_dialer(TLSContext* context,
                              TLSLayer::ORIGIN);
     else
         add_layer<TLSDialer>(chain, context, TLSLayer::ORIGIN);
-    add_layer<PoolDialer>(chain, -1ULL);
+    add_layer<PoolDialer>(chain, expiration);
     return chain.release();
 }
 

net/http/dialer.h

@@ -100,9 +100,12 @@ IDialer* new_pool_dialer(IDialer* underlay, bool ownership = false,
 // Build the standard single-vCPU stack. If context is null, a default context
 // is created and owned by the returned stack. Otherwise the caller owns the
 // context and must keep it alive until the returned stack and every stream
-// checked out from its outer pool are destroyed.
+// checked out from its outer pool are destroyed. `expiration` is the max idle
+// time in microseconds of the pooled connections; -1ULL never expires them and
+// 0 is treated as -1ULL.
 IDialer* new_http_dialer(TLSContext* context = nullptr,
-                         const std::vector<IPAddr>& bind_ips = {});
+                         const std::vector<IPAddr>& bind_ips = {},
+                         uint64_t expiration = -1ULL);
 
 // Make a cross-vCPU dispatcher. Its factory is invoked lazily on each vCPU and
 // may therefore run concurrently; its captured state must remain alive and be

net/http/test/client_function_test.cpp

@@ -134,6 +134,84 @@ TEST(http_client, response_allocation_failure_closes_connection) {
     EXPECT_EQ(2, connections); // the abandoned response cannot poison pool reuse
 }
 
+TEST(http_client, connection_idle_expiration_expires_pooled_connection) {
+    std::unique_ptr<ISocketServer> tcp(new_tcp_socket_server());
+    ASSERT_EQ(0, tcp->bind_v4localhost());
+    ASSERT_EQ(0, tcp->listen());
+    std::unique_ptr<HTTPServer> server(new_http_server());
+    server->add_handler(new SimpleHandler, true, "/simple");
+    auto handler = server->get_connection_handler();
+    int connections = 0;
+    auto counted = [&](ISocketStream* stream) {
+        ++connections;
+        return handler(stream);
+    };
+    tcp->set_handler(counted);
+    ASSERT_EQ(0, tcp->start_loop());
+
+    std::unique_ptr<Client> client(new_http_client());
+    client->timeout(5'000'000);
+    client->set_connection_idle_expiration(300'000);
+    EXPECT_EQ(300'000, client->connection_idle_expiration());
+    auto target = std::string("http://127.0.0.1:") +
+        std::to_string(tcp->getsockname().port) + "/simple";
+    auto request = [&] {
+        Client::OperationOnStack<1024> op(client.get(), Verb::GET, target);
+        ASSERT_EQ(0, op.req.headers.content_length(0));
+        op.retry = 0;
+        ASSERT_EQ(0, op.call());
+        EXPECT_EQ(200, op.resp.status_code());
+        char buf[64];
+        while (op.resp.read(buf, sizeof(buf)) > 0) {}  // drain for pool reuse
+    };
+
+    request();
+    EXPECT_EQ(1, connections);
+    request();
+    EXPECT_EQ(1, connections); // reused while idle within the timeout
+    photon::thread_usleep(4 * 300'000);
+    request();
+    EXPECT_EQ(2, connections); // the expired connection is not reused
+}
+
+TEST(http_client, zero_connection_idle_expiration_never_expires) {
+    std::unique_ptr<ISocketServer> tcp(new_tcp_socket_server());
+    ASSERT_EQ(0, tcp->bind_v4localhost());
+    ASSERT_EQ(0, tcp->listen());
+    std::unique_ptr<HTTPServer> server(new_http_server());
+    server->add_handler(new SimpleHandler, true, "/simple");
+    auto handler = server->get_connection_handler();
+    int connections = 0;
+    auto counted = [&](ISocketStream* stream) {
+        ++connections;
+        return handler(stream);
+    };
+    tcp->set_handler(counted);
+    ASSERT_EQ(0, tcp->start_loop());
+
+    std::unique_ptr<Client> client(new_http_client());
+    client->timeout(5'000'000);
+    client->set_connection_idle_expiration(0);  // rejected as -1ULL, see client.h
+    auto target = std::string("http://127.0.0.1:") +
+        std::to_string(tcp->getsockname().port) + "/simple";
+    auto request = [&] {
+        Client::OperationOnStack<1024> op(client.get(), Verb::GET, target);
+        ASSERT_EQ(0, op.req.headers.content_length(0));
+        op.retry = 0;
+        ASSERT_EQ(0, op.call());
+        EXPECT_EQ(200, op.resp.status_code());
+        char buf[64];
+        while (op.resp.read(buf, sizeof(buf)) > 0) {}  // drain for pool reuse
+    };
+
+    request();
+    EXPECT_EQ(1, connections);
+    photon::thread_usleep(500'000);
+    EXPECT_EQ(1, connections); // still reusable: 0 must not expire anything
+    request();
+    EXPECT_EQ(1, connections);
+}
+
 TEST(http_client, get) {
     system("mkdir -p /tmp/ease_ut/http_test/");
     system("echo \"this is a http_client request body text for socket stream\" > /tmp/ease_ut/http_test/ease-httpclient-gettestfile");

net/pooled_socket.cpp

@@ -194,7 +194,11 @@ class TCPSocketPool : public ISocketPool {
     join_handle* collector = thread_enable_join(
         thread_create11(&TCPSocketPool::collect, this));
 
-    TCPSocketPool(const SocketPoolArgs& args) : args(args) { }
+    TCPSocketPool(const SocketPoolArgs& args) : args(args) {
+        // 0 is treated as no expiration
+        if (this->args.expiration == 0)
+            this->args.expiration = -1ULL;
+    }
 
     ~TCPSocketPool() override {
         auto th = collector;

net/socket.h

@@ -341,7 +341,9 @@ namespace net {
         Callback<ISocketStream*> heartbeater = { };
         uint64_t heartbeat_interval = 10'000'000;
 
-        // in-stock socket streams are deleted after `expiration` us
+        // in-stock socket streams are deleted after `expiration` us;
+        // -1ULL (default) keeps them until the peer closes them, and 0 is
+        // treated as -1ULL
         uint64_t expiration = -1ULL;
     };