DragonX/ObsidianDragon
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity
Files
7b6e8a8a8e989b16ba974d2bdcbd0b2ecda7b224
ObsidianDragon/tools/fake_dragonxd/fake_dragonxd.cpp
T
DanSandClaude Opus 5.5 7b6e8a8a8e test(daemon): add fake_dragonxd, a slow dragonxd stand-in for stop/kill testing 
A scriptable fake dragonxd that reproduces, in minutes, the incident where
the GUI hard-killed a node busy with an hour-long witness rebuild:

- dragonxd's HTTP-RPC queue (8 workers + 16 queued, -rpcthreads /
  -rpcworkqueue honoured): a full queue answers an immediate plain-text
  500 "Work queue depth exceeded", and a request whose client gave up
  keeps its slot until a worker reaches it
- a global cs_main held by a "witness rebuild" that logs the real
  markers; every method except stop/help takes it
- warmup -28 with the daemon's allow-list, an optional startup replay
- dragonxd's shutdown order: stop (or SIGTERM on POSIX) unbinds RPC,
  503s open connections, waits for the rebuild, then logs
  "Shutdown: In progress..." / "Shutdown: done" and exits 0

Everything is appended to <datadir>/fake_dragonxd.events; a missing
"clean-exit" line means the process was killed. One portable C++17 file
(POSIX + Winsock), OFF by default (-DDRAGONX_BUILD_FAKE_DAEMON=ON), built
into <build>/fake_dragonxd/ so it can't shadow the real bin/dragonxd.
run-gui-with-fake.sh runs the GUI against it in a throwaway HOME.

Against master's GUI on Linux it reproduces the bug: closing during a
rebuild queues stop behind cs_main-blocked workers, then the GUI sends
SIGTERM and SIGKILL mid-rebuild, with no clean exit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-01 00:24:42 -05:00

1307 lines
50 KiB
C++
Raw Blame History

