http: refactor: use encapsulated HTTPRequestTracker

Introduces and uses a HTTPRequestTracker class to keep track of
how many HTTP requests are currently active, so we don't stop the
server before they're all handled.

This has two purposes:
1. In a next commit, allows us to untrack all requests associated
with a connection without running into lifetime issues of the
connection living longer than the request
(see bitcoin#27909 (comment))

2. Improve encapsulation by making the mutex and cv internal members,
and exposing just the WaitUntilEmpty() method that can be safely
used.

Author: stickies-v

src/httpserver.cpp

@@ -17,6 +17,7 @@
 #include <rpc/protocol.h> // For HTTP status codes
 #include <shutdown.h>
 #include <sync.h>
+#include <util/check.h>
 #include <util/strencodings.h>
 #include <util/threadnames.h>
 #include <util/translation.h>
@@ -26,6 +27,7 @@
 #include <cstdlib>
 #include <deque>
 #include <memory>
+#include <numeric>
 #include <optional>
 #include <string>
 #include <unordered_set>
@@ -149,10 +151,73 @@ static GlobalMutex g_httppathhandlers_mutex;
 static std::vector<HTTPPathHandler> pathHandlers GUARDED_BY(g_httppathhandlers_mutex);
 //! Bound listening sockets
 static std::vector<evhttp_bound_socket *> boundSockets;
+
+/**
+ * @brief Helps keep track of open `evhttp_connection`s with active `evhttp_requests`
+ *
+ */
+class HTTPRequestTracker
+{
+private:
+    mutable GlobalMutex m_mutex;
+    mutable std::condition_variable m_cv;
+    //! For each connection, keep a counter of how many requests are open
+    std::unordered_map<const evhttp_connection*, size_t> m_tracker GUARDED_BY(m_mutex);
+
+    void RemoveConnectionInternal(const evhttp_connection* conn) EXCLUSIVE_LOCKS_REQUIRED(m_mutex)
+    {
+        auto it{m_tracker.find(Assert(conn))};
+        if (it != m_tracker.end()) {
+            m_tracker.erase(it);
+            if (m_tracker.empty()) m_cv.notify_all();
+        }
+    }
+public:
+    //! Increase request counter for the associated connection by 1
+    void AddRequest(evhttp_request* req)
+    {
+        const evhttp_connection* conn{Assert(evhttp_request_get_connection(Assert(req)))};
+        LOCK(m_mutex);
+        auto it{m_tracker.find(conn)};
+        if (it == m_tracker.end()) {
+            // new connection, initialize counter to 1
+            m_tracker[conn] = 1;
+        } else {
+            ++(it->second);
+        }
+    }
+    //! Decrease request counter for the associated connection by 1, remove connection if counter is 0
+    void RemoveRequest(evhttp_request* req)
+    {
+        const evhttp_connection* conn{Assert(evhttp_request_get_connection(Assert(req)))};
+        LOCK(m_mutex);
+        auto it{m_tracker.find(conn)};
+        if(it != m_tracker.end() && it->second > 0) {
+            if(--(it->second) == 0) RemoveConnectionInternal(conn);
+        }
+    }
+    //! Remove a connection entirely
+    void RemoveConnection(const evhttp_connection* conn)
+    {
+        WITH_LOCK(m_mutex, RemoveConnectionInternal(conn));
+    }
+
+    size_t CountActiveRequests() const
+    {
+        LOCK(m_mutex);
+        return std::accumulate(m_tracker.begin(), m_tracker.end(), size_t(0),
+            [](size_t acc_count, const auto& pair) { return acc_count + pair.second; });
+    }
+    size_t CountActiveConnections() const { return WITH_LOCK(m_mutex, return m_tracker.size()); }
+    //! Wait until there are no more connections with active requests in the tracker
+    void WaitUntilEmpty() const
+    {
+        WAIT_LOCK(m_mutex, lock);
+        m_cv.wait(lock, [this]() { return CountActiveConnections() == 0; });
+    }
+};
 //! Track active requests
-static GlobalMutex g_requests_mutex;
-static std::condition_variable g_requests_cv;
-static std::unordered_set<evhttp_request*> g_requests GUARDED_BY(g_requests_mutex);
+static HTTPRequestTracker g_requests;
 
 /** Check if a network address is allowed to access the HTTP server */
 static bool ClientAllowed(const CNetAddr& netaddr)
@@ -210,14 +275,11 @@ std::string RequestMethodString(HTTPRequest::RequestMethod m)
 /** HTTP request callback */
 static void http_request_cb(struct evhttp_request* req, void* arg)
 {
-    // Track requests and notify when a request is completed.
+    // Track active requests
     {
-        WITH_LOCK(g_requests_mutex, g_requests.insert(req));
-        g_requests_cv.notify_all();
+        g_requests.AddRequest(req);
         evhttp_request_set_on_complete_cb(req, [](struct evhttp_request* req, void*) {
-            auto n{WITH_LOCK(g_requests_mutex, return g_requests.erase(req))};
-            assert(n == 1);
-            g_requests_cv.notify_all();
+            g_requests.RemoveRequest(req);
         }, nullptr);
     }
 
@@ -473,13 +535,10 @@ void StopHTTPServer()
     }
     boundSockets.clear();
     {
-        WAIT_LOCK(g_requests_mutex, lock);
-        if (!g_requests.empty()) {
-            LogPrint(BCLog::HTTP, "Waiting for %d requests to stop HTTP server\n", g_requests.size());
+        if (g_requests.CountActiveConnections() != 0) {
+            LogPrint(BCLog::HTTP, "Waiting for %d requests to stop HTTP server\n", g_requests.CountActiveRequests());
         }
-        g_requests_cv.wait(lock, []() EXCLUSIVE_LOCKS_REQUIRED(g_requests_mutex) {
-            return g_requests.empty();
-        });
+        g_requests.WaitUntilEmpty();
     }
     if (eventHTTP) {
         // Schedule a callback to call evhttp_free in the event base thread, so