feat: convert 3D scenery models on worker threads with deferred replies

The dashboard's web server answers one request at a time, so converting a big
.nif (glom files up to tens of MB) held up every other request, flairs included.

- dWeb: Web::Defer hands a request to another thread; the reply is sent from the
  web thread on its next poll (DeferredQueue). A client that leaves first cancels
  it and the late reply is dropped. The synchronous route API is unchanged.
- Web::Shutdown closes connections while the state their close events touch is
  still alive; the destructor no longer runs handlers during static destruction
  (stopping the dashboard aborted in ~WSClient).
- WorkerPool: priority lanes, with one thread only for urgent work (flairs,
  small models, textures), and limited background work.
- Scenery: mesh and texture routes (and the showcase's) convert on the pool;
  thread-safe memory and disk caches, one conversion per model at a time with
  waiters sharing it; zones are converted ahead onto the disk cache while viewed.
- Setting scenery_workers (0: half the cores, 2 to 4).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Aaron Kimbrell
2026-09-26 19:10:18 -05:00
parent 97ac7d04ef
commit bab092e5ee
21 changed files with 1309 additions and 176 deletions

View File

@@ -1,5 +1,6 @@
set(DWEB_SOURCES
"Web.cpp")
"Web.cpp"
"DeferredReply.cpp")
add_library(dWeb STATIC ${DWEB_SOURCES})

69
dWeb/DeferredReply.cpp Normal file
View File

@@ -0,0 +1,69 @@
#include "DeferredReply.h"
#include <unordered_map>
void DeferredReply::Send(HTTPReply reply) const {
if (!m_State || m_State->answered.exchange(true) || m_State->cancelled.load() || !m_State->queue) return;
reply.deferred.reset();
m_State->queue->Push(*m_State, std::move(reply));
}
std::shared_ptr<DeferredState> DeferredQueue::Begin(unsigned long connection) {
auto state = std::make_shared<DeferredState>();
state->connection = connection;
state->id = m_NextId++;
state->queue = this;
// A connection answers one request at a time, so an older entry is a request that is no longer waited on
if (const auto it = m_Pending.find(connection); it != m_Pending.end()) it->second.state->cancelled = true;
m_Pending[connection] = Waiting{ state, {}, false };
return state;
}
void DeferredQueue::SetReplyOptions(unsigned long connection, std::vector<std::string> headers, bool close) {
const auto it = m_Pending.find(connection);
if (it == m_Pending.end()) return;
it->second.headers = std::move(headers);
it->second.close = close;
}
void DeferredQueue::Close(unsigned long connection) {
const auto it = m_Pending.find(connection);
if (it == m_Pending.end()) return;
it->second.state->cancelled = true;
m_Pending.erase(it);
}
void DeferredQueue::CancelAll() {
for (auto& [connection, waiting] : m_Pending) waiting.state->cancelled = true;
m_Pending.clear();
std::lock_guard lock(m_Mutex);
m_Done.clear();
}
std::vector<DeferredQueue::Finished> DeferredQueue::Drain() {
std::vector<std::pair<uint64_t, HTTPReply>> done;
{
std::lock_guard lock(m_Mutex);
done.swap(m_Done);
}
std::vector<Finished> finished;
if (done.empty()) return finished;
std::unordered_map<uint64_t, unsigned long> connectionOf;
for (const auto& [connection, waiting] : m_Pending) connectionOf.emplace(waiting.state->id, connection);
for (auto& [id, reply] : done) {
const auto it = connectionOf.find(id);
if (it == connectionOf.end()) continue; // the client went away first
auto waiting = m_Pending.find(it->second);
auto headers = std::move(waiting->second.headers);
headers.insert(headers.end(), std::make_move_iterator(reply.headers.begin()), std::make_move_iterator(reply.headers.end()));
reply.headers = std::move(headers);
finished.push_back({ it->second, std::move(reply), waiting->second.close });
m_Pending.erase(waiting);
}
return finished;
}
void DeferredQueue::Push(const DeferredState& state, HTTPReply reply) {
std::lock_guard lock(m_Mutex);
m_Done.emplace_back(state.id, std::move(reply));
}

93
dWeb/DeferredReply.h Normal file
View File

