diff --git a/src/extensions/http/controller/rpc_queryv2_controller.h b/src/extensions/http/controller/rpc_queryv2_controller.h index 25dea399..82cc2e2b 100644 --- a/src/extensions/http/controller/rpc_queryv2_controller.h +++ b/src/extensions/http/controller/rpc_queryv2_controller.h @@ -284,6 +284,7 @@ class RpcQueryV2Controller : public HttpController const unsigned int scanHi = hasTickHint ? scanLo : system.tick; for (unsigned int tick = scanLo; tick <= scanHi && !found; tick++) { + PinScope _pinScope; // release this tick's swap-page pin each iteration (a full-epoch scan would otherwise exhaust the cache -> exit(1)) TickStorage::tickData.acquireLock(); TickData *tickData = TickStorage::tickData.getByTickIfNotEmpty(tick); if (tickData) @@ -394,6 +395,7 @@ class RpcQueryV2Controller : public HttpController std::vector> matchedBufs; for (unsigned int tick = system.initialTick; tick <= system.tick; tick++) { + PinScope _pinScope; // release this tick's swap-page pins each iteration TickData localTickData; TickStorage::tickData.acquireLock(); TickData *tickData = TickStorage::tickData.getByTickIfNotEmpty(tick); @@ -427,6 +429,7 @@ class RpcQueryV2Controller : public HttpController } for (auto &buf : matchedBufs) { + PinScope _pinScope; // transactionToJson(forceToBeProcessed=true) pins the tx's tickData page; release per iteration transactions.append(HttpUtils::transactionToJson(reinterpret_cast(buf.data()))); } @@ -592,7 +595,7 @@ class RpcQueryV2Controller : public HttpController std::string direction = (*json).get("direction", "both").asString(); const bool wantIn = (direction == "in" || direction == "both"); const bool wantOut = (direction == "out" || direction == "both"); - const unsigned int limit = (*json).get("limit", 50).asUInt(); + const unsigned int limit = std::min((*json).get("limit", 50).asUInt(), 1000u); // cap result size (also bounds hits.reserve below) Json::Value transactions(Json::arrayValue); // pair of (buf, direction); walk newest tick → oldest, stop at limit. @@ -602,6 +605,7 @@ class RpcQueryV2Controller : public HttpController for (unsigned int tick = system.tick; tick + 1 > system.initialTick && hits.size() < limit; tick--) { + PinScope _pinScope; // release this tick's swap-page pins each iteration TickData localTickData; TickStorage::tickData.acquireLock(); TickData *tickData = TickStorage::tickData.getByTickIfNotEmpty(tick); @@ -638,6 +642,7 @@ class RpcQueryV2Controller : public HttpController for (auto &h : hits) { + PinScope _pinScope; // transactionToJson pins the tx's tickData page; release per iteration Json::Value txJson = HttpUtils::transactionToJson(reinterpret_cast(h.buf.data())); txJson["direction"] = h.dir; transactions.append(txJson); @@ -1738,6 +1743,7 @@ class RpcQueryV2Controller : public HttpController // Refresh per-tick caches when tick changes. if (headerTick != cachedTick) { + PinScope _pinScope; // release this tick's refresh swap-page pins (scanning many ticks would otherwise exhaust the cache) cachedTick = headerTick; setMem(&cachedTickData, sizeof(TickData), 0); TickStorage::tickData.acquireLock(); diff --git a/src/extensions/http/controller/rpc_stats_controller.h b/src/extensions/http/controller/rpc_stats_controller.h index 86b1d414..4aead640 100644 --- a/src/extensions/http/controller/rpc_stats_controller.h +++ b/src/extensions/http/controller/rpc_stats_controller.h @@ -238,17 +238,18 @@ class RpcStatsController : public HttpController data["currentTick"] = system.tick; data["ticksInCurrentEpoch"] = system.tick - system.initialTick; unsigned int emptyTicks = 0; + // Per-iteration lock + PinScope: a whole-epoch scan must not hold tickDataLock across the loop + // (starves the tick processor) nor accumulate a swap-page pin per tick (exhausts the cache -> exit(1)). + for (unsigned int tick = system.initialTick; tick < system.tick; tick++) { + PinScope _pinScope; TickStorage::tickData.acquireLock(); - for (unsigned int tick = system.initialTick; tick < system.tick; tick++) - { - TickData *tickData = TickStorage::tickData.getByTickIfNotEmpty(tick); - if (!tickData) - { - emptyTicks++; - } - } + const bool empty = (TickStorage::tickData.getByTickIfNotEmpty(tick) == nullptr); TickStorage::tickData.releaseLock(); + if (empty) + { + emptyTicks++; + } } data["emptyTicksInCurrentEpoch"] = emptyTicks; data["epochTickQuality"] = system.tick - system.initialTick == 0 ? 0 : std::roundf((float)(system.tick - system.initialTick - emptyTicks) / (float)(system.tick - system.initialTick) * 100000.0f) / 100000.0f; diff --git a/src/extensions/http/utils.h b/src/extensions/http/utils.h index 20a85d4d..bc010266 100644 --- a/src/extensions/http/utils.h +++ b/src/extensions/http/utils.h @@ -176,6 +176,7 @@ class HttpUtils TickData localTickData; for (unsigned int tick = system.initialTick; tick <= system.tick; tick++) { + PinScope _pinScope; // release this tick's swap-page pin each iteration (full-epoch scan) TickStorage::tickData.acquireLock(); TickData *tickData = TickStorage::tickData.getByTickIfNotEmpty(tick); if (tickData) @@ -206,18 +207,26 @@ class HttpUtils static void fetch(const std::string &url, const std::string &path, const drogon::HttpMethod method, const Json::Value &body, const std::map &headers = {}, std::function callback = nullptr) { auto client = drogon::HttpClient::newHttpClient(url); - auto req = drogon::HttpRequest::newHttpJsonRequest(body); // Helper for JSON + + // Serialize per-call: newHttpJsonRequest() shares a process-wide static StreamWriterBuilder that this background thread races against drogon's loop threads (SIGSEGV in newStreamWriter()). + auto req = drogon::HttpRequest::newHttpRequest(); req->setMethod(method); req->setPath(path); + if (!body.isNull()) + { + Json::StreamWriterBuilder jsonWriter; + jsonWriter["commentStyle"] = "None"; + jsonWriter["indentation"] = ""; + req->setBody(Json::writeString(jsonWriter, body)); + } - // Set headers for (const auto& header : headers) { req->addHeader(header.first, header.second); } - // Set type to application/json req->addHeader("Content-Type", "application/json"); - client->sendRequest(req, [&](drogon::ReqResult _result, const drogon::HttpResponsePtr &_resp) { + // Capture client + callback by value: the request is async and outlives this frame, so by-reference captures dangle on completion. + client->sendRequest(req, [client, callback](drogon::ReqResult _result, const drogon::HttpResponsePtr &_resp) { if (callback != nullptr) { callback(_result, _resp); diff --git a/src/extensions/k12_state_digest_cache.h b/src/extensions/k12_state_digest_cache.h new file mode 100644 index 00000000..e711ffc6 --- /dev/null +++ b/src/extensions/k12_state_digest_cache.h @@ -0,0 +1,271 @@ +#pragma once + +// Incremental KangarooTwelve cache for smart-contract state digests. K12 hashes its input as a +// one-level tree of 8192-byte chunks (each interior chunk -> a 32-byte chaining value absorbed +// into a final node), so a byte change in one chunk only invalidates that chunk's CV. We cache +// the CVs per contract and, using mprotect-read-only + the SIGSEGV handler to learn which chunks +// changed since the last digest, re-hash only the dirty chunks and reuse the rest. The result is +// byte-identical to the one-shot KangarooTwelve because every non-cached step is the upstream +// XKCP path verbatim. All state stays in RAM. On by default on Linux; --no-k12-state-cache off. + +#include "../K12/kangaroo_twelve_xkcp.h" // XKCP primitives + K12_chunkSize/K12_suffixLeaf/maxCapacityInBytes + copyMem/setMem + +#if defined(__linux__) +#include +#include +#include +#include +#include +#endif + +namespace K12StateDigestCache +{ + inline constexpr unsigned int K12SDC_CHUNK = K12_chunkSize; // 8192 + inline constexpr unsigned int K12SDC_CV = 32; // KT128 capacity (2*128/8) + inline constexpr int K12SDC_SEC = 128; + + // Bit-exact with XKCP::KangarooTwelve(state,size,out,32). cvCache holds 32 bytes per chunk; + // only chunks with chunkDirty[k] set are re-hashed (and their CV re-cached), the rest reuse + // cvCache[k]. The finalNode absorption order mirrors KangarooTwelve_Update + _Final verbatim, + // which is what makes the output identical. cvCache[0] is unused (chunk 0 has no CV). Pure, + // self-contained (XKCP + copyMem only) so it is unit-testable without the node TU. + inline void incrementalDigest(const unsigned char* state, unsigned long long size, + unsigned char* cvCache, const unsigned char* chunkDirty, + unsigned char* out) + { + if (size <= K12SDC_CHUNK) + { + XKCP::KangarooTwelve(state, (unsigned int)size, out, K12SDC_CV); // single node: no tree + return; + } + + XKCP::KangarooTwelve_Instance kt; + XKCP::KangarooTwelve_Initialize(&kt, K12SDC_SEC, K12SDC_CV); + + // chunk 0 -> finalNode raw, then the 0x03 first-node framing (KangarooTwelve_Update path). + XKCP::TurboSHAKE_Absorb(&kt.finalNode, state, K12SDC_CHUNK); + const unsigned char pad03 = 0x03; + kt.queueAbsorbedLen = 0; + kt.blockNumber = 1; + XKCP::TurboSHAKE_Absorb(&kt.finalNode, &pad03, 1); + kt.finalNode.byteIOIndex = (kt.finalNode.byteIOIndex + 7) & ~7; // zero-pad to 64 bits + + unsigned long long off = K12SDC_CHUNK, k = 1; + while (off < size) + { + unsigned int len = (size - off < K12SDC_CHUNK) ? (unsigned int)(size - off) : K12SDC_CHUNK; + if (len == K12SDC_CHUNK) + { + unsigned char inter[maxCapacityInBytes]; + ++kt.blockNumber; + if (chunkDirty[k]) + { + XKCP::TurboSHAKE_Initialize(&kt.queueNode, K12SDC_SEC); + XKCP::TurboSHAKE_Absorb(&kt.queueNode, state + off, K12SDC_CHUNK); + XKCP::TurboSHAKE_AbsorbDomainSeparationByte(&kt.queueNode, K12_suffixLeaf); + XKCP::TurboSHAKE_Squeeze(&kt.queueNode, inter, K12SDC_CV); + copyMem(cvCache + k * K12SDC_CV, inter, K12SDC_CV); + } + else + { + copyMem(inter, cvCache + k * K12SDC_CV, K12SDC_CV); + } + XKCP::TurboSHAKE_Absorb(&kt.finalNode, inter, K12SDC_CV); + } + else + { + // partial last chunk: raw into queueNode, no CV (KangarooTwelve_Final finalizes it) + XKCP::TurboSHAKE_Initialize(&kt.queueNode, K12SDC_SEC); + XKCP::TurboSHAKE_Absorb(&kt.queueNode, state + off, len); + kt.queueAbsorbedLen = len; + } + off += len; + ++k; + } + XKCP::KangarooTwelve_Final(&kt, out, nullptr, 0); // right_encode(0) leaf + right_encode(n)||0xFFFF + suffix + } + +#if defined(__linux__) + inline bool gK12StateDigestCacheEnabled = true; // on by default; --no-k12-state-cache flips it +#else + inline bool gK12StateDigestCacheEnabled = false; // mprotect+SIGSEGV is Linux-only; one-shot elsewhere +#endif + inline bool gK12StateDigestCacheVerify = false; // --k12-state-cache-verify: also run one-shot, stall on mismatch + +#if defined(__linux__) && defined(SINGLE_COMPILE_UNIT) + + inline unsigned long long pageSize() { return (unsigned long long)sysconf(_SC_PAGESIZE); } + + // Only multi-chunk states benefit, and a chunk must be a whole number of pages so a per-chunk + // mprotect is exact (true for 8192/4096; the page-size guard falls back to one-shot otherwise). + inline constexpr unsigned int K12SDC_MIN_CHUNKS = 16; // ~128 KB + inline bool shouldCache(unsigned long long size) + { + return gK12StateDigestCacheEnabled + && size > (unsigned long long)K12SDC_CHUNK * K12SDC_MIN_CHUNKS + && (K12SDC_CHUNK % pageSize()) == 0; + } + inline bool wantsPageAlignedAlloc(unsigned long long size) { return shouldCache(size); } + + struct Record + { + unsigned char* cv = nullptr; // 32*numChunks, allocated once + volatile unsigned char* chunkDirty = nullptr; // numChunks bytes, 1 = dirty + unsigned long long size = 0; + unsigned long long allocLen = 0; // numChunks*CHUNK (page multiple) + unsigned int numChunks = 0; + bool cached = false; + bool armed = false; // false until first computeDigest arms whole region RO + }; + inline Record gCache[MAX_NUMBER_OF_CONTRACTS]; + + // Compact registry the SIGSEGV fast path scans lock-free (mirrors SwapDirtyTrack::gPools). + struct Region { unsigned char** basePtr; unsigned long long allocLen; volatile unsigned char* chunkDirty; }; + inline constexpr int MAX_REGIONS = (int)MAX_NUMBER_OF_CONTRACTS; + inline Region gRegions[MAX_REGIONS]; + inline std::atomic gRegionCount{0}; + + inline bool isCached(unsigned int i) { return gCache[i].cached; } + + // Page-aligned state buffer for cached contracts (free()-compatible with freePool). + inline bool allocStatePool(unsigned long long size, void** buffer) + { + const unsigned long long ps = pageSize(); + const unsigned long long numChunks = (size + K12SDC_CHUNK - 1) / K12SDC_CHUNK; + const unsigned long long allocLen = numChunks * K12SDC_CHUNK; // multiple of CHUNK => multiple of page + void* p = aligned_alloc((size_t)ps, (size_t)allocLen); + if (!p) + return false; + setMem(p, allocLen, 0); + *buffer = p; + return true; + } + + // Allocate derived structures + register for fault dispatch. Reads no contract bytes (CVs are + // built lazily, all-dirty, at the first computeDigest). Single-threaded, before tick threads. + inline void init() + { + if (!gK12StateDigestCacheEnabled) + return; + unsigned int cachedCount = 0; + unsigned long long trackedBytes = 0; + for (unsigned int i = 0; i < contractCount; i++) + { + const unsigned long long size = contractDescriptions[i].stateSize; + if (gCache[i].cached || !shouldCache(size) || contractStates[i] == nullptr) + continue; + Record& r = gCache[i]; + r.size = size; + r.numChunks = (unsigned int)((size + K12SDC_CHUNK - 1) / K12SDC_CHUNK); + r.allocLen = (unsigned long long)r.numChunks * K12SDC_CHUNK; + r.cv = (unsigned char*)malloc((size_t)r.numChunks * K12SDC_CV); + r.chunkDirty = (volatile unsigned char*)malloc((size_t)r.numChunks); + if (!r.cv || !r.chunkDirty) + { + free(r.cv); + free((void*)r.chunkDirty); + r.cv = nullptr; r.chunkDirty = nullptr; + continue; + } + setMem(r.cv, (size_t)r.numChunks * K12SDC_CV, 0); + setMem((void*)r.chunkDirty, r.numChunks, 1); // first digest rebuilds every CV + r.armed = false; + r.cached = true; + cachedCount++; + trackedBytes += r.allocLen; + const int idx = gRegionCount.fetch_add(1, std::memory_order_acq_rel); + if (idx < MAX_REGIONS) + gRegions[idx] = { (unsigned char**)&contractStates[i], r.allocLen, r.chunkDirty }; + } + printf("K12 state-digest cache: incrementally hashing %u contract state(s), %llu MB tracked%s\n", + cachedCount, trackedBytes >> 20, gK12StateDigestCacheVerify ? " (self-verify ON)" : ""); + } + + // SIGSEGV fast path (chained from signalHandler). Async-signal-safe: atomic load + byte store + + // mprotect over an immutable registry. Maps the faulting address to its chunk, marks it dirty, + // restores write access, returns true (kernel retries the store). + inline bool tryMarkDirty(void* addr) + { + if (!gK12StateDigestCacheEnabled) + return false; + unsigned char* a = (unsigned char*)addr; + const int n = gRegionCount.load(std::memory_order_acquire); + for (int i = 0; i < n; i++) + { + Region& rg = gRegions[i]; + unsigned char* base = *rg.basePtr; + if (base == nullptr) + continue; + if (a >= base && a < base + rg.allocLen) + { + const unsigned int chunk = (unsigned int)(((unsigned long long)(a - base)) / K12SDC_CHUNK); + rg.chunkDirty[chunk] = 1; + mprotect(base + (unsigned long long)chunk * K12SDC_CHUNK, K12SDC_CHUNK, PROT_READ | PROT_WRITE); + return true; + } + } + return false; + } + + // Runs under the caller's contractStateLock[i] read-lock (writers blocked, single-threaded) so + // arm + dirty-clear is race-free vs faults. Computes the digest, then arms the region read-only + // for the next tick: first call arms the whole region; later calls re-arm only the chunks + // faulted this cycle. + inline void computeDigest(unsigned int i, void* out) + { + Record& r = gCache[i]; + incrementalDigest(contractStates[i], r.size, r.cv, (const unsigned char*)r.chunkDirty, (unsigned char*)out); + + if (gK12StateDigestCacheVerify) + { + m256i ref; + KangarooTwelve(contractStates[i], (unsigned int)r.size, &ref, 32); + if (memcmp(&ref, out, 32) != 0) + while (true) { fprintf(stderr, "FATAL: K12 state-cache digest != one-shot (verify), contract %u\n", i); fflush(stderr); sleep(1); } + } + + unsigned char* base = contractStates[i]; + if (!r.armed) + { + mprotect(base, r.allocLen, PROT_READ); + setMem((void*)r.chunkDirty, r.numChunks, 0); + r.armed = true; + } + else + { + for (unsigned int k = 0; k < r.numChunks; k++) + if (r.chunkDirty[k]) + { + mprotect(base + (unsigned long long)k * K12SDC_CHUNK, K12SDC_CHUNK, PROT_READ); + r.chunkDirty[k] = 0; + } + } + } + + // Force a full canonical rebuild: drop protection + mark all chunks dirty. For a rollback that + // restores contract bytes by a non-faulting path, and for the quorum-mismatch recovery. + // Caller holds contractStateLock[i] (or runs single-threaded at the digest site). + inline void invalidateContract(unsigned int i) + { + Record& r = gCache[i]; + if (!r.cached) + return; + mprotect(contractStates[i], r.allocLen, PROT_READ | PROT_WRITE); + setMem((void*)r.chunkDirty, r.numChunks, 1); + r.armed = false; + } + inline void invalidateAll() { for (unsigned int i = 0; i < contractCount; i++) invalidateContract(i); } + +#else // non-Linux or non-TU: inert stubs so the core hooks compile and the path stays one-shot + + inline bool isCached(unsigned int) { return false; } + inline void init() {} + inline void computeDigest(unsigned int, void*) {} + inline bool tryMarkDirty(void*) { return false; } + inline bool wantsPageAlignedAlloc(unsigned long long) { return false; } + inline bool allocStatePool(unsigned long long, void**) { return false; } + inline void invalidateContract(unsigned int) {} + inline void invalidateAll() {} + +#endif +} diff --git a/src/extensions/overload.h b/src/extensions/overload.h index 6236f7e7..73287075 100644 --- a/src/extensions/overload.h +++ b/src/extensions/overload.h @@ -31,6 +31,7 @@ #endif #include +#include static volatile bool listOfPeersIsStaticLiteNode = false; @@ -113,8 +114,6 @@ Json::Value getCheckInData(const std::string& challenge = "") #endif static volatile bool forceDontCheckComputerDigest = false; -static std::vector forceDontUseSecurityTickChangeStack; -static volatile bool forceDontUseSecurityTick = false; //////////// Go Behind Testnet Trick \\\\\\\\ @@ -124,34 +123,20 @@ static inline bool isTestnetGoBehindTrick = false; static inline unsigned long long tickDelay = 0; -//////////// Security Tick Feature \\\\\\\\\\\\ - -static inline unsigned long long securityTick = 1; - //////////// HTTP Server Port \\\\\\\\\\\\ static inline int httpPort = 41841; +// Security-tick skip removed: every tick verifies state (cheap with the K12 state-digest cache). +// A single tick can still be suppressed by the catch-up override. bool isSystemAtSecurityTick() { - if (forceDontCheckComputerDigest) - { - return false; - } - if (forceDontUseSecurityTick || securityTick == 0 || system.tick == system.initialTick) - { - return true; - } - return ((system.tick - system.initialTick) % securityTick == 0); + return !forceDontCheckComputerDigest; } bool isNextTickIsSecurityTick() { - if (securityTick == 0 || system.tick == system.initialTick) - { - return true; - } - return (((system.tick + 1) - system.initialTick) % securityTick == 0); + return true; } ////////// Skip Solution Transaction Verification Feature \\\\\\\\\\ @@ -399,7 +384,9 @@ struct Overload { inline static std::vector threads; inline static std::unordered_map incomingSocketMap; - inline static std::unordered_map tcpDataMap; + // shared_ptr ownership: detached connect/accept threads capture the element by value so it + // survives a concurrent DestroyChild() erase (was a use-after-free on epoch-end teardown). + inline static std::unordered_map> tcpDataMap; inline static std::unordered_map eventDataMap; inline static std::unordered_map isReceiveThreadSetupMap; inline static std::unordered_map isSendThreadSetupMap; @@ -407,6 +394,7 @@ struct Overload { inline static EventQueue receiveQueue; inline static std::mutex networkingLock; + inline static std::mutex eventMapLock; // guards eventDataMap (CreateEvent/CloseEvent vs startThread callback lookup on AP worker threads) // Directly call the setup function without using custom stack. static void startThread(EFI_AP_PROCEDURE procedure, void* data, unsigned long long ProcessorNumber, EFI_EVENT WaitEvent, unsigned long long TimeoutInMicroseconds) { @@ -467,13 +455,24 @@ struct Overload { // call the event call back if (WaitEvent) { - auto it = eventDataMap.find((unsigned long long)WaitEvent); - if (it != eventDataMap.end()) { - EventData& eventData = it->second; - if (eventData.notifyFunction) { + // Copy the callback out under the lock, then release before invoking it: the callback may + // re-enter CloseEvent (which takes eventMapLock), so holding it here would self-deadlock. + void* context = nullptr; + void (*notifyFunction)(void*, void*) = nullptr; + bool found = false; + { + std::lock_guard lock(eventMapLock); + auto it = eventDataMap.find((unsigned long long)WaitEvent); + if (it != eventDataMap.end()) { + found = true; + context = it->second.context; + notifyFunction = reinterpret_cast(it->second.notifyFunction); + } + } + if (found) { + if (notifyFunction) { // Call the notify function with the context - void (*notifyFunction)(void*, void*) = reinterpret_cast(eventData.notifyFunction); - notifyFunction(WaitEvent, eventData.context); + notifyFunction(WaitEvent, context); } } else { @@ -553,7 +552,9 @@ struct Overload { *Interface = new EFI_TCP4_PROTOCOL; // Check if this is a incomming socket and set socket instance if it is if (incomingSocketMap.contains((unsigned long long)Handle)) { - TcpData& tcpData = tcpDataMap[(unsigned long long) * Interface]; + auto tcpDataPtr = std::make_shared(); + tcpDataMap[(unsigned long long) * Interface] = tcpDataPtr; + TcpData& tcpData = *tcpDataPtr; tcpData.socket = incomingSocketMap[(unsigned long long)Handle]; incomingSocketMap.erase((unsigned long long)Handle); @@ -592,7 +593,10 @@ struct Overload { eventData.event = *Event; eventData.context = NotifyContext; eventData.notifyFunction = NotifyFunction; - eventDataMap[(unsigned long long) * Event] = eventData; + { + std::lock_guard lock(eventMapLock); + eventDataMap[(unsigned long long) * Event] = eventData; + } return EFI_SUCCESS; } @@ -604,7 +608,10 @@ struct Overload { eventData.event = *Event; eventData.context = NotifyContext; eventData.notifyFunction = NotifyFunction; - eventDataMap[(unsigned long long) * Event] = eventData; + { + std::lock_guard lock(eventMapLock); + eventDataMap[(unsigned long long) * Event] = eventData; + } return EFI_SUCCESS; } @@ -613,6 +620,7 @@ struct Overload { } static EFI_STATUS CloseEvent(IN EFI_EVENT Event) { + std::lock_guard lock(eventMapLock); auto it = eventDataMap.find((unsigned long long)Event); if (it != eventDataMap.end()) { delete (unsigned long long*)Event; // Free the allocated memory for the event @@ -761,7 +769,7 @@ struct Overload { // Remove tcp4Protocol data from handle unsigned long long key = *(unsigned long long*)ChildHandle; if (tcpDataMap.contains(key)) { - TcpData& tcpData = tcpDataMap[key]; + TcpData& tcpData = *tcpDataMap[key]; if (tcpData.socket != INVALID_SOCKET) { closesocket(tcpData.socket); tcpData.socket = INVALID_SOCKET; @@ -798,7 +806,7 @@ struct Overload { TcpData* tcpData = nullptr; unsigned long long key = (unsigned long long)This; if (tcpDataMap.contains(key)) { - tcpData = &tcpDataMap[key]; + tcpData = tcpDataMap[key].get(); } else { logToConsole(L"No Tcp Data For This Connect!"); @@ -826,7 +834,7 @@ struct Overload { TcpData* tcpData = nullptr; unsigned long long key = (unsigned long long)This; if (tcpDataMap.contains(key)) { - tcpData = &tcpDataMap[key]; + tcpData = tcpDataMap[key].get(); } else { logToConsole(L"No Tcp Data For This Connect!"); @@ -864,7 +872,7 @@ struct Overload { if (Tcp4State) { if (tcpDataMap.contains((unsigned long long)This)) { - TcpData& tcpData = tcpDataMap[(unsigned long long)This]; + TcpData& tcpData = *tcpDataMap[(unsigned long long)This]; if (tcpData.socket == INVALID_SOCKET) { *Tcp4State = Tcp4StateClosed; } @@ -972,24 +980,25 @@ struct Overload { } unsigned long long key = (unsigned long long)This; - tcpDataMap.emplace(key, data); + tcpDataMap.emplace(key, std::make_shared(data)); return EFI_SUCCESS; } // Note: Only global tcp4Protocol call this function, peers don't call static EFI_STATUS Accept(IN void* This, IN EFI_TCP4_LISTEN_TOKEN* ListenToken, IN void* peer) { - TcpData* tcpData = nullptr; + std::shared_ptr tcpDataPtr; unsigned long long key = (unsigned long long)This; if (tcpDataMap.contains(key)) { - tcpData = &tcpDataMap[key]; + tcpDataPtr = tcpDataMap[key]; } else { logToConsole(L"No Tcp Data For Global Tcp Connect!"); return EFI_UNSUPPORTED; } - // accept in a thread - std::thread acceptThread([tcpData, ListenToken, peer]() { + // accept in a thread (capture shared_ptr by value so tcpData survives a concurrent DestroyChild) + std::thread acceptThread([tcpDataPtr, ListenToken, peer]() { + TcpData* tcpData = tcpDataPtr.get(); sockaddr_in addr{}; addr.sin_family = AF_INET; addr.sin_port = htons(tcpData->configData.AccessPoint.StationPort); @@ -1051,10 +1060,13 @@ struct Overload { static EFI_STATUS Connect(IN void* This, IN EFI_TCP4_CONNECTION_TOKEN* ConnectionToken) { static std::map latestConnectTimestampMap; // map of + static std::mutex latestConnectTimestampMapMutex; // guards latestConnectTimestampMap (concurrent detached connect threads) + std::shared_ptr tcpDataPtr; TcpData* tcpData = nullptr; unsigned long long key = (unsigned long long)This; if (tcpDataMap.contains(key)) { - tcpData = &tcpDataMap[key]; + tcpDataPtr = tcpDataMap[key]; + tcpData = tcpDataPtr.get(); } else { logToConsole(L"No Tcp Data For This Connect!"); @@ -1078,11 +1090,17 @@ struct Overload { #endif unsigned int ipInNumber = *(unsigned int*)tcpData->configData.AccessPoint.RemoteAddress.Addr; - // connect in a thread - std::thread connectThread([tcpData, serverAddr, ConnectionToken, ipInNumber]() { + // connect in a thread (capture shared_ptr by value so tcpData survives a concurrent DestroyChild) + std::thread connectThread([tcpDataPtr, serverAddr, ConnectionToken, ipInNumber]() { + TcpData* tcpData = tcpDataPtr.get(); auto now = std::chrono::system_clock::now(); long long ms = std::chrono::duration_cast(now.time_since_epoch()).count(); - if (ms - latestConnectTimestampMap[ipInNumber] < 2'000) { + long long lastTs; + { + std::lock_guard g(latestConnectTimestampMapMutex); + lastTs = latestConnectTimestampMap[ipInNumber]; + } + if (ms - lastTs < 2'000) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); } tcpData->connectStatus = ConnectStatus::Connecting; @@ -1099,7 +1117,10 @@ struct Overload { #endif } - latestConnectTimestampMap[ipInNumber] = ms; + { + std::lock_guard g(latestConnectTimestampMapMutex); + latestConnectTimestampMap[ipInNumber] = ms; + } }); connectThread.detach(); diff --git a/src/extensions/swapvm_dirty_track.h b/src/extensions/swapvm_dirty_track.h new file mode 100644 index 00000000..f7651ec5 --- /dev/null +++ b/src/extensions/swapvm_dirty_track.h @@ -0,0 +1,79 @@ +#pragma once + +// Auto dirty-page tracking for SwapVirtualMemory cache pools (Linux). A freshly loaded slot is armed +// read-only; the first write faults and is caught by the SIGSEGV fast path here (marks the slot dirty, +// restores write access), so eviction can skip the writeback for slots never modified since load. +// SwapVM arms its own slots; this header holds the global pieces the handler needs: toggle, registry, +// fast path. Gated by gSwapDirtyTrackEnabled. + +#if defined(__linux__) + +#include +#include +#include + +// Set by --swap-dirty-track. Off => pools are never armed read-only => no faults, no tracking. +inline bool gSwapDirtyTrackEnabled = false; + +namespace SwapDirtyTrack +{ + struct Pool + { + unsigned char** basePtr; // ¤tPage of the VM: *basePtr is the live pool base, or + // NULL while the pool is being reallocated across an epoch reset + unsigned long long slotStride; // page-aligned bytes between slots (a multiple of the OS page size) + int numSlots; // numCachePage + 1 + volatile unsigned char* dirty; // per-slot dirty bytes (a stable VM-member array) + }; + + inline constexpr int MAX_POOLS = 16; + inline Pool gPools[MAX_POOLS]; + inline std::atomic gPoolCount{0}; + + // Registered once per SwapVM at init. Keyed by ¤tPage so the entry follows the pool's realloc + // on epoch reset (and reads NULL, skipped, mid-realloc). Built before any slot is armed, so the + // fault path reads it lock-free. + inline void registerPool(unsigned char** basePtr, unsigned long long slotStride, int numSlots, volatile unsigned char* dirty) + { + int idx = gPoolCount.fetch_add(1, std::memory_order_acq_rel); + if (idx >= MAX_POOLS) + return; // more SwapVM instances than expected; tracking simply won't cover the overflow + gPools[idx].basePtr = basePtr; + gPools[idx].slotStride = slotStride; + gPools[idx].numSlots = numSlots; + gPools[idx].dirty = dirty; + } + + // SIGSEGV fast path (called from signalHandler). If addr is in a registered pool, mark the slot + // dirty, restore write access, return true (kernel retries the store); else false -> real crash + // path. Async-signal-safe: a byte store + mprotect over an immutable registry. + inline bool tryMarkDirty(void* addr) + { + if (!gSwapDirtyTrackEnabled) + return false; + unsigned char* a = (unsigned char*)addr; + const int n = gPoolCount.load(std::memory_order_acquire); + for (int i = 0; i < n; i++) + { + Pool& p = gPools[i]; + unsigned char* base = *p.basePtr; + if (base == nullptr) + continue; + unsigned char* end = base + p.slotStride * (unsigned long long)p.numSlots; + if (a >= base && a < end) + { + const int slot = (int)(((unsigned long long)(a - base)) / p.slotStride); + p.dirty[slot] = 1; + mprotect(base + (unsigned long long)slot * p.slotStride, p.slotStride, PROT_READ | PROT_WRITE); + return true; + } + } + return false; + } +} + +#else // !__linux__ : tracking unavailable; the flag exists but stays false. + +inline bool gSwapDirtyTrackEnabled = false; + +#endif diff --git a/src/oracle_core/oracle_engine.h b/src/oracle_core/oracle_engine.h index 0759872c..bc9efee5 100644 --- a/src/oracle_core/oracle_engine.h +++ b/src/oracle_core/oracle_engine.h @@ -2525,6 +2525,7 @@ class OracleEngine uint64_t storageBytesUsed = 8; for (uint32_t queryIndex = 0; queryIndex < oracleQueryCount; ++queryIndex) { + PinScope _pinScope; // release per-query swap-page pins each iteration (debug+UEFI-only consistency check) const OracleQueryMetadata& oqm = queries[queryIndex]; ASSERT(oqm.interfaceIndex < OI::oracleInterfacesCount); ASSERT(oqm.queryId); diff --git a/src/platform/virtual_memory.h b/src/platform/virtual_memory.h index 5b0e23bf..b442ca70 100644 --- a/src/platform/virtual_memory.h +++ b/src/platform/virtual_memory.h @@ -12,6 +12,7 @@ #include #include "extensions/zipper.h" +#include "extensions/swapvm_dirty_track.h" // gSwapDirtyTrackEnabled + SIGSEGV fast path + pool registry #ifdef __linux__ // Runtime toggle for SwapVM page compression (blosc2); off unless --swap-compression is passed. @@ -128,14 +129,39 @@ class VirtualMemory unsigned long long currentId; // total items in this array, aka: latest item index + 1 unsigned long long currentPageId; // current page index that's written on - volatile char memLock; // every read/write needs a memory lock, can optimize later + // Reader/writer spinlock: memLock is the writer byte, memReaders counts shared holders. Cache-hit + // fast paths take shared (lockShared); every mutator takes exclusive (acquireMemLock). + volatile char memLock; + volatile long memReaders = 0; - // Acquire memLock, but if we are the main thread (the only thread that can do IO), - // keep pumping the async IO queue while we wait. Otherwise a non-main thread parked - // inside asyncLoad() while holding memLock could never be serviced and cause deadlock. + // Exclusive: claim the writer byte, then drain shared readers. The claim pumps IO while waiting + // (a writer ahead may be parked in asyncLoad() under the lock — the documented deadlock). The + // drain does NOT: shared holders are hit-path/unpin only, never do IO, so they always finish on + // their own — a plain spin suffices. void acquireMemLock() { - acquireLockWithIOPump(memLock); + acquireLockWithIOPump(memLock); + while (memReaders) + _mm_pause(); + } + + // Shared, writer-preferring. The seq_cst inc + re-read of memLock is the Dekker handshake vs + // acquireMemLock's claim-then-drain: a writer never runs while a reader is live. + void lockShared() + { + for (;;) + { + while (memLock) + _mm_pause(); + _InterlockedIncrement(&memReaders); + if (!memLock) + return; + _InterlockedDecrement(&memReaders); + } + } + void unlockShared() + { + _InterlockedDecrement(&memReaders); } void generatePageName(CHAR16 pageName[64], unsigned long long page_id) @@ -329,6 +355,7 @@ class VirtualMemory currentId = 0; currentPageId = 0; memLock = 0; + memReaders = 0; } public: @@ -777,6 +804,10 @@ class SwapVirtualMemory : private VirtualMemory 0 is never evicted. Guarded by - // memLock. See ThreadPinArena / releaseThreadPins above. - int pinCount[numCachePage + 1] = {}; + // Threads referencing the page in this slot (via the pin arena); pinCount>0 is never evicted. + // Atomic inc/dec: pin/unpin run concurrently under the shared lock. + volatile long pinCount[numCachePage + 1] = {}; - // Monitoring counters (updated under memLock; read lock-free by the stats printer — a stale - // read is harmless). pinnedHighWater vs getNumCachePage() is the headroom metric. - int pinnedNow = 0; // slots currently pinned - int pinnedHighWater = 0; // max slots pinned at once since start + // Monitoring. pinnedNow is atomic (concurrent pin/unpin); the rest best-effort (lock-free reads). + volatile long pinnedNow = 0; // slots currently pinned + volatile long pinnedHighWater = 0; // max slots pinned at once since start unsigned long long allPinnedWaits = 0; // times eviction had to wait for a pin (should be 0) unsigned long long cacheHits = 0; // page accesses served from cache unsigned long long cacheMisses = 0; // page accesses that loaded from disk - // Bumped by reset() (under memLock). Pins recorded under an older generation are invalid — - // their cache slot was wiped/reallocated — so the unpin and the dedup check ignore them. + // Bumped by reset() (exclusive); pins from an older generation are stale (slot was reused), so + // unpin and the dedup check ignore them. unsigned long long generation = 0; - // Pin/unpin a slot for the current thread (caller holds memLock). Idempotent per thread: - // a second access to a slot this thread already pinned is a no-op (dedup via the arena). + // In-flight load coordination: a miss does its disk IO with memLock RELEASED so cache hits don't + // stall, reserving the victim by stamping cachePageId=LOADING_PAGE_ID (find/eviction skip it). + // loadingTarget[] = page coming in (dedup duplicate loads); evictingPage[] = page being written + // back (a misser of it waits, vs reading stale disk). Both transient: INVALID on init/reset. + static constexpr unsigned long long LOADING_PAGE_ID = (unsigned long long)-2; + unsigned long long loadingTarget[numCachePage + 1]; + unsigned long long evictingPage[numCachePage + 1]; + + // Per-slot dirty byte (mprotect auto-dirty). Set by the SIGSEGV fast path on the first write after + // a slot is armed RO; read at eviction to skip the writeback for clean slots. Only meaningful when + // gSwapDirtyTrackEnabled (else every occupied victim is treated as dirty). + volatile unsigned char dirtyFlags[numCachePage + 1] = {}; + unsigned long long cleanEvicts = 0; // victims evicted with no writeback (clean) -- the saving + unsigned long long dirtyEvicts = 0; // victims that needed a writeback + + // Bytes between cache slots when the pool is page-aligned for dirty tracking (== pageSize; slots + // stay contiguous). 0 => not page-aligned (non-Linux, or pageSize not a whole number of OS pages) + // => dirty tracking is off for this VM and eviction writes back every occupied victim. + unsigned long long cacheSlotStride = 0; + + // Pin a slot for this thread (under shared or exclusive). Idempotent per thread (arena dedup). void pinSlotForThread(int slot) { - if (tlPinArena.contains(this, slot, generation)) + const unsigned long long g = generation; + if (tlPinArena.contains(this, slot, g)) return; - if (pinCount[slot]++ == 0) + if (_InterlockedIncrement(&pinCount[slot]) == 1) { - pinnedNow++; - if (pinnedNow > pinnedHighWater) - pinnedHighWater = pinnedNow; + const long now = _InterlockedIncrement(&pinnedNow); + if (now > pinnedHighWater) + pinnedHighWater = now; } - tlPinArena.add(this, &SwapVirtualMemory::unpinThunk, slot, generation); + tlPinArena.add(this, &SwapVirtualMemory::unpinThunk, slot, g); } void unpinSlotInternal(int slot, unsigned long long gen) { - ACQUIRE(memLock); - // gen != generation => this pin predates a reset(); the slot was already cleared, so - // decrementing now would corrupt a fresh pin. Drop it. + // Shared: concurrent with other unpins/pins (pinCount atomic), exclusive of reset/eviction — + // which is why the generation read below can be plain. + lockShared(); + // gen mismatch => pin predates a reset() that cleared the slot; drop it, don't hit a fresh pin. if (gen == generation && slot >= 0 && slot <= (int)numCachePage && pinCount[slot] > 0) { - if (--pinCount[slot] == 0) - pinnedNow--; + if (_InterlockedDecrement(&pinCount[slot]) == 0) + _InterlockedDecrement(&pinnedNow); } - RELEASE(memLock); + unlockShared(); } static void unpinThunk(void* self, int slot, unsigned long long gen) { @@ -852,17 +902,10 @@ class SwapVirtualMemory : private VirtualMemory per-slot mprotect can't isolate slots; leave to base alloc + cacheSlotStride = pageSize; + const unsigned long long poolBytes = pageSize * (numCachePage + 1); + void* aligned = aligned_alloc((size_t)ps, (size_t)poolBytes); + if (aligned == NULL) + return false; + setMem(aligned, poolBytes, 0); + currentPage = (T*)aligned; + for (int i = 0; i <= numCachePage; i++) + cache[i] = (T*)((char*)aligned + (unsigned long long)i * pageSize); + return true; + } + void registerDirtyTrackPool() + { + if (cacheSlotStride) // only register pools we actually page-aligned + SwapDirtyTrack::registerPool((unsigned char**)¤tPage, cacheSlotStride, numCachePage + 1, dirtyFlags); + } + void mprotectSlot(int slot, int prot) + { + if (gSwapDirtyTrackEnabled && cacheSlotStride) + mprotect((void*)cache[slot], cacheSlotStride, prot); + } +#endif + void unprotectSlotForLoad(int slot) // make the slot writable so an fread can land in it + { +#if defined(__linux__) + mprotectSlot(slot, PROT_READ | PROT_WRITE); +#endif + } + void armSlotAfterLoad(int slot) // mark clean + write-protect: the next write to it faults -> dirty + { +#if defined(__linux__) + if (gSwapDirtyTrackEnabled && cacheSlotStride) { - cacheHits++; - return cache_page_id; + dirtyFlags[slot] = 0; + mprotectSlot(slot, PROT_READ); } - cacheMisses++; +#endif + } + void unprotectWholePool() // drop write-protection on the whole pool + clear dirty (before reset's bulk setMem) + { +#if defined(__linux__) + if (gSwapDirtyTrackEnabled && cacheSlotStride && currentPage != NULL) + mprotect((void*)cache[0], cacheSlotStride * (numCachePage + 1), PROT_READ | PROT_WRITE); +#endif + setMem((void*)dirtyFlags, sizeof(dirtyFlags), 0); + } + + // Read pageId's bytes into cache[slot] (decompress if enabled), or zero-fill if never written. + // Returns false on exhausted IO retries. Runs with memLock RELEASED (slot is reserved LOADING). + bool loadPageBytesIntoSlot(unsigned long long pageId, int slot) + { CHAR16 pageName[64]; generatePageName(pageName, pageId); - cache_page_id = getMostOutdatedCachePageExceptCurrentPage(); - // Every eligible slot is currently pinned. Release memLock so pin holders can make - // progress (their work-unit boundary releases the pins), then retry. With numCachePage - // sized above the max pages pinned concurrently this never triggers; the bounded wait - // turns an under-sizing into a loud failure rather than silent corruption. - if (cache_page_id == -1) - allPinnedWaits++; - int allPinnedWaitMs = 0; - while (cache_page_id == -1) - { - RELEASE(memLock); - sleepMilliseconds(1); - ACQUIRE(memLock); - int rechecked = findCachePage(pageId); - if (rechecked != -1) - return rechecked; - cache_page_id = getMostOutdatedCachePageExceptCurrentPage(); - if (++allPinnedWaitMs > 5000) // ~5s - { - setText(message, L"Fatal: swapVM all cache pages pinned (numCachePage too small) | prefix "); - appendText(message, pageDir); - logToConsole(message); - exit(1); - } - } - if (cachePageId[cache_page_id] != INVALID_PAGE_ID) - { - writePageToDisk(cachePageId[cache_page_id]); - } -#if false - auto sz = load(pageName, pageSize, (unsigned char*)cache[cache_page_id], pageDir); - lastAccessedTimestamp[cache_page_id] = now_ms(); -#else -#if !defined(NDEBUG) - { - CHAR16 debugMsg[128]; - setText(debugMsg, L"Trying to load OLD page: "); - appendNumber(debugMsg, pageId, true); - addDebugMessage(debugMsg); - } -#endif unsigned long long sz = 0; if (isPageWrittenToDisk[pageId]) { - // bounded retry on load() failure unsigned int delayMs = SWAPVM_IO_INITIAL_DELAY_MS; for (int attempt = 0; attempt < SWAPVM_IO_MAX_ATTEMPTS; attempt++) { @@ -1004,7 +1084,7 @@ class SwapVirtualMemory : private VirtualMemory decompressed = Zipper::unzip(tmp.data(), (size_t)compressedSize, 4); if (decompressed.size() == pageSize) { - copyMem(cache[cache_page_id], decompressed.data(), pageSize); + copyMem(cache[slot], decompressed.data(), pageSize); sz = pageSize; } } @@ -1013,17 +1093,44 @@ class SwapVirtualMemory : private VirtualMemory 5000) // ~5s + { + setText(message, L"Fatal: swapVM all cache pages pinned (numCachePage too small) | prefix "); + appendText(message, pageDir); + logToConsole(message); + exit(1); + } + } + if (inflightAppeared) + continue; + + // Reserve the victim under the lock; do the disk IO with the lock RELEASED. + const unsigned long long oldPage = cachePageId[victim]; + // Component B: skip the writeback for a slot never modified since load. Tracking off + // (or non-Linux) => slotDirty is always true => write back every occupied victim (Component A). + const bool slotDirty = !gSwapDirtyTrackEnabled || dirtyFlags[victim]; + const bool needWriteback = (oldPage != INVALID_PAGE_ID) && slotDirty; + if (oldPage != INVALID_PAGE_ID) + (needWriteback ? dirtyEvicts : cleanEvicts)++; + cachePageId[victim] = LOADING_PAGE_ID; // hide from find/eviction + loadingTarget[victim] = pageId; + evictingPage[victim] = needWriteback ? oldPage : INVALID_PAGE_ID; + const unsigned long long savedGeneration = generation; + + RELEASE(memLock); + if (needWriteback) + writePageToDisk(oldPage, victim); + unprotectSlotForLoad(victim); // RW so the fread can land + const bool loadOk = loadPageBytesIntoSlot(pageId, victim); + armSlotAfterLoad(victim); // clean + RO: next write faults + acquireMemLock(); + + // reset() during our IO bumped the generation and cleared the slot + in-flight state: + // our reservation is void, the victim may already belong to someone else -> retry fresh, + // touching nothing. + if (generation != savedGeneration) + continue; + + loadingTarget[victim] = INVALID_PAGE_ID; + evictingPage[victim] = INVALID_PAGE_ID; + if (!loadOk) + { + cachePageId[victim] = INVALID_PAGE_ID; + return -1; + } + cachePageId[victim] = pageId; // publish: now findable + lastAccessedTimestamp[victim] = now_ms(); + return victim; } -#endif - return cache_page_id; } // return the most outdated cache page that is neither the current page nor pinned. @@ -1076,6 +1284,8 @@ class SwapVirtualMemory : private VirtualMemory 0) // skip pinned pages (still referenced) continue; if (lastAccessedTimestamp[i] == 0) @@ -1102,11 +1312,29 @@ class SwapVirtualMemory : private VirtualMemory 5000) + { + setText(message, L"Fatal: swapVM reset drain timeout (in-flight IO leak) | prefix "); + appendText(message, pageDir); + logToConsole(message); + exit(1); + } + } generation++; - setMem(pinCount, sizeof(pinCount), 0); + setMem((void*)pinCount, sizeof(pinCount), 0); + setMem(loadingTarget, sizeof(loadingTarget), 0xff); // INVALID_PAGE_ID + setMem(evictingPage, sizeof(evictingPage), 0xff); pinnedNow = 0; // cumulative stats (hits/misses/highWater/waits) are kept across resets RELEASE(memLock); + unprotectWholePool(); // RW + clear dirty so the base's bulk pool-zero below cannot fault VMBase::reset(); setMem(isPageWrittenToDisk, isPageWrittenToDiskSize, 0); if (mode == SwapMode::OFFSET_MODE) { @@ -1114,8 +1342,18 @@ class SwapVirtualMemory : private VirtualMemory exclusive (may evict + load). + acquireMemLock(); currentPageId = requested_page_id > currentPageId ? requested_page_id : currentPageId; - int cache_page_idx = loadPageToCacheAndTryToPersist(requested_page_id); + cache_page_idx = loadPageToCacheAndTryToPersist(requested_page_id); if (cache_page_idx == -1) { setText(message, L"Fatal Error: Invalid cache page index | Line "); @@ -1179,12 +1452,27 @@ class SwapVirtualMemory : private VirtualMemory exclusive (may evict + load). + acquireMemLock(); currentPageId = requested_page_id > currentPageId ? requested_page_id : currentPageId; - int cache_page_idx = loadPageToCacheAndTryToPersist(requested_page_id); + cache_page_idx = loadPageToCacheAndTryToPersist(requested_page_id); if (cache_page_idx == -1) { setText(message, L"Fatal Error: Invalid cache page index | Line "); @@ -1204,7 +1492,7 @@ class SwapVirtualMemory : private VirtualMemory= maxBytesPerElement) + { + lockShared(); + int idx = findCachePage(pageId); + if (idx != -1) + { + cacheHits++; // best-effort stat + pinSlotForThread(idx); + lastAccessedTimestamp[idx] = now_ms(); // refresh LRU on hit (racy write, harmless) + T* hit = (T*)((unsigned char*)cache[idx] + offsetInPage); + unlockShared(); + return hit; + } + unlockShared(); + } + + // Slow path: miss or page-straddle -> exclusive. + T* result = nullptr; + long long lastElementLength = 0; + acquireMemLock(); + currentPageId = pageId > currentPageId ? pageId : currentPageId; int cache_page_idx = loadPageToCacheAndTryToPersist(pageId); if (cache_page_idx == -1) { @@ -1298,7 +1606,15 @@ class SwapVirtualMemory : private VirtualMemoryinterfaceIndex == OI::DogeShareValidation::oracleInterfaceIndex) { // Look up query tx to get query data. @@ -4447,9 +4461,11 @@ static void beginEpoch() static void endEpoch() { logger.registerNewTx(system.tick, logger.SC_END_EPOCH_TX); + logToConsole(L"endEpoch: [1/5] running contract END_EPOCH procedures..."); contractProcessorPhase = END_EPOCH; contractProcessorState = 1; WAIT_WHILE(contractProcessorState); + logToConsole(L"endEpoch: [1/5] contract END_EPOCH procedures done"); // treating endEpoch as a tick, start updating etalonTick: // this is the last tick of an epoch, should we set prevResourceTestingDigest to zero? nodes that start from scratch (for the new epoch) @@ -4461,6 +4477,7 @@ static void endEpoch() etalonTick.prevTransactionBodyDigest = etalonTick.saltedTransactionBodyDigest; // Handle IPO + logToConsole(L"endEpoch: [2/5] finishing IPOs..."); finishIPOs(); system.initialMillisecond = etalonTick.millisecond; @@ -4475,6 +4492,7 @@ static void endEpoch() // Only issue qus if the max supply is not yet reached if (spectrumInfo.totalAmount + ISSUANCE_RATE <= MAX_SUPPLY) { + logToConsole(L"endEpoch: [3/5] computing revenue (V2/multi-dim) + distributing to computors..."); // Collect mining scores for V2 for (unsigned int i = 0; i < NUMBER_OF_COMPUTORS; i++) @@ -4558,6 +4576,7 @@ static void endEpoch() } // Reorganize spectrum hash map (also updates spectrumInfo) + logToConsole(L"endEpoch: [4/5] reorganizing spectrum hash map..."); { ACQUIRE(spectrumLock); @@ -4566,7 +4585,9 @@ static void endEpoch() RELEASE(spectrumLock); } + logToConsole(L"endEpoch: [5/5] reorganizing universe/assets..."); assetsEndEpoch(); + logToConsole(L"endEpoch: [5/5] universe/assets done"); { // this is the last logging event of the epoch // a hint message for 3rd party services the end of the epoch @@ -6312,6 +6333,10 @@ static void tickProcessor(void*, unsigned long long processorNumber) unsigned int latestProcessedTick = 0; while (!shutDownNode) { +#ifdef TESTNET + std::this_thread::sleep_for(std::chrono::microseconds(500)); +#endif + PROFILE_NAMED_SCOPE("tickProcessor(): loop iteration"); if (tlPinArena.count != 0) // leaked swap pins from a prior iteration: a missing/removed PinScope boundary { @@ -6785,6 +6810,30 @@ static void tickProcessor(void*, unsigned long long processorNumber) gTickNumberOfComputors = tickNumberOfComputors; gTickTotalNumberOfComputors = tickTotalNumberOfComputors; + // K12 state-cache safety net: only at ticks where the computer digest is consensus-validated. + // If enough computors voted but quorum doesn't match our etalon, an incremental-cache error + // could be the cause, so recompute the computer digest canonically (cache bypassed) and re-tally + // once. A genuine divergence recomputes to the same value and falls through to normal recovery. + if (K12StateDigestCache::gK12StateDigestCacheEnabled + && (isSystemAtSecurityTick() || isLastTickInEpoch()) + && tickNumberOfComputors < QUORUM && tickTotalNumberOfComputors >= QUORUM) + { + static unsigned int gK12RecheckedTick = 0; + if (gK12RecheckedTick != system.tick) + { + gK12RecheckedTick = system.tick; + const m256i before = etalonTick.saltedComputerDigest; + recomputeComputerDigestFull(etalonTick.saltedComputerDigest); + if (etalonTick.saltedComputerDigest != before) + { + logToConsole(L"K12 state-cache computer digest disagreed with quorum; recomputed canonically, re-tallying votes"); + updateVotesCount(tickNumberOfComputors, tickTotalNumberOfComputors, quorumComputerDigest); + gTickNumberOfComputors = tickNumberOfComputors; + gTickTotalNumberOfComputors = tickTotalNumberOfComputors; + } + } + } + if (tickNumberOfComputors >= QUORUM) { tryForceEmptyNextTick(); @@ -6902,12 +6951,25 @@ static void tickProcessor(void*, unsigned long long processorNumber) bool isBeginEpoch = false; if (epochTransitionState == 1) { + { + CHAR16 etMsg[192]; + setText(etMsg, L"=== EPOCH TRANSITION: start | epoch "); + appendNumber(etMsg, system.epoch, FALSE); + appendText(etMsg, L" -> "); + appendNumber(etMsg, system.epoch + 1, FALSE); + appendText(etMsg, L" | last tick of epoch "); + appendNumber(etMsg, system.tick - 1, FALSE); + logToConsole(etMsg); + } // wait until all request processors are in waiting state + logToConsole(L"EPOCH TRANSITION: waiting for request processors to park..."); WAIT_WHILE(epochTransitionWaitingRequestProcessors < nRequestProcessorIDs); // end current epoch + logToConsole(L"EPOCH TRANSITION: running endEpoch() (revenue/IPO/spectrum reorg)..."); endEpoch(); + logToConsole(L"EPOCH TRANSITION: endEpoch() done"); // Save the file of revenue. This blocking save can be called from any thread // Revenue v2 data @@ -6924,12 +6986,15 @@ static void tickProcessor(void*, unsigned long long processorNumber) commonBuffers.releaseBuffer(reorgBuffer); // instruct main loop to save system and wait until it is done + logToConsole(L"EPOCH TRANSITION: saving system state..."); systemMustBeSaved = true; WAIT_WHILE(systemMustBeSaved); epochTransitionState = 2; + logToConsole(L"EPOCH TRANSITION: running beginEpoch()..."); beginEpoch(); isBeginEpoch = true; + logToConsole(L"EPOCH TRANSITION: beginEpoch() done"); // Some debug checks that we are ready for the next epoch ASSERT(system.numberOfSolutions == 0); @@ -6945,6 +7010,7 @@ static void tickProcessor(void*, unsigned long long processorNumber) ASSERT(minimumComputorScore == 0 && minimumCandidateScore == 0); // instruct main loop to save files and wait until it is done + logToConsole(L"EPOCH TRANSITION: saving spectrum/universe/computer..."); spectrumMustBeSaved = true; universeMustBeSaved = true; computerMustBeSaved = true; @@ -6958,6 +7024,16 @@ static void tickProcessor(void*, unsigned long long processorNumber) getComputerDigest(etalonTick.saltedComputerDigest); epochTransitionState = 0; + { + CHAR16 etMsg[192]; + setText(etMsg, L"=== EPOCH TRANSITION: COMPLETE | now epoch "); + appendNumber(etMsg, system.epoch, FALSE); + appendText(etMsg, L" | initialTick "); + appendNumber(etMsg, system.initialTick, FALSE); + appendText(etMsg, L" | tick "); + appendNumber(etMsg, system.tick, FALSE); + logToConsole(etMsg); + } } ASSERT(epochTransitionWaitingRequestProcessors >= 0 && epochTransitionWaitingRequestProcessors <= nRequestProcessorIDs); @@ -7001,15 +7077,6 @@ static void tickProcessor(void*, unsigned long long processorNumber) logToConsole(message); } - // Flip forceDontUseSecurityTick flag based on stack - while (!forceDontUseSecurityTickChangeStack.empty()) - { - forceDontUseSecurityTickChangeStack.pop_back(); - forceDontUseSecurityTick = !forceDontUseSecurityTick; - setText(message, L"forceDontUseSecurityTick is now ");; - appendText(message, forceDontUseSecurityTick ? L"ON" : L"OFF"); - logToConsole(message); - } } } } @@ -7345,7 +7412,11 @@ static bool initialize() for (unsigned int contractIndex = 0; contractIndex < contractCount; contractIndex++) { unsigned long long size = contractDescriptions[contractIndex].stateSize; - if (!allocPoolWithErrorLog(L"contractStates", size, (void**)&contractStates[contractIndex], __LINE__)) + // cached contracts need page-aligned state for per-chunk mprotect; off-path keeps the 64-byte alloc + const bool contractStateAllocOk = K12StateDigestCache::wantsPageAlignedAlloc(size) + ? K12StateDigestCache::allocStatePool(size, (void**)&contractStates[contractIndex]) + : allocPoolWithErrorLog(L"contractStates", size, (void**)&contractStates[contractIndex], __LINE__); + if (!contractStateAllocOk) { return false; } @@ -7748,6 +7819,9 @@ static bool initialize() emptyTickResolver.tick = 0; emptyTickResolver.lastTryClock = 0; + // contract states are now allocated, loaded and constructed; arm the per-chunk K12 digest cache + K12StateDigestCache::init(); + return true; } @@ -8338,13 +8412,6 @@ static void processKeyPresses() logToConsole(L"Requesting for force skip checking computer digest for this tick."); forceDontCheckComputerDigest = true; break; - case 's': - forceDontUseSecurityTickChangeStack.push_back(1); - // forceDontUseSecurityTick = !forceDontUseSecurityTick; - // setText(message, L"forceDontUseSecurityTick is now ");; - // appendText(message, forceDontUseSecurityTick ? L"ON" : L"OFF"); - logToConsole(message); - break; } @@ -8909,6 +8976,9 @@ EFI_STATUS efi_main(EFI_HANDLE imageHandle, EFI_SYSTEM_TABLE* systemTable) logToConsole(L"Init complete! Entering main loop ..."); while (!shutDownNode) { +#ifdef TESTNET + std::this_thread::sleep_for(std::chrono::microseconds(500)); +#endif PinScope _pinScope; // release swap-page pins taken during this main-loop iteration if (criticalSituation == 1) { @@ -9574,6 +9644,7 @@ void processArgs(int argc, const char* argv[]) { logColorToScreen("INFO", "Total RAM required " + std::to_string(getTotalRam() / (1024 * 1024 * 1024)) + " GB"); cxxopts::Options options("Qubic Core Lite", "The lite version of Qubic Core that can run directly on the OS without a UEFI environment."); + options.allow_unrecognised_options(); options.add_options() ("p,peers", "Public peers", cxxopts::value()) ("m,mode", "Core mode", cxxopts::value()) @@ -9589,19 +9660,38 @@ void processArgs(int argc, const char* argv[]) { ("oa,operator-alias", "Operator alias for RPC tick-info", cxxopts::value()) ("fv, force-verify-solutions", "Passcode to access http server", cxxopts::value()) ("fbis, force-broadcast-invalid-solution", "TEST: each tick, broadcast a random-nonce solution tx signed by a random own-computor to exercise the reprocessSolutionTransaction() rollback path", cxxopts::value()) - ("s,security-tick", "Core will verify state after x tick, to reduce computational to the node", cxxopts::value()->default_value("1")) ("http-port", "Port for the built-in HTTP/RPC server to listen on", cxxopts::value()->default_value("41841")) ("static-peers", "Run in static peer mode: do not add/remove peers, do not churn 25% of non-fullnode peers every 2 minutes, do not accept new incoming connections. Useful for nodes far from the network's center of mass where the default churn drops good peers before they're classified as fullnodes.") ("swap-compression", "Compress SwapVM disk pages with blosc2 on save/load (Linux only). Trades CPU for less disk I/O and footprint. Off by default.") + ("swap-dirty-track", "Auto-track dirty SwapVM cache pages via mprotect+SIGSEGV (Linux only): skip the writeback (and compression) for pages never modified since load. Trades a small mprotect/fault cost for less disk I/O. Off by default.") ("auto-flush-stuck-seconds", "If the tick processor sits on the same system.tick for longer than N seconds, automatically wipe the local tickData of system.tick+1 so the request loop re-fetches it from peers. 0 disables. Reasonable production values: 60-120. Recovers automatically from corrupt-tickData stalls.", cxxopts::value()->default_value("0")) + ("no-k12-state-cache", "Disable the incremental smart-contract state digest cache (Linux). Default on: only changed 8KB chunks are re-hashed via mprotect+SIGSEGV dirty tracking. Pass to fall back to full one-shot K12 every tick.") + ("k12-state-cache-verify", "Self-check the K12 state-digest cache: each digest also runs the one-shot and stalls loudly on any mismatch. For soak/CI; small per-tick cost. Off by default.") ("max-inbound", "Max number of inbound connection slots that may accept. Lower during catch-up to stop serving inbound peers (0 = reject all inbound, like static). Default = all incoming slots.", cxxopts::value()->default_value("-1")); auto result = options.parse(argc, argv); + auto unmatched = result.unmatched(); + for (auto& u : unmatched) { + std::cerr << "Warning: unknown option: " << u << "\n"; + } + #ifdef __linux__ if (result.count("swap-compression")) { gSwapCompressionEnabled = true; logColorToScreen("INFO", "Swap compression enabled: SwapVM disk pages will be compressed with blosc2 on save/load"); } + if (result.count("swap-dirty-track")) { + gSwapDirtyTrackEnabled = true; + logColorToScreen("INFO", "Swap dirty tracking enabled: clean SwapVM cache pages skip writeback on eviction"); + } + if (result.count("no-k12-state-cache")) { + K12StateDigestCache::gK12StateDigestCacheEnabled = false; + logColorToScreen("INFO", "K12 state-digest cache disabled: full one-shot K12 every tick"); + } + if (result.count("k12-state-cache-verify")) { + K12StateDigestCache::gK12StateDigestCacheVerify = true; + logColorToScreen("INFO", "K12 state-digest cache self-verify enabled: each digest also runs the one-shot"); + } #endif if (result.count("peers")) { @@ -9649,10 +9739,6 @@ void processArgs(int argc, const char* argv[]) { logColorToScreen("INFO", "Lite node operator ID: " + myOperatorId + (!isOperatorIdProvided ? " (default)" : "")); } - if (result.count("security-tick")) { - securityTick = result["security-tick"].as(); - logColorToScreen("INFO", "Security tick set to " + std::to_string(securityTick)); - } { int mi = result["max-inbound"].as(); @@ -9896,7 +9982,15 @@ void watchAndCheckin() #endif #ifdef __linux__ -void signalHandler(int sig) { +void signalHandler(int sig, siginfo_t* si, void* /*ucontext*/) { + // Component B fast path: a write to an armed read-only SwapVM cache page faults here. Mark the + // slot dirty, restore write access, and resume the store. Any other fault hits the crash path below. + if (sig == SIGSEGV && si && SwapDirtyTrack::tryMarkDirty(si->si_addr)) + return; + // A write to an armed read-only contract-state chunk faults here: mark the chunk dirty (needs si_addr), + // restore write access, resume the store. Only changed chunks are re-hashed on the next digest. + if (sig == SIGSEGV && si && K12StateDigestCache::tryMarkDirty(si->si_addr)) + return; boost::stacktrace::safe_dump_to("crash.dump"); // Send to server in a child process pid_t pid = fork(); @@ -9936,7 +10030,8 @@ void signalHandler(int sig) { void setupSignalHandlers() { struct sigaction sa; memset(&sa, 0, sizeof(sa)); - sa.sa_handler = signalHandler; + sa.sa_sigaction = signalHandler; // SA_SIGINFO form: handler receives siginfo_t* (si_addr) + sa.sa_flags = SA_SIGINFO; sigemptyset(&sa.sa_mask); // Common crash signals to catch: diff --git a/src/ticking/tick_storage.h b/src/ticking/tick_storage.h index e0dda04d..71e03142 100644 --- a/src/ticking/tick_storage.h +++ b/src/ticking/tick_storage.h @@ -720,6 +720,10 @@ class TickStorage appendNumber(m, vm.getCacheHits(), false); appendText(m, (CHAR16*)L" miss "); appendNumber(m, vm.getCacheMisses(), false); + appendText(m, (CHAR16*)L" cleanEvict "); + appendNumber(m, vm.getCleanEvicts(), false); + appendText(m, (CHAR16*)L" dirtyEvict "); + appendNumber(m, vm.getDirtyEvicts(), false); logToConsole(m); } #endif @@ -1030,6 +1034,7 @@ class TickStorage #ifndef NDEBUG for (unsigned int tickIndex = 0; tickIndex < MAX_NUMBER_OF_TICKS_PER_EPOCH; tickIndex++) { + PinScope _pinScope; // release this tick's swap-page pins each iteration (debug consistency scan over the whole epoch) TickData &tickData = TickStorage::tickData[tickIndex]; ASSERT(isAllBytesZero(&tickData, sizeof(tickData))); diff --git a/test/kangaroo_twelve.cpp b/test/kangaroo_twelve.cpp index 038b302b..23b3c6bf 100644 --- a/test/kangaroo_twelve.cpp +++ b/test/kangaroo_twelve.cpp @@ -8,11 +8,15 @@ #include "../src/K12/kangaroo_twelve_xkcp.h" #include "../src/kangaroo_twelve.h" #include "../src/platform/memory.h" +#include "../src/extensions/k12_state_digest_cache.h" #include #include "gtest/gtest.h" #include #include +#include +#include +#include TEST(TestCoreK12, PerformanceDigest32Of1GB) @@ -95,3 +99,155 @@ TEST(TestCoreK12, CompareK12Implementations) ASSERT_EQ(memcmp(outputArrayXKCP, outputArray, outputN), 0); delete [] inputPtr; } + +// ---- K12 incremental state-digest cache (src/extensions/k12_state_digest_cache.h) ---- +// incrementalDigest must be byte-identical to the one-shot KangarooTwelve. Dirty bits are injected +// directly (no mprotect needed); the oracle is the native one-shot used by consensus. + +static void k12sdc_fillRandom(unsigned char* p, size_t n) +{ + size_t i = 0; + for (; i + 4 <= n; i += 4) { unsigned int v; _rdrand32_step(&v); memcpy(p + i, &v, 4); } + for (; i < n; ++i) { unsigned int v; _rdrand32_step(&v); p[i] = (unsigned char)v; } +} + +TEST(TestCoreK12, IncrementalAllDirtyEqualsOneShot) +{ + const size_t sizes[] = { 1, 100, 8191, 8192, 8193, 16383, 16384, 16385, + 8192 * 3, 8192 * 3 + 1, 8192 * 100 + 777, 10u * 1024 * 1024 + 777 }; + for (size_t S : sizes) + { + std::vector buf(S); + k12sdc_fillRandom(buf.data(), S); + const size_t numChunks = (S + 8192 - 1) / 8192; + std::vector cv(numChunks * 32, 0), dirty(numChunks, 1); + unsigned char a[32], b[32]; + KangarooTwelve(buf.data(), (unsigned int)S, a, 32); + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), b); + ASSERT_EQ(memcmp(a, b, 32), 0) << "all-dirty mismatch at size " << S; + } +} + +TEST(TestCoreK12, IncrementalSelectiveDirtyAfterMutation) +{ + const size_t S = 10u * 1024 * 1024 + 777; + std::vector buf(S); + k12sdc_fillRandom(buf.data(), S); + const size_t numChunks = (S + 8192 - 1) / 8192; + std::vector cv(numChunks * 32, 0), dirty(numChunks, 1); + unsigned char d[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), d); // populate cv + std::fill(dirty.begin(), dirty.end(), 0); + for (size_t k = 1; k + 1 < numChunks; k += 7) + { + unsigned int off; _rdrand32_step(&off); + buf[k * 8192 + (off % 8192)] ^= 0xA5; + dirty[k] = 1; // only the touched interior full chunk + } + unsigned char c[32], e[32]; + KangarooTwelve(buf.data(), (unsigned int)S, c, 32); + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), e); + ASSERT_EQ(memcmp(c, e, 32), 0); +} + +TEST(TestCoreK12, IncrementalCleanReuseNoMutation) +{ + const size_t S = 8192 * 50 + 33; + std::vector buf(S); + k12sdc_fillRandom(buf.data(), S); + const size_t numChunks = (S + 8192 - 1) / 8192; + std::vector cv(numChunks * 32, 0), dirty(numChunks, 1); + unsigned char first[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), first); + std::fill(dirty.begin(), dirty.end(), 0); // all clean -> pure reuse + unsigned char reuse[32], oneShot[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), reuse); + KangarooTwelve(buf.data(), (unsigned int)S, oneShot, 32); + ASSERT_EQ(memcmp(first, reuse, 32), 0); + ASSERT_EQ(memcmp(reuse, oneShot, 32), 0); +} + +TEST(TestCoreK12, IncrementalStaleDirtyIsHarmless) +{ + const size_t S = 8192 * 20 + 5; + std::vector buf(S); + k12sdc_fillRandom(buf.data(), S); + const size_t numChunks = (S + 8192 - 1) / 8192; + std::vector cv(numChunks * 32, 0), dirty(numChunks, 1); + unsigned char base[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), base); + std::fill(dirty.begin(), dirty.end(), 0); // mark some dirty WITHOUT mutating + for (size_t k = 0; k < numChunks; k += 3) dirty[k] = 1; + unsigned char again[32], oneShot[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), again); + KangarooTwelve(buf.data(), (unsigned int)S, oneShot, 32); + ASSERT_EQ(memcmp(again, oneShot, 32), 0); + ASSERT_EQ(memcmp(again, base, 32), 0); +} + +TEST(TestCoreK12, IncrementalChunk0AndTailNeedNoDirtyBit) +{ + const size_t S = 8192 * 8 + 1234; // multi-chunk with partial tail + std::vector buf(S); + k12sdc_fillRandom(buf.data(), S); + const size_t numChunks = (S + 8192 - 1) / 8192; + std::vector cv(numChunks * 32, 0), dirty(numChunks, 1); + unsigned char tmp[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), tmp); // populate + std::fill(dirty.begin(), dirty.end(), 0); + buf[123] ^= 0x5A; // mutate chunk 0 only, no dirty bit + unsigned char r0[32], o0[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), r0); + KangarooTwelve(buf.data(), (unsigned int)S, o0, 32); + ASSERT_EQ(memcmp(r0, o0, 32), 0) << "chunk 0 not re-absorbed"; + buf[S - 7] ^= 0x3C; // mutate partial tail only, no dirty bit + unsigned char r1[32], o1[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), r1); + KangarooTwelve(buf.data(), (unsigned int)S, o1, 32); + ASSERT_EQ(memcmp(r1, o1, 32), 0) << "partial tail not re-absorbed"; +} + +TEST(TestCoreK12, IncrementalSingleChunkPath) +{ + for (size_t S : { (size_t)1, (size_t)8192 }) + { + std::vector buf(S); + k12sdc_fillRandom(buf.data(), S); + std::vector cv(32, 0), dirtyOn(1, 1), dirtyOff(1, 0); + unsigned char a[32], b[32], o[32]; + KangarooTwelve(buf.data(), (unsigned int)S, o, 32); + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirtyOn.data(), a); + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirtyOff.data(), b); + ASSERT_EQ(memcmp(a, o, 32), 0) << "single-chunk dirty at size " << S; + ASSERT_EQ(memcmp(b, o, 32), 0) << "single-chunk clean at size " << S; + } +} + +TEST(TestCoreK12, IncrementalCacheLifecycle) +{ + const size_t S = 8192 * 40 + 99; + std::vector buf(S); + k12sdc_fillRandom(buf.data(), S); + const size_t numChunks = (S + 8192 - 1) / 8192; + std::vector cv(numChunks * 32, 0), dirty(numChunks, 1); + unsigned char dig[32], oneShot[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), dig); // initial build + for (int round = 0; round < 5; ++round) + { + std::fill(dirty.begin(), dirty.end(), 0); + for (int m = 0; m < 3; ++m) // mutate a few interior full chunks this "tick" + { + unsigned int kk, off; _rdrand32_step(&kk); _rdrand32_step(&off); + const size_t k = 1 + (kk % (numChunks > 2 ? numChunks - 2 : 1)); + const size_t pos = k * 8192 + (off % 8192); + if (pos < S) { buf[pos] ^= 0x11; dirty[k] = 1; } + } + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), dig); + KangarooTwelve(buf.data(), (unsigned int)S, oneShot, 32); + ASSERT_EQ(memcmp(dig, oneShot, 32), 0) << "lifecycle round " << round; + std::fill(dirty.begin(), dirty.end(), 0); // clean recompute equals the same + unsigned char clean[32]; + K12StateDigestCache::incrementalDigest(buf.data(), S, cv.data(), dirty.data(), clean); + ASSERT_EQ(memcmp(clean, oneShot, 32), 0) << "lifecycle clean round " << round; + } +} diff --git a/test/virtual_memory.cpp b/test/virtual_memory.cpp index 6b14d6cf..3c50e1ba 100644 --- a/test/virtual_memory.cpp +++ b/test/virtual_memory.cpp @@ -7,6 +7,12 @@ #include #include +#include +#include +#include +#if defined(__linux__) +#include +#endif #include "network_messages/transactions.h" @@ -489,4 +495,107 @@ TEST(TestSwapVirtualMemory, TestSwapVirtualMemory_TestCacheBuffer) EXPECT_TRUE((uint64_t)tx == (uint64_t)test_vm.getCacheBuffer(0) + i * test_vm.getPageSize()); } } -} \ No newline at end of file +} + +namespace { +struct E16 { unsigned long long a, b; }; // 16 bytes; pageCapacity 256 -> pageSize 4096 (one OS page) +} + +// Component A: every page must survive eviction+reload even though the working set (64 pages) far +// exceeds the cache (4 slots). Small + fast, exercising the IO-outside-the-lock evict/load path. +TEST(TestSwapVirtualMemory, IndexEvictReloadIntegritySmall) { + initFilesystem(); + registerAsynFileIO(NULL); + SwapVirtualMemory vm; + vm.init(); + const unsigned long long NPAGES = 64, CAP = 256; + for (unsigned long long p = 0; p < NPAGES; p++) { + PinScope _; + E16& e = vm.getRef(p * CAP); + e.a = p + 1; e.b = (p + 1) * 7; + } + for (unsigned long long p = 0; p < NPAGES; p++) { + PinScope _; + E16& e = vm.getRef(p * CAP); + EXPECT_EQ(e.a, p + 1); + EXPECT_EQ(e.b, (p + 1) * 7); + } +} + +// Component A: cache hits must keep working (no corruption, no deadlock) while another thread churns +// misses that evict+load under released memLock. A churner walks 200 pages (constant eviction) while +// four hitters hammer the resident sentinel page 0; page 0 is only ever written 0xABCD, so any other +// value read back signals a torn evict/reload. +TEST(TestSwapVirtualMemory, ConcurrentHitsDuringMiss) { + initFilesystem(); + registerAsynFileIO(NULL); + SwapVirtualMemory vm; + vm.init(); + const unsigned long long CAP = 256; + { PinScope _; vm.getRef(0).a = 0xABCD; } + + std::atomic stop{false}; + std::atomic errors{0}; + std::thread churner([&]{ + unsigned long long p = 1; + while (!stop.load(std::memory_order_relaxed)) { + PinScope _; + vm.getRef(p * CAP).a = p; + p = (p % 200) + 1; + } + }); + std::vector hitters; + for (int t = 0; t < 4; t++) hitters.emplace_back([&]{ + for (int i = 0; i < 100000; i++) { + PinScope _; + if (vm.getRef(0).a != 0xABCD) errors++; + } + }); + for (auto& h : hitters) h.join(); + stop = true; churner.join(); + EXPECT_EQ(errors.load(), 0); +} + +#if defined(__linux__) +// Component B: with dirty tracking on, a page only read since load is evicted without a writeback +// (clean), while a written page is written back and survives. Tests install the SIGSEGV fast path +// that the node installs in signalHandler (the test binary has no handler of its own). +TEST(TestSwapVirtualMemory, DirtyTrackSkipsCleanWriteback) { + initFilesystem(); + registerAsynFileIO(NULL); + + struct sigaction sa, oldSa; + memset(&sa, 0, sizeof(sa)); + sa.sa_sigaction = [](int sig, siginfo_t* si, void*) { + if (sig == SIGSEGV && si && SwapDirtyTrack::tryMarkDirty(si->si_addr)) return; + signal(SIGSEGV, SIG_DFL); raise(SIGSEGV); + }; + sa.sa_flags = SA_SIGINFO; + sigemptyset(&sa.sa_mask); + sigaction(SIGSEGV, &sa, &oldSa); + gSwapDirtyTrackEnabled = true; + + { + SwapVirtualMemory vm; + vm.init(); + const unsigned long long CAP = 256; + for (unsigned long long p = 0; p < 16; p++) { + PinScope _; + E16& e = vm.getRef(p * CAP); + if (p % 2 == 0) e.a = p + 100; // written -> dirty + else { volatile unsigned long long r = e.a; (void)r; } // read only -> clean + } + EXPECT_GT(vm.getCleanEvicts(), 0u); + EXPECT_GT(vm.getDirtyEvicts(), 0u); + for (unsigned long long p = 0; p < 16; p++) { + PinScope _; + E16& e = vm.getRef(p * CAP); + if (p % 2 == 0) EXPECT_EQ(e.a, p + 100); // dirty page written back + reloaded + else EXPECT_EQ(e.a, 0u); // clean page never written -> reloads as zero + } + } + + gSwapDirtyTrackEnabled = false; + sigaction(SIGSEGV, &oldSa, nullptr); +} +#endif \ No newline at end of file