// fake_dragonxd — a deliberately slow, scriptable stand-in for dragonxd (test tooling only).
//
// Reproduces, in minutes and without a 13k-note wallet, the conditions behind the 2026-09 "GUI
// hard-kills a busy node" incident (see README.md):
// * dragonxd's HTTP-RPC queue: N workers + a bounded queue (8 + 16 by default, -rpcthreads /
// -rpcworkqueue honoured). A full queue gets an immediate plain-text HTTP 500 "Work queue depth
// exceeded" before any handler runs, and a request whose client gave up keeps its slot until a
// worker reaches it (libevent has no close callback);
// * a global cs_main that a "witness rebuild" holds for minutes, logging the real markers. Every
// method except the lock-free `stop` / `help` takes it;
// * dragonxd's shutdown: `stop` (or SIGTERM/SIGINT on POSIX; Windows has no signal channel, as in
// v1.3.0) unbinds the listener at once, answers new requests on open connections with 503, then
// waits for the rebuild (Phase 2 is not abortable) before "Shutdown: In progress..." -> flush ->
// "Shutdown: done" -> exit 0.
// Everything it does is appended to <datadir>/fake_dragonxd.events, so a test can tell a clean exit
// ("clean-exit") from a hard kill (no "clean-exit"). Knobs are FAKE_* environment variables; see
// README.md. Built only with -DDRAGONX_BUILD_FAKE_DAEMON=ON and never shipped.
#include <nlohmann/json.hpp>
#include <algorithm>
#include <atomic>
#include <cctype>
#include <chrono>
#include <condition_variable>
#include <csignal>
#include <cstdarg>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <deque>
#include <filesystem>
#include <fstream>
#include <map>
#include <memory>
#include <mutex>
#include <random>
#include <string>
#include <thread>
#include <vector>
#ifdef _WIN32
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif
# ifndef NOMINMAX
# define NOMINMAX
# endif
# include <winsock2.h>
# include <ws2tcpip.h>
# include <windows.h>
using socket_t = SOCKET;
static const socket_t kBadSocket = INVALID_SOCKET;
static void closeSocket(socket_t s) { closesocket(s); }
#else
# include <arpa/inet.h>
# include <fcntl.h>
# include <netinet/in.h>
# include <sys/select.h>
# include <sys/socket.h>
# include <unistd.h>
# ifdef __linux__
# include <sys/prctl.h>
# endif
using socket_t = int;
static const socket_t kBadSocket = -1;
static void closeSocket(socket_t s) { ::close(s); }
#endif
#ifndef MSG_NOSIGNAL
# define MSG_NOSIGNAL 0 // Windows/macOS: SIGPIPE is ignored process-wide instead (POSIX) or absent
#endif
namespace fs = std::filesystem;
using json = nlohmann::json;
using Clock = std::chrono::steady_clock;
using namespace std::chrono_literals;
namespace {
// =============================================================================================
// Configuration
// =============================================================================================
double envDouble(const char* name, double def)
{
const char* v = std::getenv(name);
if (!v || !*v) return def;
char* end = nullptr;
const double d = std::strtod(v, &end);
return (end && *end == '\0') ? d : def;
}
long envLong(const char* name, long def) { return static_cast<long>(envDouble(name, static_cast<double>(def))); }
struct Config {
fs::path datadir;
bool printToConsole = false;
int rpcPort = 21769;
int rpcThreads = 8; // DEFAULT_HTTP_THREADS
int rpcWorkQueue = 16; // DEFAULT_HTTP_WORKQUEUE
std::string rpcUser, rpcPassword;
// FAKE_* knobs. Times are seconds.
double warmupSecs = 3; // -28 "Loading block index..." before anything else
double replaySecs = 0; // -28 "Activating best chain..." with UpdateTip lines (startup replay)
double replayBlocksPerSec = 12;
double rebuildDelaySecs = 20; // after "Done Loading", so the GUI connects before the rebuild
double rebuildSecs = 300; // how long the rebuild holds cs_main; 0 = no rebuild
double phase1Secs = 2; // the abortable, logged part of the rebuild
double postRebuildLockSecs = 0; // cs_main held after the end line (periodic flush + SetBestChain)
double flushSecs = 3; // "Shutdown: In progress..." -> "Shutdown: done"
double updateTipSecs = 36; // block interval once loaded
double noiseSecs = 20; // an unrelated log line this often (defeats tail heuristics)
double serverTimeoutSecs = 30; // -rpcservertimeout: idle keep-alive connections are closed
long prefill = 0; // phantom abandoned requests injected when the rebuild starts
long height = 3327145;
long rebuildTotal = 62583;
long notes = 13632;
long version = 1030050; // getinfo "version" (real v1.3.0 reports 1030050)
bool stuckShutdown = false; // the Sep 29 orphan: datadir lock held, no listener, never exits
bool mining = false; // internal miner on: `stop` must wait for cs_main first
};
Config g_cfg;
std::map<std::string, std::string> parseArgs(int argc, char** argv)
{
std::map<std::string, std::string> out;
for (int i = 1; i < argc; ++i) {
std::string a = argv[i];
const size_t start = a.find_first_not_of('-');
if (start == 0 || start == std::string::npos) continue;
a = a.substr(start);
const size_t eq = a.find('=');
if (eq == std::string::npos) out[a] = "1";
else out[a.substr(0, eq)] = a.substr(eq + 1);
}
return out;
}
std::map<std::string, std::string> parseConf(const fs::path& path)
{
std::map<std::string, std::string> out;
std::ifstream f(path);
std::string line;
while (std::getline(f, line)) {
if (!line.empty() && line.back() == '\r') line.pop_back();
const size_t hash = line.find('#');
if (hash != std::string::npos) line.erase(hash);
const size_t eq = line.find('=');
if (eq == std::string::npos) continue;
auto trim = [](std::string s) {
const size_t b = s.find_first_not_of(" \t");
const size_t e = s.find_last_not_of(" \t");
return b == std::string::npos ? std::string() : s.substr(b, e - b + 1);
};
out[trim(line.substr(0, eq))] = trim(line.substr(eq + 1));
}
return out;
}
fs::path defaultDatadir()
{
#ifdef _WIN32
const char* appdata = std::getenv("APPDATA");
return fs::path(appdata ? appdata : ".") / "Hush" / "DRAGONX";
#else
const char* home = std::getenv("HOME");
const fs::path h = home ? home : ".";
# ifdef __APPLE__
return h / "Library" / "Application Support" / "Hush" / "DRAGONX";
# else
return h / ".hush" / "DRAGONX";
# endif
#endif
}
// =============================================================================================
// Time, logging, events
// =============================================================================================
long long unixMs()
{
return std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch()).count();
}
std::string utcStamp(std::time_t t)
{
std::tm tm{};
#ifdef _WIN32
gmtime_s(&tm, &t);
#else
gmtime_r(&t, &tm);
#endif
char buf[32];
std::strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M:%S", &tm);
return buf;
}
std::string vformat(const char* fmt, va_list ap)
{
char buf[2048];
std::vsnprintf(buf, sizeof(buf), fmt, ap);
return buf;
}
// dragonxd's LogPrintf: with -printtoconsole every line goes to stdout INSTEAD of debug.log and
// without a timestamp; otherwise to <datadir>/debug.log with a "YYYY-MM-DD HH:MM:SS " UTC prefix.
class Logger {
public:
void open(const fs::path& datadir, bool console)
{
console_ = console;
if (!console_) file_ = std::fopen((datadir / "debug.log").string().c_str(), "ab");
}
void line(const std::string& msg)
{
std::lock_guard<std::mutex> lk(mx_);
if (console_) {
std::fputs((msg + "\n").c_str(), stdout); // EPIPE after the reader left is ignored, like dragonxd
std::fflush(stdout);
} else if (file_) {
std::fprintf(file_, "%s %s\n", utcStamp(std::time(nullptr)).c_str(), msg.c_str());
std::fflush(file_);
}
}
void close()
{
std::lock_guard<std::mutex> lk(mx_);
if (file_) std::fclose(file_);
file_ = nullptr;
}
private:
std::mutex mx_;
FILE* file_ = nullptr;
bool console_ = false;
};
Logger g_log;
void logf(const char* fmt, ...)
{
va_list ap;
va_start(ap, fmt);
const std::string s = vformat(fmt, ap);
va_end(ap);
g_log.line(s);
}
// <datadir>/fake_dragonxd.events: one "<name> t=<unix ms> [k=v ...]" line per event, for tests.
std::mutex g_eventsMx;
FILE* g_events = nullptr;
void event(const std::string& name, const std::string& extra = "")
{
std::lock_guard<std::mutex> lk(g_eventsMx);
if (!g_events) return;
std::fprintf(g_events, "%s t=%lld%s%s\n", name.c_str(), unixMs(), extra.empty() ? "" : " ", extra.c_str());
std::fflush(g_events);
}
// =============================================================================================
// Shared daemon state
// =============================================================================================
std::mutex g_csMain; // the daemon's cs_main
std::atomic<long> g_height{0};
std::atomic<bool> g_warmup{true};
std::mutex g_warmupMx;
std::string g_warmupStatus = "Loading block index...";
std::atomic<bool> g_minerOn{false};
enum ShutdownReason { kNoShutdown = 0, kRpcStop = 1, kSignal = 2 };
std::atomic<int> g_shutdownReason{kNoShutdown};
std::atomic<int> g_stopCount{0};
volatile std::sig_atomic_t g_signalled = 0;
extern "C" void onTermSignal(int) { g_signalled = 1; }
void requestShutdown(ShutdownReason reason)
{
if (reason == kRpcStop) {
const int n = ++g_stopCount;
event("stop-received", "n=" + std::to_string(n));
}
int expected = kNoShutdown;
if (g_shutdownReason.compare_exchange_strong(expected, reason) && reason == kSignal) event("sigterm");
}
bool shutdownRequested()
{
if (g_signalled && g_shutdownReason.load() == kNoShutdown) requestShutdown(kSignal);
return g_shutdownReason.load() != kNoShutdown;
}
// Sleeps up to `secs`; returns false (early) once a shutdown is requested.
bool sleepUnlessShutdown(double secs)
{
const auto end = Clock::now() + std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(secs));
while (Clock::now() < end) {
if (shutdownRequested()) return false;
std::this_thread::sleep_for(50ms);
}
return !shutdownRequested();
}
void setInitMessage(const std::string& msg)
{
logf("init message: %s", msg.c_str());
std::lock_guard<std::mutex> lk(g_warmupMx);
g_warmupStatus = msg;
}
std::string warmupStatus()
{
std::lock_guard<std::mutex> lk(g_warmupMx);
return g_warmupStatus;
}
struct Stats {
std::atomic<long> requests{0}, queueFull{0}, unavailable{0}, unhandled{0}, unauthorized{0};
std::atomic<int> abandonedNow{0}, maxAbandoned{0}, maxSlots{0};
};
Stats g_stats;
bool raiseMax(std::atomic<int>& max, int value)
{
int cur = max.load();
while (value > cur) {
if (max.compare_exchange_weak(cur, value)) return true;
}
return false;
}
void noteAbandoned(int delta)
{
const int now = (g_stats.abandonedNow += delta);
if (delta > 0 && raiseMax(g_stats.maxAbandoned, now)) event("max-abandoned", "n=" + std::to_string(now));
}
std::string statsLine()
{
return "requests=" + std::to_string(g_stats.requests.load()) +
" queue_full_500=" + std::to_string(g_stats.queueFull.load()) +
" unavailable_503=" + std::to_string(g_stats.unavailable.load()) +
" unhandled_500=" + std::to_string(g_stats.unhandled.load()) +
" unauthorized_401=" + std::to_string(g_stats.unauthorized.load()) +
" max_slots=" + std::to_string(g_stats.maxSlots.load()) +
" max_abandoned=" + std::to_string(g_stats.maxAbandoned.load()) +
" abandoned_now=" + std::to_string(g_stats.abandonedNow.load()) +
" stops=" + std::to_string(g_stopCount.load());
}
// =============================================================================================
// Canned RPC results
// =============================================================================================
std::uint64_t splitmix64(std::uint64_t x)
{
x += 0x9E3779B97F4A7C15ULL;
x = (x ^ (x >> 30)) * 0xBF58476D1CE4E5B9ULL;
x = (x ^ (x >> 27)) * 0x94D049BB133111EBULL;
return x ^ (x >> 31);
}
std::string fakeHash(long h)
{
char buf[80];
std::snprintf(buf, sizeof(buf), "%016llx%016llx%016llx%016llx",
(unsigned long long)splitmix64(h * 4 + 0), (unsigned long long)splitmix64(h * 4 + 1),
(unsigned long long)splitmix64(h * 4 + 2), (unsigned long long)splitmix64(h * 4 + 3));
return buf;
}
const char* const kFakeZAddr =
"zs1fakedragonxd0000000000000000000000000000000000000000000000000000000000000";
const char* const kFakeTAddr = "RFakeDragonxdTransparent0000000000";
struct Canned {
bool known = true;
json result;
int errCode = 0;
std::string errMsg;
};
Canned cannedReply(const std::string& m, const json& params)
{
Canned c;
const long h = g_height.load();
const long long now = unixMs() / 1000;
const double difficulty = 152341.734;
const std::string sub = "/DragonX:" + std::to_string(g_cfg.version / 1000000) + "." +
std::to_string(g_cfg.version / 10000 % 100) + "." +
std::to_string(g_cfg.version / 100 % 100) + "/";
auto param0 = [&]() -> std::string {
if (params.empty()) return "";
return params[0].is_string() ? params[0].get<std::string>() : params[0].dump();
};
if (m == "getinfo") {
c.result = {{"version", g_cfg.version}, {"protocolversion", 170012}, {"walletversion", 60000},
{"balance", 0.0}, {"blocks", h}, {"longestchain", h}, {"notarized", 0},
{"tiptime", now}, {"difficulty", difficulty}, {"keypoololdest", now - 86400},
{"keypoolsize", 101}, {"paytxfee", 0.0}, {"timeoffset", 0}, {"connections", 2},
{"tls_connections", 2}, {"proxy", ""}, {"testnet", false}, {"relayfee", 0.000001},
{"errors", ""}, {"name", "DRAGONX"}, {"p2pport", 21768}, {"rpcport", g_cfg.rpcPort}};
} else if (m == "getblockchaininfo") {
c.result = {{"chain", "main"}, {"blocks", h}, {"synced", true}, {"headers", h},
{"bestblockhash", fakeHash(h)}, {"difficulty", difficulty}, {"verificationprogress", 1.0},
{"chainwork", std::string(48, '0') + "1f3a5c7e9b2d4f60"}, {"pruned", false},
{"valuePools", json::array()}, {"softforks", json::array()}, {"upgrades", json::object()},
{"longestchain", h}, {"notarized", 0}};
} else if (m == "getblockcount") {
c.result = h;
} else if (m == "getbestblockhash") {
c.result = fakeHash(h);
} else if (m == "getblockhash") {
c.result = fakeHash(params.empty() || !params[0].is_number() ? h : params[0].get<long>());
} else if (m == "getblock") {
long bh = h;
const std::string p = param0();
if (!p.empty() && p.size() < 12) bh = std::strtol(p.c_str(), nullptr, 10);
c.result = {{"hash", fakeHash(bh)}, {"confirmations", std::max(1L, h - bh + 1)}, {"height", bh},
{"version", 4}, {"time", now - (h - bh) * 36}, {"difficulty", difficulty},
{"previousblockhash", fakeHash(bh - 1)}, {"tx", json::array({fakeHash(bh + 7777777)})},
{"size", 1765}};
} else if (m == "getwalletinfo") {
c.result = {{"walletversion", 60000}, {"balance", 0.0}, {"unconfirmed_balance", 0.0},
{"immature_balance", 0.0}, {"txcount", 42}, {"keypoololdest", now - 86400},
{"keypoolsize", 101}, {"paytxfee", 0.0}};
} else if (m == "z_gettotalbalance") {
c.result = {{"transparent", "0.00"}, {"private", "1234.5678"}, {"total", "1234.5678"}};
} else if (m == "z_getbalance") {
c.result = 1234.5678;
} else if (m == "getbalance") {
c.result = 0.0;
} else if (m == "getmininginfo") {
c.result = {{"blocks", h}, {"currentblocksize", 0}, {"currentblocktx", 0}, {"difficulty", difficulty},
{"errors", ""}, {"genproclimit", -1}, {"localsolps", 0.0}, {"networksolps", 182345},
{"networkhashps", 182345}, {"pooledtx", 0}, {"testnet", false}, {"chain", "main"},
{"generate", g_minerOn.load()}, {"numthreads", -1}};
} else if (m == "getlocalsolps") {
c.result = 0.0;
} else if (m == "getnetworkhashps" || m == "getnetworksolps") {
c.result = 182345;
} else if (m == "setgenerate") {
c.result = nullptr; // the miner flag is flipped in execOne: this table stays side-effect free
} else if (m == "getgenerate") {
c.result = g_minerOn.load();
} else if (m == "getpeerinfo") {
// Documentation-range (TEST-NET) addresses only.
json peers = json::array();
const char* addrs[] = {"203.0.113.10:21768", "198.51.100.23:21768"};
for (int i = 0; i < 2; ++i) {
peers.push_back({{"id", i + 1}, {"addr", addrs[i]}, {"services", "0000000000000005"},
{"tls_verified", false}, {"lastsend", now}, {"lastrecv", now},
{"bytessent", 123456}, {"bytesrecv", 654321}, {"conntime", now - 3600},
{"timeoffset", 0}, {"pingtime", 0.052}, {"version", 170012}, {"subver", sub},
{"inbound", false}, {"startingheight", h - 100}, {"banscore", 0},
{"synced_headers", h}, {"synced_blocks", h}, {"inflight", json::array()},
{"whitelisted", false}});
}
c.result = peers;
} else if (m == "getnetworkinfo") {
c.result = {{"version", g_cfg.version}, {"subversion", sub}, {"protocolversion", 170012},
{"localservices", "0000000000000005"}, {"timeoffset", 0}, {"connections", 2},
{"tls_connections", 2}, {"networks", json::array()}, {"relayfee", 0.000001},
{"localaddresses", json::array()}, {"warnings", ""}};
} else if (m == "getconnectioncount") {
c.result = 2;
} else if (m == "getmempoolinfo") {
c.result = {{"size", 0}, {"bytes", 0}, {"usage", 0}};
} else if (m == "listbanned" || m == "listunspent" || m == "z_listunspent" || m == "listtransactions" ||
m == "z_listreceivedbyaddress" || m == "z_listsentbyaddress" || m == "z_getoperationstatus" ||
m == "z_getoperationresult") {
c.result = json::array();
} else if (m == "setban" || m == "clearbanned") {
c.result = nullptr;
} else if (m == "z_listaddresses") {
c.result = json::array({kFakeZAddr});
} else if (m == "getaddressesbyaccount" || m == "listaddresses") {
c.result = json::array({kFakeTAddr});
} else if (m == "getrescaninfo") {
c.result = {{"rescanning", false}};
} else if (m == "validateaddress") {
c.result = {{"isvalid", true}, {"address", param0()}, {"ismine", false}};
} else if (m == "z_validateaddress") {
c.result = {{"isvalid", true}, {"address", param0()}, {"type", "sapling"}, {"ismine", false}};
} else if (m == "z_exportkey" || m == "z_exportmnemonic") {
c.errCode = -4; // RPC_WALLET_ERROR: keeps seed-derived features (chat) quietly off
c.errMsg = "fake_dragonxd holds no keys";
} else {
c.known = false;
}
return c;
}
// dragonxd's warmup allow-list (rpc/server.cpp CRPCTable::execute).
bool allowedInWarmup(const std::string& m)
{
static const char* const kAllowed[] = {
"stop", "help", "z_listaddresses", "z_exportkey", "getNotarizationsForBlock", "scanNotarizationsDB",
"getaddressesbyaccount", "listaddresses", "z_exportwallet", "notaries", "signmessage",
"decoderawtransaction", "dumpprivkey", "getpeerinfo", "getnetworkinfo", "abortrescan"};
for (const char* a : kAllowed)
if (m == a) return true;
return false;
}
std::string helpText(const json& params)
{
if (params.empty() || !params[0].is_string())
return "== Control ==\nhelp ( \"command\" )\nstop\n\n(fake_dragonxd: canned responses only)";
const std::string cmd = params[0].get<std::string>();
if (cmd == "stop") return "stop\n\nStop Hush server.";
if (cmd == "help") return "help ( \"command\" )\n\nList all commands, or get help for a specified command.";
if (cannedReply(cmd, json::array()).known) return cmd + "\n\n(fake_dragonxd canned method)";
return "help: unknown command: " + cmd;
}
json rpcReply(const json& result, const json& error, const json& id)
{
return {{"result", result}, {"error", error}, {"id", id}};
}
// One JSON-RPC call. `status` gets the HTTP status dragonxd would use (httprpc.cpp JSONErrorReply).
json execOne(const json& req, int& status)
{
const json id = req.is_object() && req.contains("id") ? req["id"] : json(nullptr);
auto fail = [&](int code, const std::string& msg) {
status = code == -32600 ? 400 : code == -32601 ? 404 : 500;
return rpcReply(nullptr, {{"code", code}, {"message", msg}}, id);
};
if (!req.is_object() || !req.contains("method") || !req["method"].is_string())
return fail(-32600, "Missing method");
const std::string m = req["method"].get<std::string>();
json params = req.contains("params") ? req["params"] : json::array();
if (!params.is_array()) params = json::array();
status = 200;
const bool lockFree = (m == "stop" || m == "help");
if (!lockFree && !cannedReply(m, params).known) return fail(-32601, "Method not found");
if (g_warmup.load() && !allowedInWarmup(m)) return fail(-28, warmupStatus());
if (m == "help") return rpcReply(helpText(params), nullptr, id);
if (m == "stop") {
// stop's GenerateBitcoins(false) joins the miner threads, which may be waiting on cs_main.
if (g_minerOn.load()) {
std::lock_guard<std::mutex> miners(g_csMain);
g_minerOn = false;
}
requestShutdown(kRpcStop);
return rpcReply("DRAGONX server stopping...", nullptr, id);
}
Canned c;
{
std::lock_guard<std::mutex> lk(g_csMain);
c = cannedReply(m, params);
if (m == "setgenerate") g_minerOn = !params.empty() && params[0].is_boolean() && params[0].get<bool>();
}
if (c.errCode != 0) return fail(c.errCode, c.errMsg);
return rpcReply(c.result, nullptr, id);
}
// =============================================================================================
// HTTP server: N workers + a bounded queue, like dragonxd's httpserver.cpp WorkQueue
// =============================================================================================
const char* const kTextType = "text/html; charset=ISO-8859-1"; // libevent's default for a bare reply
std::string g_expectedAuth; // "Basic <base64(user:pass)>"
std::atomic<bool> g_httpInterrupted{false};
std::string base64(const std::string& in)
{
static const char* const t = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
std::string out;
std::uint32_t val = 0;
int bits = -6;
for (unsigned char c : in) {
val = (val << 8) | c;
bits += 8;
while (bits >= 0) {
out.push_back(t[(val >> bits) & 0x3F]);
bits -= 6;
}
}
if (bits > -6) out.push_back(t[((val << 8) >> (bits + 8)) & 0x3F]);
while (out.size() % 4) out.push_back('=');
return out;
}
struct Item {
bool isPost = true;
bool prefill = false;
std::string body, auth;
std::mutex mx;
std::condition_variable cv;
bool done = false;
int status = 0;
std::string contentType, payload;
};
using ItemPtr = std::shared_ptr<Item>;
void complete(Item& it, int status, const std::string& contentType, const std::string& payload)
{
{
std::lock_guard<std::mutex> lk(it.mx);
it.status = status;
it.contentType = contentType;
it.payload = payload;
it.done = true;
}
it.cv.notify_all();
if (it.prefill) noteAbandoned(-1);
}
void processItem(Item& it)
{
if (!it.isPost) {
complete(it, 405, kTextType, "JSONRPC server handles only POST requests");
return;
}
if (it.auth != g_expectedAuth) {
std::this_thread::sleep_for(250ms); // httprpc.cpp: MilliSleep(250) to deter brute forcing
++g_stats.unauthorized;
complete(it, 401, "", "");
return;
}
const json req = json::parse(it.body, nullptr, false);
int status = 200;
std::string payload;
if (req.is_discarded()) {
status = 500;
payload = rpcReply(nullptr, {{"code", -32700}, {"message", "Parse error"}}, nullptr).dump() + "\n";
} else if (req.is_array()) {
json out = json::array();
for (const auto& r : req) {
int ignored = 200;
out.push_back(execOne(r, ignored));
}
payload = out.dump() + "\n";
} else {
payload = execOne(req, status).dump() + "\n";
}
complete(it, status, "application/json", payload);
}
class WorkQueue {
public:
void configure(size_t maxDepth, int threads)
{
maxDepth_ = maxDepth;
capacity_ = maxDepth + static_cast<size_t>(threads);
}
bool enqueue(const ItemPtr& item)
{
std::lock_guard<std::mutex> lk(mx_);
if (queue_.size() >= maxDepth_) return false;
queue_.push_back(item);
raiseMax(g_stats.maxSlots, static_cast<int>(queue_.size()) + inProgress_);
cv_.notify_one();
return true;
}
size_t slotsInUse()
{
std::lock_guard<std::mutex> lk(mx_);
return queue_.size() + static_cast<size_t>(inProgress_);
}
size_t capacity() const { return capacity_; }
void run()
{
for (;;) {
ItemPtr it;
{
std::unique_lock<std::mutex> lk(mx_);
cv_.wait(lk, [&] { return !running_ || !queue_.empty(); });
if (!running_) return; // queued items are left for drain(), as in dragonxd
it = queue_.front();
queue_.pop_front();
++inProgress_;
}
processItem(*it);
std::lock_guard<std::mutex> lk(mx_);
--inProgress_;
}
}
void interrupt()
{
std::lock_guard<std::mutex> lk(mx_);
running_ = false;
cv_.notify_all();
}
std::deque<ItemPtr> drain()
{
std::lock_guard<std::mutex> lk(mx_);
std::deque<ItemPtr> out;
out.swap(queue_);
return out;
}
private:
std::mutex mx_;
std::condition_variable cv_;
std::deque<ItemPtr> queue_;
bool running_ = true;
int inProgress_ = 0;
size_t maxDepth_ = 16;
size_t capacity_ = 24;
};
WorkQueue g_queue;
bool waitReadable(socket_t s, int timeoutMs)
{
fd_set rf;
FD_ZERO(&rf);
FD_SET(s, &rf);
timeval tv{timeoutMs / 1000, (timeoutMs % 1000) * 1000};
return select(static_cast<int>(s) + 1, &rf, nullptr, nullptr, &tv) > 0;
}
bool sendAll(socket_t s, const std::string& data)
{
size_t off = 0;
while (off < data.size()) {
const int n = send(s, data.data() + off, static_cast<int>(std::min<size_t>(data.size() - off, 1 << 20)), MSG_NOSIGNAL);
if (n <= 0) return false;
off += static_cast<size_t>(n);
}
return true;
}
// The client hung up (e.g. its curl timeout fired) while its request still holds a slot.
bool peerClosed(socket_t s)
{
if (!waitReadable(s, 0)) return false;
char c;
return recv(s, &c, 1, MSG_PEEK) <= 0;
}
const char* reasonPhrase(int status)
{
switch (status) {
case 200: return "OK";
case 400: return "Bad Request";
case 401: return "Unauthorized";
case 404: return "Not Found";
case 405: return "Method Not Allowed";
case 500: return "Internal Server Error";
case 503: return "Service Unavailable";
default: return "Unknown";
}
}
std::string httpReply(int status, const std::string& contentType, const std::string& body, bool close)
{
std::string r = "HTTP/1.1 " + std::to_string(status) + " " + reasonPhrase(status) + "\r\n";
if (!contentType.empty()) r += "Content-Type: " + contentType + "\r\n";
if (status == 401) r += "WWW-Authenticate: Basic realm=\"jsonrpc\"\r\n";
r += "Content-Length: " + std::to_string(body.size()) + "\r\n";
if (close) r += "Connection: close\r\n";
r += "\r\n";
r += body;
return r;
}
struct HttpRequest {
std::string method, body, auth;
bool close = false;
};
enum class ReadResult { Ok, Closed, Timeout, Bad };
ReadResult recvMore(socket_t s, std::string& buf, Clock::time_point deadline)
{
for (;;) {
const auto left = std::chrono::duration_cast<std::chrono::milliseconds>(deadline - Clock::now()).count();
if (left <= 0) return ReadResult::Timeout;
if (!waitReadable(s, static_cast<int>(std::min<long long>(left, 500)))) continue;
char tmp[8192];
const int n = recv(s, tmp, sizeof(tmp), 0);
if (n <= 0) return ReadResult::Closed;
buf.append(tmp, static_cast<size_t>(n));
return ReadResult::Ok;
}
}
std::string lower(std::string s)
{
std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return static_cast<char>(std::tolower(c)); });
return s;
}
ReadResult readRequest(socket_t s, std::string& buf, HttpRequest& out)
{
const auto idle = std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(g_cfg.serverTimeoutSecs));
auto deadline = Clock::now() + idle;
size_t headerEnd;
while ((headerEnd = buf.find("\r\n\r\n")) == std::string::npos) {
if (buf.size() > 64 * 1024) return ReadResult::Bad;
const ReadResult r = recvMore(s, buf, deadline);
if (r != ReadResult::Ok) return r;
}
const std::string head = buf.substr(0, headerEnd);
size_t lineEnd = head.find("\r\n");
const std::string requestLine = head.substr(0, lineEnd);
out = HttpRequest{};
out.method = requestLine.substr(0, requestLine.find(' '));
const bool http10 = requestLine.find("HTTP/1.0") != std::string::npos;
size_t contentLength = 0;
bool keepAlive = false;
while (lineEnd != std::string::npos) {
const size_t next = head.find("\r\n", lineEnd + 2);
const std::string line = head.substr(lineEnd + 2, next == std::string::npos ? std::string::npos : next - lineEnd - 2);
lineEnd = next;
const size_t colon = line.find(':');
if (colon == std::string::npos) continue;
const std::string name = lower(line.substr(0, colon));
std::string value = line.substr(colon + 1);
value.erase(0, value.find_first_not_of(" \t"));
while (!value.empty() && (value.back() == ' ' || value.back() == '\t')) value.pop_back();
if (name == "content-length") contentLength = static_cast<size_t>(std::strtoul(value.c_str(), nullptr, 10));
else if (name == "authorization") out.auth = value;
else if (name == "connection") {
out.close = lower(value) == "close";
keepAlive = lower(value) == "keep-alive";
}
}
if (http10 && !keepAlive) out.close = true;
if (contentLength > 16 * 1024 * 1024) return ReadResult::Bad;
const size_t total = headerEnd + 4 + contentLength;
deadline = Clock::now() + idle;
while (buf.size() < total) {
const ReadResult r = recvMore(s, buf, deadline);
if (r != ReadResult::Ok) return r;
}
out.body = buf.substr(headerEnd + 4, contentLength);
buf.erase(0, total);
return ReadResult::Ok;
}
void serveConnection(socket_t s)
{
std::string buf;
for (;;) {
HttpRequest req;
if (readRequest(s, buf, req) != ReadResult::Ok) break;
++g_stats.requests;
const bool isStop = req.body.find("\"stop\"") != std::string::npos;
// After InterruptHTTPServer, requests on already-open connections get evhttp_send_error(503).
if (g_httpInterrupted.load()) {
++g_stats.unavailable;
if (isStop) event("stop-rejected", "status=503");
sendAll(s, httpReply(503, "text/html",
"<HTML><HEAD>\n<TITLE>503 Service Unavailable</TITLE>\n</HEAD><BODY>\n"
"<H1>Service Unavailable</H1>\n</BODY></HTML>\n",
true));
break;
}
auto item = std::make_shared<Item>();
item->isPost = req.method == "POST";
item->body = std::move(req.body);
item->auth = std::move(req.auth);
if (!g_queue.enqueue(item)) {
if (++g_stats.queueFull == 1) event("queue-full", "slots=" + std::to_string(g_queue.slotsInUse()));
if (isStop) event("stop-rejected", "status=500-queue-full");
if (!sendAll(s, httpReply(500, kTextType, "Work queue depth exceeded", req.close)) || req.close) break;
continue;
}
bool gone = false;
for (;;) {
{
std::unique_lock<std::mutex> lk(item->mx);
if (item->cv.wait_for(lk, 200ms, [&] { return item->done; })) break;
}
if (!gone && peerClosed(s)) {
gone = true; // the slot stays taken until a worker reaches it
noteAbandoned(+1);
}
}
if (gone) {
noteAbandoned(-1);
break;
}
if (!sendAll(s, httpReply(item->status, item->contentType, item->payload, req.close)) || req.close) break;
}
closeSocket(s);
}
void acceptLoop(socket_t listener)
{
while (!g_httpInterrupted.load()) {
if (!waitReadable(listener, 200)) continue;
sockaddr_in addr{};
socklen_t len = sizeof(addr);
const socket_t c = accept(listener, reinterpret_cast<sockaddr*>(&addr), &len);
if (c == kBadSocket) continue;
std::thread(serveConnection, c).detach();
}
closeSocket(listener); // unbound: new connects are refused (or time out, on Windows)
}
socket_t bindListener(int port)
{
const socket_t s = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (s == kBadSocket) return kBadSocket;
#ifdef _WIN32
BOOL exclusive = TRUE;
setsockopt(s, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, reinterpret_cast<const char*>(&exclusive), sizeof(exclusive));
#else
int one = 1;
setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
#endif
sockaddr_in a{};
a.sin_family = AF_INET;
a.sin_port = htons(static_cast<unsigned short>(port));
a.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
if (bind(s, reinterpret_cast<sockaddr*>(&a), sizeof(a)) != 0 || listen(s, 64) != 0) {
closeSocket(s);
return kBadSocket;
}
return s;
}
// =============================================================================================
// Background "node" activity
// =============================================================================================
void logUpdateTip(long h)
{
const std::time_t blockTime = std::time(nullptr) - 20;
logf("UpdateTip: new best=%s height=%ld log2_work=56.82813377 tx=%ld date=%s progress=1.000000 cache=1.2MiB(345tx)",
fakeHash(h).c_str(), h, h * 3 + 12345, utcStamp(blockTime).c_str());
}
// A new block every FAKE_UPDATETIP_SECS; blocks on cs_main like ConnectTip, so none during a rebuild.
void updateTipLoop()
{
while (sleepUnlessShutdown(g_cfg.updateTipSecs)) {
std::lock_guard<std::mutex> lk(g_csMain);
if (shutdownRequested()) return; // net threads are interrupted: no more blocks connect
logUpdateTip(++g_height);
}
}
// Unrelated lines from other threads, as on the real node (they defeat "newest line" heuristics).
void noiseLoop()
{
static const char* const kLines[] = {
"socket sending timeout: 1201s",
"TLS: connection from 203.0.113.10:21768 has been established (initiated)",
"receive version message: /DragonX:1.3.0/: version 170012, blocks=3327101, us=0.0.0.0:0, peer=17",
"Using /16 prefix for IP bucketing",
};
size_t i = 0;
while (sleepUnlessShutdown(g_cfg.noiseSecs)) logf("%s", kLines[i++ % (sizeof(kLines) / sizeof(kLines[0]))]);
}
void prefillAbandoned(long n)
{
long injected = 0;
for (; injected < n; ++injected) {
auto item = std::make_shared<Item>();
item->prefill = true;
item->auth = g_expectedAuth;
item->body = R"({"jsonrpc":"1.0","id":"prefill","method":"getinfo","params":[]})";
noteAbandoned(+1);
// Give the workers a moment to take the first items, so the queue depth alone isn't the limit;
// stop once every slot (workers + queue) is taken.
bool queued = false;
while (!(queued = g_queue.enqueue(item)) && g_queue.slotsInUse() < g_queue.capacity())
std::this_thread::sleep_for(5ms);
if (!queued) {
noteAbandoned(-1);
break;
}
}
event("prefill", "n=" + std::to_string(injected));
}
// BuildWitnessCache: Phase 1 logs "Reading blocks for witness rebuild: <offset> / <total>" whenever
// the ABSOLUTE height is a multiple of 5000, and aborts on shutdown. Phase 2 logs nothing and ignores
// shutdown. Both hold cs_main.
void rebuildLoop()
{
if (!sleepUnlessShutdown(g_cfg.rebuildDelaySecs)) return;
std::unique_lock<std::mutex> lk(g_csMain);
if (shutdownRequested()) return;
const auto started = Clock::now();
const long endHeight = g_height.load();
const long startHeight = endHeight - g_cfg.rebuildTotal;
event("rebuild-start", "total=" + std::to_string(g_cfg.rebuildTotal) + " secs=" + std::to_string(g_cfg.rebuildSecs));
if (g_cfg.prefill > 0) std::thread(prefillAbandoned, g_cfg.prefill).detach();
std::vector<long> offsets;
for (long h = (startHeight / 5000 + 1) * 5000; h <= endHeight; h += 5000) offsets.push_back(h - startHeight);
const double step = offsets.empty() ? 0 : g_cfg.phase1Secs / static_cast<double>(offsets.size());
for (long off : offsets) {
if (shutdownRequested()) {
logf("BuildWitnessCache: shutdown requested, aborting witness rebuild");
event("rebuild-aborted");
return;
}
logf("Reading blocks for witness rebuild: %ld / %ld", off, g_cfg.rebuildTotal);
std::this_thread::sleep_for(std::chrono::duration<double>(step));
}
const auto phase2End = started + std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(g_cfg.rebuildSecs));
std::this_thread::sleep_until(phase2End); // Phase 2: silent, and deaf to shutdown
const long long ms = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - started).count();
logf("BuildWitnessCache: rebuilt %ld note witness cache(s) to height %ld in %lldms using %d thread(s)",
g_cfg.notes, endHeight, ms, 8);
event("rebuild-end", "ms=" + std::to_string(ms));
if (g_cfg.postRebuildLockSecs > 0)
std::this_thread::sleep_for(std::chrono::duration<double>(g_cfg.postRebuildLockSecs));
}
// =============================================================================================
// Datadir lock, pid file, cookie
// =============================================================================================
#ifdef _WIN32
HANDLE g_lockHandle = INVALID_HANDLE_VALUE;
#else
int g_lockFd = -1;
#endif
bool lockDatadir(const fs::path& datadir)
{
const std::string path = (datadir / ".lock").string();
#ifdef _WIN32
g_lockHandle = CreateFileA(path.c_str(), GENERIC_READ | GENERIC_WRITE,
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, nullptr, OPEN_ALWAYS,
FILE_ATTRIBUTE_NORMAL, nullptr);
if (g_lockHandle == INVALID_HANDLE_VALUE) return false;
OVERLAPPED ov{};
return LockFileEx(g_lockHandle, LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY, 0, MAXDWORD, MAXDWORD, &ov) != 0;
#else
g_lockFd = ::open(path.c_str(), O_RDWR | O_CREAT, 0644);
if (g_lockFd < 0) return false;
struct flock fl{};
fl.l_type = F_WRLCK;
fl.l_whence = SEEK_SET;
return fcntl(g_lockFd, F_SETLK, &fl) == 0; // what boost::interprocess::file_lock uses
#endif
}
std::string randomHex(size_t bytes)
{
std::random_device rd;
std::string out;
char b[3];
for (size_t i = 0; i < bytes; ++i) {
std::snprintf(b, sizeof(b), "%02x", rd() & 0xFF);
out += b;
}
return out;
}
[[noreturn]] void exitNow(int code)
{
g_log.close();
{
std::lock_guard<std::mutex> lk(g_eventsMx);
if (g_events) std::fclose(g_events);
g_events = nullptr;
}
std::fflush(stdout);
std::_Exit(code);
}
[[noreturn]] void initError(const std::string& msg)
{
logf("Error: %s", msg.c_str());
std::fprintf(stderr, "Error: %s\n", msg.c_str());
event("init-error", "msg=\"" + msg + "\"");
exitNow(1);
}
void setThreadName(const char* name)
{
#ifdef __linux__
prctl(PR_SET_NAME, name, 0, 0, 0); // the main thread's name is /proc/<pid>/comm
#else
(void)name;
#endif
}
void loadConfig(int argc, char** argv)
{
const auto args = parseArgs(argc, argv);
g_cfg.datadir = args.count("datadir") ? fs::path(args.at("datadir")) : defaultDatadir();
std::error_code ec;
fs::create_directories(g_cfg.datadir, ec);
const auto conf = parseConf(g_cfg.datadir / "DRAGONX.conf");
auto get = [&](const std::string& key, const std::string& def) {
if (args.count(key)) return args.at(key);
if (conf.count(key)) return conf.at(key);
return def;
};
g_cfg.printToConsole = get("printtoconsole", "0") != "0";
g_cfg.rpcPort = std::atoi(get("rpcport", "21769").c_str());
g_cfg.rpcThreads = std::max(1, std::atoi(get("rpcthreads", "8").c_str()));
g_cfg.rpcWorkQueue = std::max(1, std::atoi(get("rpcworkqueue", "16").c_str()));
g_cfg.rpcUser = get("rpcuser", "");
g_cfg.rpcPassword = get("rpcpassword", "");
g_cfg.warmupSecs = envDouble("FAKE_WARMUP_SECS", g_cfg.warmupSecs);
g_cfg.replaySecs = envDouble("FAKE_REPLAY_SECS", g_cfg.replaySecs);
g_cfg.replayBlocksPerSec = std::max(0.1, envDouble("FAKE_REPLAY_BPS", g_cfg.replayBlocksPerSec));
g_cfg.rebuildDelaySecs = envDouble("FAKE_REBUILD_DELAY_SECS", g_cfg.rebuildDelaySecs);
g_cfg.rebuildSecs = envDouble("FAKE_REBUILD_SECS", g_cfg.rebuildSecs);
g_cfg.phase1Secs = std::min(envDouble("FAKE_PHASE1_SECS", g_cfg.phase1Secs), g_cfg.rebuildSecs);
g_cfg.postRebuildLockSecs = envDouble("FAKE_POST_REBUILD_LOCK_SECS", g_cfg.postRebuildLockSecs);
g_cfg.flushSecs = envDouble("FAKE_FLUSH_SECS", g_cfg.flushSecs);
g_cfg.updateTipSecs = std::max(1.0, envDouble("FAKE_UPDATETIP_SECS", g_cfg.updateTipSecs));
g_cfg.noiseSecs = std::max(1.0, envDouble("FAKE_NOISE_SECS", g_cfg.noiseSecs));
g_cfg.serverTimeoutSecs = std::max(1.0, envDouble("FAKE_SERVER_TIMEOUT_SECS", g_cfg.serverTimeoutSecs));
g_cfg.prefill = envLong("FAKE_PREFILL", g_cfg.prefill);
g_cfg.height = envLong("FAKE_HEIGHT", g_cfg.height);
g_cfg.rebuildTotal = std::max(1L, envLong("FAKE_REBUILD_TOTAL", g_cfg.rebuildTotal));
g_cfg.notes = envLong("FAKE_NOTES", g_cfg.notes);
g_cfg.version = envLong("FAKE_VERSION", g_cfg.version);
g_cfg.stuckShutdown = envLong("FAKE_STUCK_SHUTDOWN", 0) != 0;
g_cfg.mining = envLong("FAKE_MINING", 0) != 0;
}
std::string knobsLine()
{
char buf[512];
std::snprintf(buf, sizeof(buf),
"warmup=%gs replay=%gs rebuild=%gs after=%gs phase1=%gs post_lock=%gs flush=%gs tip=%gs "
"prefill=%ld version=%ld workers=%d depth=%d port=%d console=%d mining=%d",
g_cfg.warmupSecs, g_cfg.replaySecs, g_cfg.rebuildSecs, g_cfg.rebuildDelaySecs, g_cfg.phase1Secs,
g_cfg.postRebuildLockSecs, g_cfg.flushSecs, g_cfg.updateTipSecs, g_cfg.prefill, g_cfg.version,
g_cfg.rpcThreads, g_cfg.rpcWorkQueue, g_cfg.rpcPort, g_cfg.printToConsole ? 1 : 0,
g_cfg.mining ? 1 : 0);
return buf;
}
} // namespace
int main(int argc, char** argv)
{
#ifdef _WIN32
WSADATA wsa;
WSAStartup(MAKEWORD(2, 2), &wsa);
#else
std::signal(SIGPIPE, SIG_IGN); // dragonxd ignores SIGPIPE too (init.cpp)
std::signal(SIGHUP, SIG_IGN);
#endif
setThreadName("dragonxd");
loadConfig(argc, argv);
g_log.open(g_cfg.datadir, g_cfg.printToConsole);
g_events = std::fopen((g_cfg.datadir / "fake_dragonxd.events").string().c_str(), "ab");
#ifdef _WIN32
const long pid = static_cast<long>(GetCurrentProcessId());
#else
const long pid = static_cast<long>(getpid());
#endif
const std::string ver = "v" + std::to_string(g_cfg.version / 1000000) + "." +
std::to_string(g_cfg.version / 10000 % 100) + "." +
std::to_string(g_cfg.version / 100 % 100);
logf("DragonX version %s-fakestub", ver.c_str());
logf("fake_dragonxd: %s", knobsLine().c_str());
logf("Using data directory %s", g_cfg.datadir.string().c_str());
event("start", "pid=" + std::to_string(pid) + " " + knobsLine());
if (!lockDatadir(g_cfg.datadir))
initError("Cannot obtain a lock on data directory " + g_cfg.datadir.string() +
". Full node is probably already running.");
if (g_cfg.stuckShutdown) {
// The Sep 29 orphan: an old node wedged in shutdown. It holds the datadir lock, has already
// unbound RPC, and nothing short of a hard kill ends it (on POSIX SIGTERM just re-sets
// fRequestShutdown, which changes nothing).
#ifndef _WIN32
std::signal(SIGTERM, SIG_IGN);
std::signal(SIGINT, SIG_IGN);
#endif
logf("fake_dragonxd: FAKE_STUCK_SHUTDOWN=1, holding the datadir lock with no RPC listener until killed");
event("stuck");
for (;;) std::this_thread::sleep_for(1h);
}
#ifndef _WIN32
std::signal(SIGTERM, onTermSignal);
std::signal(SIGINT, onTermSignal);
const fs::path pidFile = g_cfg.datadir / "hushd.pid";
{
std::ofstream(pidFile) << pid << "\n";
}
#endif
fs::path cookieFile;
if (g_cfg.rpcPassword.empty()) {
g_cfg.rpcUser = "__cookie__";
g_cfg.rpcPassword = randomHex(32);
cookieFile = g_cfg.datadir / ".cookie";
std::ofstream(cookieFile) << g_cfg.rpcUser << ":" << g_cfg.rpcPassword;
logf("Generated RPC authentication cookie %s", cookieFile.string().c_str());
}
g_expectedAuth = "Basic " + base64(g_cfg.rpcUser + ":" + g_cfg.rpcPassword);
g_height = g_cfg.height;
g_minerOn = g_cfg.mining;
const socket_t listener = bindListener(g_cfg.rpcPort);
if (listener == kBadSocket) initError("Unable to bind any endpoint for RPC server");
g_queue.configure(static_cast<size_t>(g_cfg.rpcWorkQueue), g_cfg.rpcThreads);
logf("HTTP: creating work queue of depth %d", g_cfg.rpcWorkQueue);
logf("HTTP: starting %d worker threads", g_cfg.rpcThreads);
std::vector<std::thread> workers;
for (int i = 0; i < g_cfg.rpcThreads; ++i) workers.emplace_back([] { g_queue.run(); });
std::thread acceptThread(acceptLoop, listener);
event("listening", "port=" + std::to_string(g_cfg.rpcPort));
// ---- Init: warmup, optional startup replay, then "Done Loading" -----------------------------
std::thread tipThread, noiseThread, rebuildThread;
setInitMessage("Loading block index...");
bool initOk = sleepUnlessShutdown(g_cfg.warmupSecs);
if (initOk && g_cfg.replaySecs > 0) {
// ActivateBestChain: cs_main per step, checks ShutdownRequested between steps.
setInitMessage("Activating best chain...");
event("replay-start");
const auto end = Clock::now() + std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(g_cfg.replaySecs));
const auto stepDur = std::chrono::duration<double>(1.0 / g_cfg.replayBlocksPerSec);
while (initOk && Clock::now() < end) {
{
std::lock_guard<std::mutex> lk(g_csMain);
logUpdateTip(++g_height);
}
initOk = sleepUnlessShutdown(stepDur.count());
}
event(initOk ? "replay-end" : "replay-interrupted");
}
if (initOk) {
setInitMessage("Full Node Done Loading! :)");
g_warmup = false;
event("warmup-end");
tipThread = std::thread(updateTipLoop);
noiseThread = std::thread(noiseLoop);
if (g_cfg.rebuildSecs > 0) rebuildThread = std::thread(rebuildLoop);
}
// ---- Run until `stop` / SIGTERM --------------------------------------------------------------
auto lastStats = Clock::now();
std::string lastStatsLine;
while (!shutdownRequested()) {
std::this_thread::sleep_for(50ms);
if (Clock::now() - lastStats >= 30s) {
lastStats = Clock::now();
const std::string s = statsLine();
if (s != lastStatsLine) event("stats", s);
lastStatsLine = s;
}
}
// ---- Shutdown ----------------------------------------------------------------------------------
// Interrupt(): unbind RPC and reject new requests on open connections; workers finish only the
// item they hold. Then the thread-group join waits for the rebuild (Phase 2 can't be aborted),
// and only then does Shutdown() log "In progress...".
g_httpInterrupted = true;
g_queue.interrupt();
acceptThread.join();
event("http-interrupted");
logf("tor: Thread interrupt");
logf("scheduler thread interrupt");
logf("net thread interrupt");
if (noiseThread.joinable()) noiseThread.join();
if (tipThread.joinable()) tipThread.join();
if (rebuildThread.joinable()) rebuildThread.join();
logf("Shutdown: In progress...");
setThreadName("hush-shutoff");
event("shutdown-begin");
for (auto& w : workers) w.join();
for (const auto& it : g_queue.drain()) { // StopHTTPServer: ~HTTPRequest on never-run items
logf("~HTTPRequest: Unhandled request");
++g_stats.unhandled;
complete(*it, 500, kTextType, "Unhandled request");
}
std::this_thread::sleep_for(std::chrono::duration<double>(g_cfg.flushSecs));
logf("Shutdown: done");
#ifndef _WIN32
std::error_code ec;
fs::remove(pidFile, ec);
#endif
if (!cookieFile.empty()) {
std::error_code ec2;
fs::remove(cookieFile, ec2);
}
std::this_thread::sleep_for(200ms); // let connection threads flush their last replies
event("clean-exit", statsLine());
exitNow(0);
}
Reference in New Issue View Git Blame Copy Permalink