Skip to content

Commit

Permalink
http: refactor: use encapsulated HTTPRequestTracker
Browse files Browse the repository at this point in the history
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.
  • Loading branch information
stickies-v committed Sep 29, 2023
1 parent 9d5150a commit dc44bba
Showing 1 changed file with 74 additions and 15 deletions.
89 changes: 74 additions & 15 deletions src/httpserver.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand All @@ -26,6 +27,7 @@
#include <cstdlib>
#include <deque>
#include <memory>
#include <numeric>
#include <optional>
#include <string>
#include <unordered_set>
Expand Down Expand Up @@ -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 m_tracker.empty(); });
}
};
//! 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)
Expand Down Expand Up @@ -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);
}

Expand Down Expand Up @@ -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
Expand Down

0 comments on commit dc44bba

Please sign in to comment.