@@ -0,0 +1,93 @@
#pragma once
#include <atomic>
#include <cstdint>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <vector>
#include "HTTPReply.h"
class DeferredQueue;
// Shared between the web thread and whoever answers a deferred request
struct DeferredState {
unsigned long connection{};
uint64_t id{}; // unique per request, so a late answer never reaches a later request
DeferredQueue* queue{};
std::atomic<bool> cancelled{};
std::atomic<bool> answered{};
};
/**
* A request a route answers later, from any thread: for slow work (converting a big model) that would otherwise hold
* up every other request, since the web server answers requests one at a time on one thread. Get one with
* Web::Defer, hand it to a worker, and Send the reply when the work is done; the web thread sends it on its next
* poll. When the client has gone first, Cancelled() turns true and the reply is dropped.
*/
class DeferredReply {
public:
DeferredReply() = default;
explicit DeferredReply(std::shared_ptr<DeferredState> state) : m_State(std::move(state)) {}
// Answer the request (any thread). Only the first answer counts; it is dropped when the client has gone.
void Send(HTTPReply reply) const;
// Whether the client has gone (closed the connection, or the server is stopping), so the work can be skipped
bool Cancelled() const { return !m_State || m_State->cancelled.load(); }
explicit operator bool() const { return m_State != nullptr; }
private:
std::shared_ptr<DeferredState> m_State;
};
/**
* Deferred requests waiting for their answers, and the answers that have arrived. Everything but Push (which
* DeferredReply::Send calls from any thread) is for the web thread. No sockets, so it can be unit tested.
*/
class DeferredQueue {
public:
struct Finished {
unsigned long connection{};
HTTPReply reply;
bool close{}; // the client asked for Connection: close
};
// The request on `connection` will be answered later
std::shared_ptr<DeferredState> Begin(unsigned long connection);
// Headers the reply is sent with besides its own (e.g. a session cookie middleware refreshed), and whether the
// connection closes after it
void SetReplyOptions(unsigned long connection, std::vector<std::string> headers, bool close);
// The connection closed (or the request was answered another way): a late answer is dropped
void Close(unsigned long connection);
// Cancel every pending request (the server is stopping)
void CancelAll();
// The answers that arrived for requests still pending; those requests stop being pending
std::vector<Finished> Drain();
bool IsPending(unsigned long connection) const { return m_Pending.contains(connection); }
size_t Pending() const { return m_Pending.size(); }
// DeferredReply::Send's half: any thread
void Push(const DeferredState& state, HTTPReply reply);
private:
struct Waiting {
std::shared_ptr<DeferredState> state;
std::vector<std::string> headers;
bool close{};
};
std::map<unsigned long, Waiting> m_Pending; // web thread only
uint64_t m_NextId{ 1 };
std::mutex m_Mutex;
std::vector<std::pair<uint64_t, HTTPReply>> m_Done; // guarded by m_Mutex, keyed by DeferredState::id
};

View File

