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.

Github-Pull: bitcoin#28551
Rebased-From: 41f9027
  • Loading branch information
stickies-v authored and fanquake committed Oct 4, 2023
1 parent f31899d commit ae86ada
Showing 1 changed file with 70 additions and 16 deletions.
86 changes: 70 additions & 16 deletions src/httpserver.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,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/syscall_sandbox.h>
#include <util/system.h>
Expand All @@ -26,9 +27,10 @@
#include <cstdlib>
#include <deque>
#include <memory>
#include <numeric>
#include <optional>
#include <string>
#include <unordered_set>
#include <unordered_map>

#include <sys/types.h>
#include <sys/stat.h>
Expand Down Expand Up @@ -149,10 +151,68 @@ 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 Mutex 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 decltype(m_tracker)::iterator it) EXCLUSIVE_LOCKS_REQUIRED(m_mutex)
{
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) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
const evhttp_connection* conn{Assert(evhttp_request_get_connection(Assert(req)))};
WITH_LOCK(m_mutex, ++m_tracker[conn]);
}
//! Decrease request counter for the associated connection by 1, remove connection if counter is 0
void RemoveRequest(evhttp_request* req) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
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(it);
}
}
//! Remove a connection entirely
void RemoveConnection(const evhttp_connection* conn) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
LOCK(m_mutex);
auto it{m_tracker.find(Assert(conn))};
if (it != m_tracker.end()) RemoveConnectionInternal(it);
}

size_t CountActiveRequests() const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
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 EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
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 EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
WAIT_LOCK(m_mutex, lock);
m_cv.wait(lock, [this]() EXCLUSIVE_LOCKS_REQUIRED(m_mutex) { 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 @@ -214,14 +274,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 @@ -477,13 +534,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 ae86ada

Please sign in to comment.