Fix #2530: Replace blocking requestsBufferSize future with lock-free atomic

harshal24-chavan · Maratha96 · commit 4902ba6ae100 · 2026-06-25T21:29:07.000+05:30
- Introduced `outstandingRequests()` to `HttpClient` to reflect total connection load (buffered requests + in-flight pipelined requests).
- Implemented an `std::atomic&lt;std::size_t&gt;` tracker for `pipeliningCallbacks_` for inspection of in-flight requests from the main thread.
- Retained helper functions for `requestsBuffer_` to safely guard `std::list` mutations.
- Added `HttpClientOutstandingRequests` integration test using `PipeliningTest::normalPipe`.
diff --git a/lib/inc/drogon/HttpClient.h b/lib/inc/drogon/HttpClient.h
@@ -185,6 +185,11 @@ class DROGON_EXPORT HttpClient : public trantor::NonCopyable
      */
     virtual std::size_t requestsBufferSize() = 0;
 
+    /**
+     * @brief Get the total number of outstanding requests (buffered + in-flight).
+     */
+    virtual std::size_t outstandingRequests() const = 0;
+
     /// Set the pipelining depth, which is the number of requests that are not
     /// responding.
     /**
diff --git a/lib/src/HttpClientImpl.cc b/lib/src/HttpClientImpl.cc
@@ -79,7 +79,10 @@ void HttpClientImpl::createTcpClient()
                                      thisPtr->requestsBuffer_.front().first);
                     thisPtr->pipeliningCallbacks_.push(
                         std::move(thisPtr->requestsBuffer_.front()));
-                    thisPtr->requestsBuffer_.pop_front();
+                    thisPtr->pipeliningCallbacksSize_.fetch_add(
+                        1, std::memory_order_relaxed);
+
+                    thisPtr->popFrontRequest();
                 }
             }
             else
@@ -340,7 +343,7 @@ void HttpClientImpl::sendRequestInLoop(const HttpRequestPtr &req,
                 {
                     if (iter->first == callbackParamsPtr->requestPtr)
                     {
-                        thisPtr->requestsBuffer_.erase(iter);
+                        thisPtr->eraseRequest(iter);
                         break;
                     }
                 }
@@ -419,12 +422,12 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,
     {
         auto callbackPtr =
             std::make_shared<drogon::HttpReqCallback>(std::move(callback));
-        requestsBuffer_.push_back(
-            {req,
-             [thisPtr = shared_from_this(),
-              callbackPtr](ReqResult result, const HttpResponsePtr &response) {
-                 (*callbackPtr)(result, response);
-             }});
+        enqueueRequest(req,
+                       [thisPtr = shared_from_this(),
+                        callbackPtr](ReqResult result,
+                                     const HttpResponsePtr &response) {
+                           (*callbackPtr)(result, response);
+                       });
 
         if (domain_.empty() || !isDomainName_)
         {
@@ -436,7 +439,7 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,
             // No ip address and no domain, respond with BadServerAddress
             else
             {
-                requestsBuffer_.pop_front();
+                popFrontRequest();
                 (*callbackPtr)(ReqResult::BadServerAddress, nullptr);
                 assert(requestsBuffer_.empty());
             }
@@ -480,7 +483,8 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,
                     {
                         auto &reqAndCb = (thisPtr->requestsBuffer_).front();
                         reqAndCb.second(ReqResult::BadServerAddress, nullptr);
-                        (thisPtr->requestsBuffer_).pop_front();
+
+                        thisPtr->popFrontRequest();
                     }
                 });
             });
@@ -495,13 +499,11 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,
     // Not connected, push request to buffer and wait for connection
     if (!connPtr || connPtr->disconnected())
     {
-        requestsBuffer_.push_back(
-            {req,
-             [thisPtr,
-              callback = std::move(callback)](ReqResult result,
-                                              const HttpResponsePtr &response) {
-                 callback(result, response);
-             }});
+        enqueueRequest(req,
+                       [thisPtr, callback = std::move(callback)](
+                           ReqResult result, const HttpResponsePtr &response) {
+                           callback(result, response);
+                       });
         return;
     }
 
@@ -517,16 +519,15 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,
                                               const HttpResponsePtr &response) {
                  callback(result, response);
              }});
+        pipeliningCallbacksSize_.fetch_add(1, std::memory_order_relaxed);
     }
     else
     {
-        requestsBuffer_.push_back(
-            {req,
-             [thisPtr,
-              callback = std::move(callback)](ReqResult result,
-                                              const HttpResponsePtr &response) {
-                 callback(result, response);
-             }});
+        enqueueRequest(req,
+                       [thisPtr, callback = std::move(callback)](
+                           ReqResult result, const HttpResponsePtr &response) {
+                           callback(result, response);
+                       });
     }
 }
 
@@ -562,6 +563,7 @@ void HttpClientImpl::handleResponse(
 #endif
     auto cb = std::move(reqAndCb);
     pipeliningCallbacks_.pop();
+    pipeliningCallbacksSize_.fetch_sub(1, std::memory_order_relaxed);
     handleCookies(resp);
     cb.second(ReqResult::Ok, resp);
 
@@ -576,7 +578,8 @@ void HttpClientImpl::handleResponse(
             auto &reqAndCallback = requestsBuffer_.front();
             sendReq(connPtr, reqAndCallback.first);
             pipeliningCallbacks_.push(std::move(reqAndCallback));
-            requestsBuffer_.pop_front();
+            pipeliningCallbacksSize_.fetch_add(1, std::memory_order_relaxed);
+            popFrontRequest();
         }
         else
         {
@@ -592,6 +595,7 @@ void HttpClientImpl::handleResponse(
         {
             auto cb = std::move(pipeliningCallbacks_.front());
             pipeliningCallbacks_.pop();
+            pipeliningCallbacksSize_.fetch_sub(1, std::memory_order_relaxed);
             cb.second(ReqResult::NetworkFailure, nullptr);
         }
     }
@@ -674,12 +678,13 @@ void HttpClientImpl::onError(ReqResult result)
     {
         auto cb = std::move(pipeliningCallbacks_.front());
         pipeliningCallbacks_.pop();
+        pipeliningCallbacksSize_.fetch_sub(1, std::memory_order_relaxed);
         cb.second(result, nullptr);
     }
     while (!requestsBuffer_.empty())
     {
         auto cb = std::move(requestsBuffer_.front().second);
-        requestsBuffer_.pop_front();
+        popFrontRequest();
         cb(result, nullptr);
     }
     tcpClientPtr_.reset();
diff --git a/lib/src/HttpClientImpl.h b/lib/src/HttpClientImpl.h
@@ -19,6 +19,7 @@
 #include <trantor/net/EventLoop.h>
 #include <trantor/net/Resolver.h>
 #include <trantor/net/TcpClient.h>
+#include <atomic>
 #include <cstddef>
 #include <functional>
 #include <future>
@@ -118,21 +119,42 @@ class HttpClientImpl final : public HttpClient,
         sockOptCallback_ = std::move(cb);
     }
 
+    std::size_t outstandingRequests() const override
+    {
+        return requestsBufferSize_.load(std::memory_order_relaxed) +
+               pipeliningCallbacksSize_.load(std::memory_order_relaxed);
+    }
+
     std::size_t requestsBufferSize() override
     {
-        if (loop_->isInLoopThread())
-        {
-            return requestsBuffer_.size();
-        }
-        else
+        return requestsBufferSize_.load(std::memory_order_relaxed);
+    }
+
+    void enqueueRequest(const HttpRequestPtr &req,
+                        const HttpReqCallback &callback)
+    {
+        requestsBuffer_.push_back({req, callback});
+        requestsBufferSize_.fetch_add(1, std::memory_order_relaxed);
+    }
+
+    void popFrontRequest()
+    {
+        if (!requestsBuffer_.empty())
         {
-            std::promise<std::size_t> bufferSize;
-            loop_->queueInLoop(
-                [&] { bufferSize.set_value(requestsBuffer_.size()); });
-            return bufferSize.get_future().get();
+            requestsBuffer_.pop_front();
+            requestsBufferSize_.fetch_sub(1, std::memory_order_relaxed);
         }
     }
 
+    using RequestBufferIter =
+        std::list<std::pair<HttpRequestPtr, HttpReqCallback>>::iterator;
+
+    void eraseRequest(RequestBufferIter iter)
+    {
+        requestsBuffer_.erase(iter);
+        requestsBufferSize_.fetch_sub(1, std::memory_order_relaxed);
+    }
+
   private:
     std::shared_ptr<trantor::TcpClient> tcpClientPtr_;
     trantor::EventLoop *loop_;
@@ -157,6 +179,8 @@ class HttpClientImpl final : public HttpClient,
     void onError(ReqResult result);
     std::string domain_;
     bool isDomainName_{true};  // true if domain_ is name
+    std::atomic<std::size_t> requestsBufferSize_{0};
+    std::atomic<std::size_t> pipeliningCallbacksSize_{0};
     size_t pipeliningDepth_{0};
     bool enableCookies_{false};
     std::vector<Cookie> validCookies_;
diff --git a/lib/tests/CMakeLists.txt b/lib/tests/CMakeLists.txt
@@ -64,6 +64,7 @@ if (BUILD_CTL)
       integration_test/server/CustomHeaderFilter.cc
       integration_test/server/DoNothingPlugin.cc
       integration_test/server/ForwardCtrl.cc
+      integration_test/server/HttpClientOutstandingRequests.cc
       integration_test/server/JsonTestController.cc
       integration_test/server/ListParaCtl.cc
       integration_test/server/PipeliningTest.cc
diff --git a/lib/tests/integration_test/server/HttpClientOutstandingRequests.cc b/lib/tests/integration_test/server/HttpClientOutstandingRequests.cc
@@ -0,0 +1,52 @@
+#include <drogon/drogon_test.h>
+#include <drogon/HttpClient.h>
+#include <atomic>
+#include <thread>
+#include <chrono>
+
+using namespace drogon;
+
+DROGON_TEST(HttpClientOutstandingRequests)
+{
+    auto client = HttpClient::newHttpClient("http://127.0.0.1:8848");
+
+    // Enable pipelining to allow multiple in-flight requests simultaneously
+    client->setPipeliningDepth(64);
+
+    CHECK(client->requestsBufferSize() == 0);
+    CHECK(client->outstandingRequests() == 0);
+
+    const int totalRequests = 50;
+    std::atomic<int> completedRequests{0};
+
+    for (int i = 0; i < totalRequests; ++i)
+    {
+        auto req = HttpRequest::newHttpRequest();
+        req->setPath("/PipeliningTest/normalPipe");
+
+        client->sendRequest(
+            req,
+            [&completedRequests](ReqResult result,
+                                 const HttpResponsePtr &response) {
+                if (result == ReqResult::Ok)
+                {
+                    completedRequests++;
+                }
+            });
+    }
+
+    size_t currentOutstanding = client->outstandingRequests();
+    size_t currentBuffer = client->outstandingRequests();
+
+    CHECK(currentOutstanding > 0);
+
+    CHECK(currentOutstanding >= currentBuffer);
+
+    while (completedRequests < totalRequests)
+    {
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+    CHECK(client->requestsBufferSize() == 0);
+    CHECK(client->outstandingRequests() == 0);
+}

Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,10 @@ void HttpClientImpl::createTcpClient()`
`79`	`79`	`thisPtr->requestsBuffer_.front().first);`
`80`	`80`	`thisPtr->pipeliningCallbacks_.push(`
`81`	`81`	`std::move(thisPtr->requestsBuffer_.front()));`
`82`		`- thisPtr->requestsBuffer_.pop_front();`
	`82`	`+ thisPtr->pipeliningCallbacksSize_.fetch_add(`
	`83`	`+ 1, std::memory_order_relaxed);`
	`84`	`+`
	`85`	`+ thisPtr->popFrontRequest();`
`83`	`86`	`}`
`84`	`87`	`}`
`85`	`88`	`else`
`@@ -340,7 +343,7 @@ void HttpClientImpl::sendRequestInLoop(const HttpRequestPtr &req,`
`340`	`343`	`{`
`341`	`344`	`if (iter->first == callbackParamsPtr->requestPtr)`
`342`	`345`	`{`
`343`		`- thisPtr->requestsBuffer_.erase(iter);`
	`346`	`+ thisPtr->eraseRequest(iter);`
`344`	`347`	`break;`
`345`	`348`	`}`
`346`	`349`	`}`
`@@ -419,12 +422,12 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,`
`419`	`422`	`{`
`420`	`423`	`auto callbackPtr =`
`421`	`424`	`std::make_shared<drogon::HttpReqCallback>(std::move(callback));`
`422`		`- requestsBuffer_.push_back(`
`423`		`- {req,`
`424`		`- [thisPtr = shared_from_this(),`
`425`		`- callbackPtr](ReqResult result, const HttpResponsePtr &response) {`
`426`		`- (*callbackPtr)(result, response);`
`427`		`- }});`
	`425`	`+ enqueueRequest(req,`
	`426`	`+ [thisPtr = shared_from_this(),`
	`427`	`+ callbackPtr](ReqResult result,`
	`428`	`+ const HttpResponsePtr &response) {`
	`429`	`+ (*callbackPtr)(result, response);`
	`430`	`+ });`
`428`	`431`
`429`	`432`	`if (domain_.empty() \|\| !isDomainName_)`
`430`	`433`	`{`
`@@ -436,7 +439,7 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,`
`436`	`439`	`// No ip address and no domain, respond with BadServerAddress`
`437`	`440`	`else`
`438`	`441`	`{`
`439`		`- requestsBuffer_.pop_front();`
	`442`	`+ popFrontRequest();`
`440`	`443`	`(*callbackPtr)(ReqResult::BadServerAddress, nullptr);`
`441`	`444`	`assert(requestsBuffer_.empty());`
`442`	`445`	`}`
`@@ -480,7 +483,8 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,`
`480`	`483`	`{`
`481`	`484`	`auto &reqAndCb = (thisPtr->requestsBuffer_).front();`
`482`	`485`	`reqAndCb.second(ReqResult::BadServerAddress, nullptr);`
`483`		`- (thisPtr->requestsBuffer_).pop_front();`
	`486`	`+`
	`487`	`+ thisPtr->popFrontRequest();`
`484`	`488`	`}`
`485`	`489`	`});`
`486`	`490`	`});`
`@@ -495,13 +499,11 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,`
`495`	`499`	`// Not connected, push request to buffer and wait for connection`
`496`	`500`	`if (!connPtr \|\| connPtr->disconnected())`
`497`	`501`	`{`
`498`		`- requestsBuffer_.push_back(`
`499`		`- {req,`
`500`		`- [thisPtr,`
`501`		`- callback = std::move(callback)](ReqResult result,`
`502`		`- const HttpResponsePtr &response) {`
`503`		`- callback(result, response);`
`504`		`- }});`
	`502`	`+ enqueueRequest(req,`
	`503`	`+ [thisPtr, callback = std::move(callback)](`
	`504`	`+ ReqResult result, const HttpResponsePtr &response) {`
	`505`	`+ callback(result, response);`
	`506`	`+ });`
`505`	`507`	`return;`
`506`	`508`	`}`
`507`	`509`
`@@ -517,16 +519,15 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,`
`517`	`519`	`const HttpResponsePtr &response) {`
`518`	`520`	`callback(result, response);`
`519`	`521`	`}});`
	`522`	`+ pipeliningCallbacksSize_.fetch_add(1, std::memory_order_relaxed);`
`520`	`523`	`}`
`521`	`524`	`else`
`522`	`525`	`{`
`523`		`- requestsBuffer_.push_back(`
`524`		`- {req,`
`525`		`- [thisPtr,`
`526`		`- callback = std::move(callback)](ReqResult result,`
`527`		`- const HttpResponsePtr &response) {`
`528`		`- callback(result, response);`
`529`		`- }});`
	`526`	`+ enqueueRequest(req,`
	`527`	`+ [thisPtr, callback = std::move(callback)](`
	`528`	`+ ReqResult result, const HttpResponsePtr &response) {`
	`529`	`+ callback(result, response);`
	`530`	`+ });`
`530`	`531`	`}`
`531`	`532`	`}`
`532`	`533`
`@@ -562,6 +563,7 @@ void HttpClientImpl::handleResponse(`
`562`	`563`	`#endif`
`563`	`564`	`auto cb = std::move(reqAndCb);`
`564`	`565`	`pipeliningCallbacks_.pop();`
	`566`	`+ pipeliningCallbacksSize_.fetch_sub(1, std::memory_order_relaxed);`
`565`	`567`	`handleCookies(resp);`
`566`	`568`	`cb.second(ReqResult::Ok, resp);`
`567`	`569`
`@@ -576,7 +578,8 @@ void HttpClientImpl::handleResponse(`
`576`	`578`	`auto &reqAndCallback = requestsBuffer_.front();`
`577`	`579`	`sendReq(connPtr, reqAndCallback.first);`
`578`	`580`	`pipeliningCallbacks_.push(std::move(reqAndCallback));`
`579`		`- requestsBuffer_.pop_front();`
	`581`	`+ pipeliningCallbacksSize_.fetch_add(1, std::memory_order_relaxed);`
	`582`	`+ popFrontRequest();`
`580`	`583`	`}`
`581`	`584`	`else`
`582`	`585`	`{`
`@@ -592,6 +595,7 @@ void HttpClientImpl::handleResponse(`
`592`	`595`	`{`
`593`	`596`	`auto cb = std::move(pipeliningCallbacks_.front());`
`594`	`597`	`pipeliningCallbacks_.pop();`
	`598`	`+ pipeliningCallbacksSize_.fetch_sub(1, std::memory_order_relaxed);`
`595`	`599`	`cb.second(ReqResult::NetworkFailure, nullptr);`
`596`	`600`	`}`
`597`	`601`	`}`
`@@ -674,12 +678,13 @@ void HttpClientImpl::onError(ReqResult result)`
`674`	`678`	`{`
`675`	`679`	`auto cb = std::move(pipeliningCallbacks_.front());`
`676`	`680`	`pipeliningCallbacks_.pop();`
	`681`	`+ pipeliningCallbacksSize_.fetch_sub(1, std::memory_order_relaxed);`
`677`	`682`	`cb.second(result, nullptr);`
`678`	`683`	`}`
`679`	`684`	`while (!requestsBuffer_.empty())`
`680`	`685`	`{`
`681`	`686`	`auto cb = std::move(requestsBuffer_.front().second);`
`682`		`- requestsBuffer_.pop_front();`
	`687`	`+ popFrontRequest();`
`683`	`688`	`cb(result, nullptr);`
`684`	`689`	`}`
`685`	`690`	`tcpClientPtr_.reset();`