@@ -27,6 +27,7 @@ struct HTTPContext {
// Client information
std::string clientIP{};
unsigned long connectionId = 0; // the web server's id of the connection (for Web::Defer)
// Authentication information (populated by auth middleware)
bool isAuthenticated = false;

37
dWeb/HTTPReply.h Normal file
View File

@@ -0,0 +1,37 @@
#pragma once
#include <memory>
#include <string>
#include <vector>
#include "eHTTPStatusCode.h"
// Content type enum for HTTP responses
enum class eContentType {
APPLICATION_JSON,
TEXT_HTML,
TEXT_CSS,
TEXT_JAVASCRIPT,
TEXT_PLAIN,
TEXT_CSV,
IMAGE_PNG,
IMAGE_JPEG,
APPLICATION_OCTET_STREAM,
TEXT_PROMETHEUS // the Prometheus text exposition format
};
struct DeferredState;
// For passing HTTP messages between functions
struct HTTPReply {
eHTTPStatusCode status = eHTTPStatusCode::NOT_FOUND;
std::string message = "{\"error\":\"Not Found\"}";
eContentType contentType = eContentType::APPLICATION_JSON;
std::string location = ""; // For redirect responses (Location header)
std::vector<std::string> headers{}; // Extra raw headers, e.g. "Set-Cookie: a=b"
// When set on a 200 reply, this file is streamed from disk as the body (with contentType and headers) instead of
// message, so large downloads never sit in memory
std::string file{};
// Set by Web::Defer: the handler answers later, from another thread (DeferredReply), so nothing is sent now
std::shared_ptr<DeferredState> deferred{};
};

View File

@@ -72,6 +72,9 @@ namespace {
// Global middleware applied to all routes
std::vector<MiddlewarePtr> g_GlobalMiddleware;
// Requests answered later by Web::Defer
DeferredQueue g_Deferred;
// Helper to extract client IP from mongoose connection
static std::string GetClientIP(mg_connection* connection) {
@@ -124,6 +127,7 @@ namespace {
// Get client IP
context.clientIP = GetClientIP(connection);
context.connectionId = connection ? connection->id : 0;
}
const char* ContentTypeToString(eContentType contentType) {
@@ -180,6 +184,53 @@ namespace {
}
}
// Send a reply; http_msg is the request (null for a deferred reply, whose request is gone)
static void SendReply(mg_connection* connection, const HTTPReply& reply, const mg_http_message* http_msg) {
// Build headers
std::string headers = std::string("Content-Type: ") + ContentTypeToString(reply.contentType) + "\r\n";
if (!reply.location.empty()) {
headers += "Location: " + reply.location + "\r\n";
}
// A route's own header replaces a default header of the same name
const auto headerName = [](const std::string& header) {
std::string name = header.substr(0, header.find(':'));
std::transform(name.begin(), name.end(), name.begin(), ::tolower);
return name;
};
for (const auto& header : Game::web.GetDefaultHeaders()) {
const auto name = headerName(header);
if (std::ranges::none_of(reply.headers, [&](const std::string& h) { return headerName(h) == name; })) headers += header + "\r\n";
}
for (const auto& header : reply.headers) headers += header + "\r\n";
if (!reply.file.empty() && reply.status == eHTTPStatusCode::OK) {
// Streamed in chunks by mongoose. Content-Type comes from the mime override (it adds its own header).
std::string extraHeaders = headers.substr(headers.find("\r\n") + 2);
const std::string mimeTypes = std::string("*=") + ContentTypeToString(reply.contentType);
mg_http_serve_opts opts{};
opts.extra_headers = extraHeaders.c_str();
opts.mime_types = mimeTypes.c_str();
// Without Accept-Encoding, so a stale "<file>.gz" next to the file is never served instead
mg_http_message request{};
if (http_msg) request = *http_msg;
else request.method = mg_str("GET");
for (auto& header : request.headers) {
if (header.name.len && mg_strcasecmp(header.name, mg_str("Accept-Encoding")) == 0) header.name = mg_str("X-Ignored");
}
mg_http_serve_file(connection, &request, reply.file.c_str(), &opts);
return;
}
// Written by hand rather than with mg_http_reply: that pads Content-Length with spaces (it fills the number in
// afterwards), which strict clients such as Node's fetch reject, and it can't send binary bodies
headers += "Content-Length: " + std::to_string(reply.message.size()) + "\r\n";
const auto status = static_cast<int>(reply.status);
std::string resp = "HTTP/1.1 " + std::to_string(status) + " " + ReasonPhrase(status) + "\r\n" + headers + "\r\n";
mg_send(connection, resp.data(), resp.size());
mg_send(connection, reply.message.data(), reply.message.size());
connection->is_resp = 0;
}
void HandleHTTPMessage(mg_connection* connection, const mg_http_message* http_msg) {
if (g_HTTPRoutes.empty()) return;
@@ -394,47 +445,16 @@ void HandleHTTPMessage(mg_connection* connection, const mg_http_message* http_ms
}
}
// Build headers
std::string headers = std::string("Content-Type: ") + ContentTypeToString(reply.contentType) + "\r\n";
if (!reply.location.empty()) {
headers += "Location: " + reply.location + "\r\n";
}
// A route's own header replaces a default header of the same name
const auto headerName = [](const std::string& header) {
std::string name = header.substr(0, header.find(':'));
std::transform(name.begin(), name.end(), name.begin(), ::tolower);
return name;
};
for (const auto& header : Game::web.GetDefaultHeaders()) {
const auto name = headerName(header);
if (std::ranges::none_of(reply.headers, [&](const std::string& h) { return headerName(h) == name; })) headers += header + "\r\n";
}
for (const auto& header : reply.headers) headers += header + "\r\n";
if (!reply.file.empty() && reply.status == eHTTPStatusCode::OK && http_msg) {
// Streamed in chunks by mongoose. Content-Type comes from the mime override (it adds its own header).
std::string extraHeaders = headers.substr(headers.find("\r\n") + 2);
const std::string mimeTypes = std::string("*=") + ContentTypeToString(reply.contentType);
mg_http_serve_opts opts{};
opts.extra_headers = extraHeaders.c_str();
opts.mime_types = mimeTypes.c_str();
// Without Accept-Encoding, so a stale "<file>.gz" next to the file is never served instead
mg_http_message request = *http_msg;
for (auto& header : request.headers) {
if (header.name.len && mg_strcasecmp(header.name, mg_str("Accept-Encoding")) == 0) header.name = mg_str("X-Ignored");
}
mg_http_serve_file(connection, &request, reply.file.c_str(), &opts);
if (reply.deferred) {
// Answered later (Web::Defer): the connection keeps is_resp set, so mongoose reads no further request on it
const auto* cc = http_msg ? mg_http_get_header(const_cast<mg_http_message*>(http_msg), "Connection") : nullptr;
g_Deferred.SetReplyOptions(connection->id, reply.headers, cc && mg_strcasecmp(*cc, mg_str("close")) == 0);
return;
}
// The handler deferred and then failed: its late answer is dropped
if (g_Deferred.IsPending(connection->id)) g_Deferred.Close(connection->id);
// Written by hand rather than with mg_http_reply: that pads Content-Length with spaces (it fills the number in
// afterwards), which strict clients such as Node's fetch reject, and it can't send binary bodies
headers += "Content-Length: " + std::to_string(reply.message.size()) + "\r\n";
const auto status = static_cast<int>(reply.status);
std::string resp = "HTTP/1.1 " + std::to_string(status) + " " + ReasonPhrase(status) + "\r\n" + headers + "\r\n";
mg_send(connection, resp.data(), resp.size());
mg_send(connection, reply.message.data(), reply.message.size());
connection->is_resp = 0;
SendReply(connection, reply, http_msg);
}
@@ -554,6 +574,8 @@ void HandleMessages(mg_connection* connection, int message, void* message_data)
break;
case MG_EV_CLOSE:
g_AuthenticatedWSConnections.erase(connection);
// A deferred answer that comes after this is dropped
g_Deferred.Close(connection->id);
break;
default:
break;
@@ -638,7 +660,21 @@ Web::Web() {
}
Web::~Web() {
// Static destruction: the maps and queues the close events touch are gone by now, so the handlers are off
// (HandleMessages checks enabled). Servers call Shutdown first, while they're still there.
enabled = false;
if (!managerFreed) mg_mgr_free(&mgr);
managerFreed = true;
}
void Web::Shutdown() {
if (managerFreed) return;
// Closing the connections fires their close events, which still clean up (WebSocket clients, deferred requests)
g_Deferred.CancelAll();
mg_mgr_free(&mgr);
managerFreed = true;
enabled = false;
g_AuthenticatedWSConnections.clear();
}
bool Web::Startup(const std::string& listen_ip, const uint32_t listen_port) {
@@ -677,9 +713,30 @@ bool Web::Startup(const std::string& listen_ip, const uint32_t listen_port) {
void Web::ReceiveRequests(int timeoutMs) {
mg_mgr_poll(&mgr, timeoutMs);
SendDeferredReplies();
RecheckDueWebSockets();
}
void Web::SendDeferredReplies() {
for (auto& finished : g_Deferred.Drain()) {
mg_connection* connection = mgr.conns;
while (connection && connection->id != finished.connection) connection = connection->next;
if (!connection || connection->is_closing) continue;
// Clears is_resp once the reply is out, so mongoose reads the connection's next request again
SendReply(connection, finished.reply, nullptr);
if (finished.close) connection->is_draining = 1;
}
}
DeferredReply Web::Defer(HTTPReply& reply, const HTTPContext& context) {
reply.deferred = g_Deferred.Begin(context.connectionId);
return DeferredReply(reply.deferred);
}
size_t Web::PendingDeferred() const {
return g_Deferred.Pending();
}
void Web::RecheckWebSockets(uint32_t accountId) {
for (auto& [connection, client] : g_AuthenticatedWSConnections) {
if (accountId == 0 || client.accountId == accountId) client.nextCheck = {};

View File

@@ -10,6 +10,8 @@
#include "json_fwd.hpp"
#include "eHTTPStatusCode.h"
#include "HTTPContext.h"
#include "HTTPReply.h"
#include "DeferredReply.h"
#include "IHTTPMiddleware.h"
// Forward declarations for game namespace
@@ -24,32 +26,6 @@ enum class eHTTPMethod;
// Forward declaration for mongoose manager
typedef struct mg_mgr mg_mgr;
// Content type enum for HTTP responses
enum class eContentType {
APPLICATION_JSON,
TEXT_HTML,
TEXT_CSS,
TEXT_JAVASCRIPT,
TEXT_PLAIN,
TEXT_CSV,
IMAGE_PNG,
IMAGE_JPEG,
APPLICATION_OCTET_STREAM,
TEXT_PROMETHEUS // the Prometheus text exposition format
};
// For passing HTTP messages between functions
struct HTTPReply {
eHTTPStatusCode status = eHTTPStatusCode::NOT_FOUND;
std::string message = "{\"error\":\"Not Found\"}";
eContentType contentType = eContentType::APPLICATION_JSON;
std::string location = ""; // For redirect responses (Location header)
std::vector<std::string> headers{}; // Extra raw headers, e.g. "Set-Cookie: a=b"
// When set on a 200 reply, this file is streamed from disk as the body (with contentType and headers) instead of
// message, so large downloads never sit in memory
std::string file{};
};
// HTTP route structure
// This structure is used to register HTTP routes
// with the server. Each route has a path, method, optional middleware,
@@ -109,6 +85,14 @@ public:
void RegisterWSSubscription(const std::string& subscription, uint8_t minLevel = 0);
// The level is looked up each time (for permissions that can change while running)
void RegisterWSSubscription(const std::string& subscription, std::function<uint8_t()> minLevel);
/**
* Answer this request later (from any thread) instead of when the handler returns, for slow work that would hold
* up every other request: the handler hands the returned DeferredReply to a worker and returns; the web thread
* sends the worker's reply on its next poll. Headers middleware set on `reply` are sent with it.
*/
static DeferredReply Defer(HTTPReply& reply, const HTTPContext& context);
// Deferred requests still waiting for their answers
size_t PendingDeferred() const;
// Add global middleware that applies to all routes
void AddGlobalMiddleware(MiddlewarePtr middleware);
// Set WebSocket authentication callback for token validation
@@ -117,6 +101,12 @@ public:
void SetWSApiAccessCallback(std::function<bool(uint8_t)> callback) { wsApiAccessCallback = std::move(callback); }
// Returns if the web server is enabled
bool IsEnabled() const { return enabled; };
/**
* Close every connection and stop listening, before the server's other state is torn down (the destructor runs
* during static destruction, when what the connections' close events touch may be gone). Deferred requests still
* waiting are cancelled; stop the workers that answer them first.
*/
void Shutdown();
// Send a message to all connected WebSocket clients that are subscribed to the given topic
void static SendWSMessage(std::string sub, nlohmann::json& message);
// Send a message on a topic only to the subscribed connections of one account
@@ -133,10 +123,14 @@ public:
WSAuthCallback GetWSAuthCallback() const { return wsAuthCallback; }
const std::function<bool(uint8_t)>& GetWSApiAccessCallback() const { return wsApiAccessCallback; }
private:
// Send the answers of deferred requests that have arrived
void SendDeferredReplies();
// mongoose manager
mg_mgr mgr;
// If the web server is enabled
bool enabled = false;
// mg_mgr_free has run (Shutdown)
bool managerFreed = false;
// WebSocket authentication callback
WSAuthCallback wsAuthCallback = nullptr;
std::function<bool(uint8_t)> wsApiAccessCallback = nullptr;