highperfocused/bitcoin
1
0
Fork 0
mirror of https://github.com/bitcoin/bitcoin.git synced 2026-09-12 21:52:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

http: remove libevent usage from this subsystem

Browse Source
This commit is contained in:
Matthew Zipkin
2025-03-13 13:57:58 -04:00
parent e427c227fa
commit 8c1eea0777
4 changed files with 21 additions and 662 deletions
Download Patch File Download Diff File Expand all files Collapse all files

481
src/httpserver.cpp
View File

@@ -43,16 +43,6 @@
#include <sys/types.h>
#include <sys/stat.h>
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <event2/http.h>
#include <event2/http_struct.h>
#include <event2/keyvalq_struct.h>
#include <event2/thread.h>
#include <event2/util.h>
#include <support/events.h>
//! The set of sockets cannot be modified while waiting, so
//! the sleep time needs to be small to avoid new sockets stalling.
static constexpr auto SELECT_TIMEOUT{50ms};
@@ -63,9 +53,6 @@ static constexpr int SOCKET_OPTION_TRUE{1};
using common::InvalidPortErrMsg;
using http_bitcoin::HTTPRequest;
/** Maximum size of http request (request line + headers) */
static const size_t MAX_HEADERS_SIZE = 8192;
struct HTTPPathHandler
{
HTTPPathHandler(std::string _prefix, bool _exactMatch, HTTPRequestHandler _handler):
@@ -79,78 +66,17 @@ struct HTTPPathHandler
/** HTTP module state */
//! libevent event loop
static struct event_base* eventBase = nullptr;
//! HTTP server
static struct evhttp* eventHTTP = nullptr;
static std::unique_ptr<http_bitcoin::HTTPServer> g_http_server{nullptr};
//! List of subnets to allow RPC connections from
static std::vector<CSubNet> rpc_allow_subnets;
//! Handlers for (sub)paths
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;
/// \anchor http_pool
//! Http thread pool - future: encapsulate in HttpContext
static ThreadPool g_threadpool_http("http");
static int g_max_queue_depth{100};
/**
* @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 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 HTTPRequestTracker g_requests;
/** Check if a network address is allowed to access the HTTP server */
static bool ClientAllowed(const CNetAddr& netaddr)
{
@@ -279,57 +205,6 @@ static void RejectRequest(std::unique_ptr<http_bitcoin::HTTPRequest> hreq)
hreq->WriteReply(HTTP_SERVICE_UNAVAILABLE);
}
/** HTTP request callback */
static void http_request_cb(struct evhttp_request* req, void* arg)
{
evhttp_connection* conn{evhttp_request_get_connection(req)};
// Track active requests
{
g_requests.AddRequest(req);
evhttp_request_set_on_complete_cb(req, [](struct evhttp_request* req, void*) {
g_requests.RemoveRequest(req);
}, nullptr);
evhttp_connection_set_closecb(conn, [](evhttp_connection* conn, void* arg) {
g_requests.RemoveConnection(conn);
}, nullptr);
}
// Disable reading to work around a libevent bug, fixed in 2.1.9
// See https://github.com/libevent/libevent/commit/5ff8eb26371c4dc56f384b2de35bea2d87814779
// and https://github.com/bitcoin/bitcoin/pull/11593.
if (event_get_version_number() >= 0x02010600 && event_get_version_number() < 0x02010900) {
if (conn) {
bufferevent* bev = evhttp_connection_get_bufferevent(conn);
if (bev) {
bufferevent_disable(bev, EV_READ);
}
}
}
auto hreq{std::make_shared<http_libevent::HTTPRequest>(req, *static_cast<const util::SignalInterrupt*>(arg))};
// Disabled now that http_libevent is deprecated, or code won't compile.
// This line is currently unreachable and will be cleaned up in a future commit.
// MaybeDispatchRequestToWorker(std::move(hreq));
Assume(false);
}
/** Callback to reject HTTP requests after shutdown. */
static void http_reject_request_cb(struct evhttp_request* req, void*)
{
LogDebug(BCLog::HTTP, "Rejecting request while shutting down\n");
evhttp_send_error(req, HTTP_SERVUNAVAIL, nullptr);
}
/** Event dispatcher thread */
static void ThreadHTTP(struct event_base* base)
{
util::ThreadRename("http");
LogDebug(BCLog::HTTP, "Entering http event loop\n");
event_base_dispatch(base);
// Event loop will be interrupted by InterruptHTTPServer()
LogDebug(BCLog::HTTP, "Exited http event loop\n");
}
static std::vector<std::pair<std::string, uint16_t>> GetBindAddresses()
{
uint16_t http_port{static_cast<uint16_t>(gArgs.GetIntArg("-rpcport", BaseParams().RPCPort()))};
@@ -363,362 +238,6 @@ static std::vector<std::pair<std::string, uint16_t>> GetBindAddresses()
return endpoints;
}
/** libevent event log callback */
static void libevent_log_cb(int severity, const char *msg)
{
switch (severity) {
case EVENT_LOG_DEBUG:
LogDebug(BCLog::LIBEVENT, "%s", msg);
break;
case EVENT_LOG_MSG:
LogInfo("libevent: %s", msg);
break;
case EVENT_LOG_WARN:
LogWarning("libevent: %s", msg);
break;
default: // EVENT_LOG_ERR and others are mapped to error
LogError("libevent: %s", msg);
break;
}
}
namespace http_libevent {
/** Bind HTTP server to specified addresses */
static bool HTTPBindAddresses(struct evhttp* http)
{
std::vector<std::pair<std::string, uint16_t>> endpoints{GetBindAddresses()};
for (std::vector<std::pair<std::string, uint16_t> >::iterator i = endpoints.begin(); i != endpoints.end(); ++i) {
LogInfo("Binding RPC on address %s port %i", i->first, i->second);
evhttp_bound_socket *bind_handle = evhttp_bind_socket_with_handle(http, i->first.empty() ? nullptr : i->first.c_str(), i->second);
if (bind_handle) {
const std::optional<CNetAddr> addr{LookupHost(i->first, false)};
if (i->first.empty() || (addr.has_value() && addr->IsBindAny())) {
LogWarning("The RPC server is not safe to expose to untrusted networks such as the public internet");
}
// Set the no-delay option (disable Nagle's algorithm) on the TCP socket.
evutil_socket_t fd = evhttp_bound_socket_get_fd(bind_handle);
int one = 1;
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<char*>(&one), sizeof(one)) == SOCKET_ERROR) {
LogInfo("WARNING: Unable to set TCP_NODELAY on RPC server socket, continuing anyway");
}
boundSockets.push_back(bind_handle);
} else {
LogWarning("Binding RPC on address %s port %i failed.", i->first, i->second);
}
}
return !boundSockets.empty();
}
bool InitHTTPServer(const util::SignalInterrupt& interrupt)
{
if (!InitHTTPAllowList())
return false;
// Redirect libevent's logging to our own log
event_set_log_callback(&libevent_log_cb);
// Update libevent's log handling.
UpdateHTTPServerLogging(LogInstance().WillLogCategory(BCLog::LIBEVENT));
#ifdef WIN32
evthread_use_windows_threads();
#else
evthread_use_pthreads();
#endif
raii_event_base base_ctr = obtain_event_base();
/* Create a new evhttp object to handle requests. */
raii_evhttp http_ctr = obtain_evhttp(base_ctr.get());
struct evhttp* http = http_ctr.get();
if (!http) {
LogError("Couldn't create evhttp. Exiting.");
return false;
}
evhttp_set_timeout(http, gArgs.GetIntArg("-rpcservertimeout", DEFAULT_HTTP_SERVER_TIMEOUT));
evhttp_set_max_headers_size(http, MAX_HEADERS_SIZE);
evhttp_set_max_body_size(http, MAX_SIZE);
evhttp_set_gencb(http, http_request_cb, (void*)&interrupt);
if (!HTTPBindAddresses(http)) {
LogError("Unable to bind any endpoint for RPC server");
return false;
}
LogDebug(BCLog::HTTP, "Initialized HTTP server\n");
g_max_queue_depth = std::max(gArgs.GetArg("-rpcworkqueue", DEFAULT_HTTP_WORKQUEUE), 1);
LogDebug(BCLog::HTTP, "set work queue of depth %d", g_max_queue_depth);
// transfer ownership to eventBase/HTTP via .release()
eventBase = base_ctr.release();
eventHTTP = http_ctr.release();
return true;
}
void UpdateHTTPServerLogging(bool enable) {
if (enable) {
event_enable_debug_logging(EVENT_DBG_ALL);
} else {
event_enable_debug_logging(EVENT_DBG_NONE);
}
}
static std::thread g_thread_http;
void StartHTTPServer()
{
int rpcThreads = std::max(gArgs.GetArg("-rpcthreads", DEFAULT_HTTP_THREADS), 1);
LogInfo("Starting HTTP server with %d worker threads", rpcThreads);
g_threadpool_http.Start(rpcThreads);
g_thread_http = std::thread(ThreadHTTP, eventBase);
}
void InterruptHTTPServer()
{
LogDebug(BCLog::HTTP, "Interrupting HTTP server\n");
if (eventHTTP) {
// Reject requests on current connections
evhttp_set_gencb(eventHTTP, http_reject_request_cb, nullptr);
}
// Interrupt pool after disabling requests
g_threadpool_http.Interrupt();
}
void StopHTTPServer()
{
LogDebug(BCLog::HTTP, "Stopping HTTP server\n");
LogDebug(BCLog::HTTP, "Waiting for HTTP worker threads to exit\n");
g_threadpool_http.Stop();
// Unlisten sockets, these are what make the event loop running, which means
// that after this and all connections are closed the event loop will quit.
for (evhttp_bound_socket *socket : boundSockets) {
evhttp_del_accept_socket(eventHTTP, socket);
}
boundSockets.clear();
{
if (const auto n_connections{g_requests.CountActiveConnections()}; n_connections != 0) {
LogDebug(BCLog::HTTP, "Waiting for %d connections to stop HTTP server\n", n_connections);
}
g_requests.WaitUntilEmpty();
}
if (eventHTTP) {
// Schedule a callback to call evhttp_free in the event base thread, so
// that evhttp_free does not need to be called again after the handling
// of unfinished request connections that follows.
event_base_once(eventBase, -1, EV_TIMEOUT, [](evutil_socket_t, short, void*) {
evhttp_free(eventHTTP);
eventHTTP = nullptr;
}, nullptr, nullptr);
}
if (eventBase) {
LogDebug(BCLog::HTTP, "Waiting for HTTP event thread to exit\n");
if (g_thread_http.joinable()) g_thread_http.join();
event_base_free(eventBase);
eventBase = nullptr;
}
LogDebug(BCLog::HTTP, "Stopped HTTP server\n");
}
struct event_base* EventBase()
{
return eventBase;
}
} // namespace http_libevent
static void httpevent_callback_fn(evutil_socket_t, short, void* data)
{
// Static handler: simply call inner handler
HTTPEvent *self = static_cast<HTTPEvent*>(data);
self->handler();
if (self->deleteWhenTriggered)
delete self;
}
HTTPEvent::HTTPEvent(struct event_base* base, bool _deleteWhenTriggered, const std::function<void()>& _handler):
deleteWhenTriggered(_deleteWhenTriggered), handler(_handler)
{
ev = event_new(base, -1, 0, httpevent_callback_fn, this);
assert(ev);
}
HTTPEvent::~HTTPEvent()
{
event_free(ev);
}
void HTTPEvent::trigger(struct timeval* tv)
{
if (tv == nullptr)
event_active(ev, 0, 0); // immediately trigger event in main thread
else
evtimer_add(ev, tv); // trigger after timeval passed
}
namespace http_libevent {
HTTPRequest::HTTPRequest(struct evhttp_request* _req, const util::SignalInterrupt& interrupt, bool _replySent)
: req(_req), m_interrupt(interrupt), replySent(_replySent)
{
}
HTTPRequest::~HTTPRequest()
{
if (!replySent) {
// Keep track of whether reply was sent to avoid request leaks
LogWarning("Unhandled HTTP request");
WriteReply(HTTP_INTERNAL_SERVER_ERROR, "Unhandled request");
}
// evhttpd cleans up the request, as long as a reply was sent.
}
std::pair<bool, std::string> HTTPRequest::GetHeader(const std::string& hdr) const
{
const struct evkeyvalq* headers = evhttp_request_get_input_headers(req);
assert(headers);
const char* val = evhttp_find_header(headers, hdr.c_str());
if (val)
return std::make_pair(true, val);
else
return std::make_pair(false, "");
}
std::string HTTPRequest::ReadBody()
{
struct evbuffer* buf = evhttp_request_get_input_buffer(req);
if (!buf)
return "";
size_t size = evbuffer_get_length(buf);
/** Trivial implementation: if this is ever a performance bottleneck,
* internal copying can be avoided in multi-segment buffers by using
* evbuffer_peek and an awkward loop. Though in that case, it'd be even
* better to not copy into an intermediate string but use a stream
* abstraction to consume the evbuffer on the fly in the parsing algorithm.
*/
const char* data = (const char*)evbuffer_pullup(buf, size);
if (!data) // returns nullptr in case of empty buffer
return "";
std::string rv(data, size);
evbuffer_drain(buf, size);
return rv;
}
void HTTPRequest::WriteHeader(const std::string& hdr, const std::string& value)
{
struct evkeyvalq* headers = evhttp_request_get_output_headers(req);
assert(headers);
evhttp_add_header(headers, hdr.c_str(), value.c_str());
}
/** Closure sent to main thread to request a reply to be sent to
* a HTTP request.
* Replies must be sent in the main loop in the main http thread,
* this cannot be done from worker threads.
*/
void HTTPRequest::WriteReply(int nStatus, std::span<const std::byte> reply)
{
assert(!replySent && req);
if (m_interrupt) {
WriteHeader("Connection", "close");
}
// Send event to main http thread to send reply message
struct evbuffer* evb = evhttp_request_get_output_buffer(req);
assert(evb);
evbuffer_add(evb, reply.data(), reply.size());
auto req_copy = req;
HTTPEvent* ev = new HTTPEvent(eventBase, true, [req_copy, nStatus]{
evhttp_send_reply(req_copy, nStatus, nullptr, nullptr);
// Re-enable reading from the socket. This is the second part of the libevent
// workaround above.
if (event_get_version_number() >= 0x02010600 && event_get_version_number() < 0x02010900) {
evhttp_connection* conn = evhttp_request_get_connection(req_copy);
if (conn) {
bufferevent* bev = evhttp_connection_get_bufferevent(conn);
if (bev) {
bufferevent_enable(bev, EV_READ | EV_WRITE);
}
}
}
});
ev->trigger(nullptr);
replySent = true;
req = nullptr; // transferred back to main thread
}
CService HTTPRequest::GetPeer() const
{
evhttp_connection* con = evhttp_request_get_connection(req);
CService peer;
if (con) {
// evhttp retains ownership over returned address string
const char* address = "";
uint16_t port = 0;
#ifdef HAVE_EVHTTP_CONNECTION_GET_PEER_CONST_CHAR
evhttp_connection_get_peer(con, &address, &port);
#else
evhttp_connection_get_peer(con, (char**)&address, &port);
#endif // HAVE_EVHTTP_CONNECTION_GET_PEER_CONST_CHAR
peer = MaybeFlipIPv6toCJDNS(LookupNumeric(address, port));
}
return peer;
}
std::string HTTPRequest::GetURI() const
{
return evhttp_request_get_uri(req);
}
HTTPRequestMethod HTTPRequest::GetRequestMethod() const
{
switch (evhttp_request_get_command(req)) {
case EVHTTP_REQ_GET:
return HTTPRequestMethod::GET;
case EVHTTP_REQ_POST:
return HTTPRequestMethod::POST;
case EVHTTP_REQ_HEAD:
return HTTPRequestMethod::HEAD;
case EVHTTP_REQ_PUT:
return HTTPRequestMethod::PUT;
default:
return HTTPRequestMethod::UNKNOWN;
}
}
std::optional<std::string> HTTPRequest::GetQueryParameter(const std::string& key) const
{
const char* uri{evhttp_request_get_uri(req)};
return GetQueryParameterFromUri(uri, key);
}
std::optional<std::string> GetQueryParameterFromUri(const char* uri, const std::string& key)
{
evhttp_uri* uri_parsed{evhttp_uri_parse(uri)};
if (!uri_parsed) {
throw std::runtime_error("URI parsing failed, it likely contained RFC 3986 invalid characters");
}
const char* query{evhttp_uri_get_query(uri_parsed)};
std::optional<std::string> result;
if (query) {
// Parse the query string into a key-value queue and iterate over it
struct evkeyvalq params_q;
evhttp_parse_query_str(query, &params_q);
for (struct evkeyval* param{params_q.tqh_first}; param != nullptr; param = param->next.tqe_next) {
if (param->key == key) {
result = param->value;
break;
}
}
evhttp_clear_headers(&params_q);
}
evhttp_uri_free(uri_parsed);
return result;
}
} // namespace http_libevent
void RegisterHTTPHandler(const std::string &prefix, bool exactMatch, const HTTPRequestHandler &handler)
{
LogDebug(BCLog::HTTP, "Registering HTTP handler for %s (exactmatch %d)\n", prefix, exactMatch);