From b6233961867051263334ac7c1c0a10fae3afc1a7 Mon Sep 17 00:00:00 2001 From: Krystie Date: Sat, 27 Jun 2026 18:17:59 -0700 Subject: [PATCH] perf+sec: 15 improvements across consensus, DB, network, sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CONSENSUS SECURITY (main.cpp): - Re-enable PoS kernel verification post-IBD (was unconditionally disabled) - Re-enable coinstake reward validation post-IBD (was commented out) - Re-enable anti-spam difficulty check (was if(false && ...)) SYNC PERFORMANCE (main.cpp): - Batch address index writes in ConnectBlock (hundreds of DB ops → one per address) - Throttle IBD printfs (per-block → per-10K-blocks or fDebug-gated) DATABASE (txdb-rocksdb.cpp/h, txdb-base.cpp): - Non-batched WriteRaw: WAL sync=false (was fsync per write) - UTXO cache: FIFO eviction → true LRU with access-order tracking - RocksDB memtable: 64MB → 256MB + max_write_buffer_number=4 - pendingBatch: std::map → std::unordered_map (O(log n) → O(1)) - max_open_files: 1000 → unlimited NETWORK (net.cpp, netbase.cpp): - TCP_NODELAY on all sockets (disable Nagle's algorithm) - SO_KEEPALIVE on all sockets (faster dead-peer detection) - Adaptive MilliSleep: 1ms during IBD, 10ms otherwise - writev() scatter-gather I/O for send() coalescing (up to 16 msgs/syscall) - O(1) CountInFlight counter (was O(n) scan of entire header map) --- src/main.cpp | 116 +++++++++++++++++++++++++++++-------------- src/net.cpp | 80 +++++++++++++++++++++++++---- src/netbase.cpp | 14 ++++++ src/syncmanager.cpp | 25 +++++++--- src/txdb-base.cpp | 56 +++++++++++++++------ src/txdb-rocksdb.cpp | 16 ++++-- src/txdb-rocksdb.h | 3 +- 7 files changed, 239 insertions(+), 71 deletions(-) diff --git a/src/main.cpp b/src/main.cpp index 48c5a60..1f740ba 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -337,8 +337,9 @@ bool AddOrphanTx(const CTransaction& tx) for (const CTxIn& txin : tx.vin) mapOrphanTransactionsByPrev[txin.prevout.hash].insert(hash); - printf("stored orphan tx %s (mapsz %" PRIszu ")\n", hash.ToString().substr(0,10).c_str(), - mapOrphanTransactions.size()); + if (fDebug) + printf("stored orphan tx %s (mapsz %" PRIszu ")\n", hash.ToString().substr(0,10).c_str(), + mapOrphanTransactions.size()); return true; } @@ -2026,10 +2027,13 @@ bool CBlock::ConnectBlock(CTxDBBase& txdb, CBlockIndex* pindex, bool fJustCheck) int64_t nCalculatedStakeReward = GetProofOfStakeReward(nCoinAge, nFees); - // TEMP: Skip coinstake reward check during sync — UTXO set incomplete causes nCalculatedStakeReward=0 - // Will re-enable after full sync completes - // if (nStakeReward > nCalculatedStakeReward) - // return DoS(100, error("ConnectBlock() : coinstake pays too much(actual=%" PRId64 " vs calculated=%" PRId64 ")", nStakeReward, nCalculatedStakeReward)); + // Enforce coinstake reward check only after IBD completes. + // During IBD the UTXO set is incomplete, causing nCalculatedStakeReward=0. + if (!IsInitialBlockDownload()) + { + if (nStakeReward > nCalculatedStakeReward) + return DoS(100, error("ConnectBlock() : coinstake pays too much(actual=%" PRId64 " vs calculated=%" PRId64 ")", nStakeReward, nCalculatedStakeReward)); + } } } @@ -2085,6 +2089,11 @@ bool CBlock::ConnectBlock(CTxDBBase& txdb, CBlockIndex* pindex, bool fJustCheck) // Update address index if (fAddressIndex) { + // Batch balance deltas: accumulate net change per address, then + // do a single read-modify-write per unique address at the end. + // This avoids hundreds of per-output DB reads/writes per block. + std::map, int64_t> mapBalanceDeltas; + for (unsigned int i = 0; i < vtx.size(); i++) { const CTransaction& tx = vtx[i]; @@ -2096,24 +2105,41 @@ bool CBlock::ConnectBlock(CTxDBBase& txdb, CBlockIndex* pindex, bool fJustCheck) for (unsigned int j = 0; j < tx.vin.size(); j++) { const CTxIn& txin = tx.vin[j]; - CTransaction txPrev; - CTxIndex txindex; - if (txdb.ReadDiskTx(txin.prevout.hash, txPrev, txindex)) + bool fFoundPrevout = false; + + // Check mapPendingUtxos first to avoid a DB hit for + // outputs created earlier in this same block. + auto itPending = mapPendingUtxos.find(txin.prevout); + if (itPending != mapPendingUtxos.end()) { - if (txin.prevout.n < txPrev.vout.size()) + const CUtxoEntry& utxo = itPending->second; + int nType; + uint160 hashBytes; + if (GetAddressFromScript(utxo.scriptPubKey, nType, hashBytes)) { - const CTxOut& prevout = txPrev.vout[txin.prevout.n]; - int nType; - uint160 hashBytes; - if (GetAddressFromScript(prevout.scriptPubKey, nType, hashBytes)) + txdb.EraseAddressUtxo(nType, hashBytes, txin.prevout.hash, txin.prevout.n); + mapBalanceDeltas[std::make_pair(nType, hashBytes)] -= utxo.nValue; + } + fFoundPrevout = true; + } + + // Fall back to reading the full transaction from disk + if (!fFoundPrevout) + { + CTransaction txPrev; + CTxIndex txindex; + if (txdb.ReadDiskTx(txin.prevout.hash, txPrev, txindex)) + { + if (txin.prevout.n < txPrev.vout.size()) { - // Remove spent UTXO - txdb.EraseAddressUtxo(nType, hashBytes, txin.prevout.hash, txin.prevout.n); - // Decrease balance - int64_t nBalance = 0; - txdb.ReadAddressBalance(nType, hashBytes, nBalance); - nBalance -= prevout.nValue; - txdb.WriteAddressBalance(nType, hashBytes, nBalance); + const CTxOut& prevout = txPrev.vout[txin.prevout.n]; + int nType; + uint160 hashBytes; + if (GetAddressFromScript(prevout.scriptPubKey, nType, hashBytes)) + { + txdb.EraseAddressUtxo(nType, hashBytes, txin.prevout.hash, txin.prevout.n); + mapBalanceDeltas[std::make_pair(nType, hashBytes)] -= prevout.nValue; + } } } } @@ -2134,16 +2160,25 @@ bool CBlock::ConnectBlock(CTxDBBase& txdb, CBlockIndex* pindex, bool fJustCheck) // Add new UTXO txdb.WriteAddressUtxo(nType, hashBytes, txhash, k, txout.nValue, pindex->nHeight, txout.scriptPubKey); - // Increase balance - int64_t nBalance = 0; - txdb.ReadAddressBalance(nType, hashBytes, nBalance); - nBalance += txout.nValue; - txdb.WriteAddressBalance(nType, hashBytes, nBalance); + // Accumulate balance increase (batched write at end) + mapBalanceDeltas[std::make_pair(nType, hashBytes)] += txout.nValue; // Record tx in address history txdb.WriteAddressTxId(nType, hashBytes, pindex->nHeight, i, txhash); } } } + + // Batch-write all accumulated balance changes: one read + one write + // per unique address instead of per-output. + for (const auto& entry : mapBalanceDeltas) + { + if (entry.second == 0) + continue; + int64_t nBalance = 0; + txdb.ReadAddressBalance(entry.first.first, entry.first.second, nBalance); + nBalance += entry.second; + txdb.WriteAddressBalance(entry.first.first, entry.first.second, nBalance); + } } // Update block index on disk without changing it in memory. @@ -2946,12 +2981,20 @@ bool CBlock::AcceptBlock() uint256 hashProofOfStake = 0, targetProofOfStake = 0; if (IsProofOfStake()) { - // Skip expensive PoS kernel verification for blocks covered by hardcoded checkpoint. - // The checkpoint at height 2,186,940 already guarantees chain integrity. - // TEMP: Skip PoS kernel check during sync — read txPrev fails on incomplete index - // Will re-enable after full sync completes - printf("SKIP: PoS kernel check skipped for block %d during sync\n", nHeight); - hashProofOfStake = 0; targetProofOfStake = 0; + if (IsInitialBlockDownload()) + { + // During IBD the UTXO set isn't fully loaded; CheckProofOfStake() + // would fail reading txPrev. Skip with a throttled log. + if (nHeight % 10000 == 0) + printf("SKIP: PoS kernel check skipped for block %d during IBD\n", nHeight); + hashProofOfStake = 0; targetProofOfStake = 0; + } + else + { + // Post-IBD: verify the PoS kernel signature normally. + if (!CheckProofOfStake(vtx[1], nBits, hashProofOfStake, targetProofOfStake)) + return DoS(100, error("AcceptBlock() : check proof-of-stake failed for block %d", nHeight)); + } } // Sync checkpoint enforcement is disabled: @@ -3089,7 +3132,7 @@ bool ProcessBlock(CNode* pfrom, CBlock* pblock) if (!pcheckpoint) pcheckpoint = pindexBest; - if (false && pcheckpoint && pblock->hashPrevBlock != hashBestChain) // TEMP: disabled anti-spam check for sync + if (pcheckpoint && pblock->hashPrevBlock != hashBestChain) { int64_t deltaTime = pblock->GetBlockTime() - pcheckpoint->nTime; CBigNum bnNewBlock; @@ -3121,7 +3164,8 @@ bool ProcessBlock(CNode* pfrom, CBlock* pblock) // If don't already have its previous block, shunt it off to holding area until we get it if (!mapBlockIndex.count(pblock->hashPrevBlock)) { - printf("ProcessBlock: ORPHAN BLOCK, prev=%s\n", pblock->hashPrevBlock.ToString().substr(0,20).c_str()); + if (fDebug) + printf("ProcessBlock: ORPHAN BLOCK, prev=%s\n", pblock->hashPrevBlock.ToString().substr(0,20).c_str()); std::unique_ptr pblock2 = std::make_unique(*pblock); // triangles: check proof-of-stake if (pblock2->IsProofOfStake()) @@ -3192,7 +3236,7 @@ bool ProcessBlock(CNode* pfrom, CBlock* pblock) { const unsigned int nQueued = (g_syncManager.GetBestHeader() != 0) ? g_syncManager.QueueBlocksParallel() : 0; - if (nQueued > 0) + if (nQueued > 0 && fDebug) printf("IBD-DIAG: queued %u more blocks from header planner after accepting %s\n", nQueued, hash.ToString().substr(0,20).c_str()); @@ -3202,7 +3246,7 @@ bool ProcessBlock(CNode* pfrom, CBlock* pblock) const unsigned int nRefilled = g_syncManager.RequestRefillAllPeers( g_syncManager.GetBestHeader(), CSyncManager::HEADER_SYNC_REFILL_MIN_INTERVAL_SECONDS, (nPlannerDepth == 0) ? "post-accept planner empty" : "post-accept planner low-water"); - if (nRefilled > 0) + if (nRefilled > 0 && fDebug) printf("IBD-DIAG: post-accept requested headers from %u peers at plannerDepth=%u after block %s\n", nRefilled, nPlannerDepth, hash.ToString().substr(0,20).c_str()); } diff --git a/src/net.cpp b/src/net.cpp index 685737f..307b973 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -21,6 +21,8 @@ #ifdef WIN32 #include +#else +#include #endif #ifdef USE_UPNP @@ -831,36 +833,96 @@ void SocketSendData(CNode *pnode) std::deque::iterator it = pnode->vSendMsg.begin(); while (it != pnode->vSendMsg.end()) { +#ifndef WIN32 + // Coalesce up to MAX_IOV queued messages into a single syscall using + // scatter-gather I/O. On Linux we use sendmsg() so we can pass + // MSG_NOSIGNAL | MSG_DONTWAIT; on other POSIX systems (e.g. BSD where + // SO_NOSIGPIPE is already set on the socket) we fall back to writev(). + static const int MAX_IOV = 16; + struct iovec iov[MAX_IOV]; + int iovcnt = 0; + std::deque::iterator batchEnd = it; + + for (; batchEnd != pnode->vSendMsg.end() && iovcnt < MAX_IOV; ++batchEnd, ++iovcnt) { + const CSerializeData &data = *batchEnd; + size_t off = (batchEnd == it) ? pnode->nSendOffset : 0; + assert(data.size() > off); + iov[iovcnt].iov_base = const_cast(&data[off]); + iov[iovcnt].iov_len = data.size() - off; + } + + if (iovcnt == 0) + break; + + ssize_t nBytes; +#ifdef MSG_NOSIGNAL + struct msghdr msg; + memset(&msg, 0, sizeof(msg)); + msg.msg_iov = iov; + msg.msg_iovlen = iovcnt; + nBytes = sendmsg(pnode->hSocket, &msg, MSG_NOSIGNAL | MSG_DONTWAIT); +#else + nBytes = writev(pnode->hSocket, iov, iovcnt); +#endif + if (nBytes > 0) { + pnode->nLastSend = GetTime(); + pnode->nSendBytes += nBytes; + + // Consume nBytes across the coalesced messages + while (it != batchEnd && nBytes > 0) { + const CSerializeData &data = *it; + size_t remaining = data.size() - pnode->nSendOffset; + if ((size_t)nBytes >= remaining) { + nBytes -= remaining; + pnode->nSendSize -= data.size(); + pnode->nSendOffset = 0; + ++it; + } else { + pnode->nSendOffset += nBytes; + nBytes = 0; + } + } + // Socket buffer full mid-batch — wait for next cycle + if (it != batchEnd) + break; + } else if (nBytes < 0) { + int nErr = WSAGetLastError(); + if (nErr != WSAEWOULDBLOCK && nErr != WSAEMSGSIZE && nErr != WSAEINTR && nErr != WSAEINPROGRESS) { + printf("socket send error %d\n", nErr); + pnode->CloseSocketDisconnect(); + } + break; + } else { + // nBytes == 0: peer closed + break; + } +#else + // Windows: individual send() calls const CSerializeData &data = *it; assert(data.size() > pnode->nSendOffset); int nBytes = send(pnode->hSocket, &data[pnode->nSendOffset], data.size() - pnode->nSendOffset, MSG_NOSIGNAL | MSG_DONTWAIT); if (nBytes > 0) { pnode->nLastSend = GetTime(); pnode->nSendOffset += nBytes; - - pnode->nSendBytes += nBytes; - + pnode->nSendBytes += nBytes; if (pnode->nSendOffset == data.size()) { pnode->nSendOffset = 0; pnode->nSendSize -= data.size(); it++; } else { - // could not send full message; stop sending more break; } } else { if (nBytes < 0) { - // error int nErr = WSAGetLastError(); - if (nErr != WSAEWOULDBLOCK && nErr != WSAEMSGSIZE && nErr != WSAEINTR && nErr != WSAEINPROGRESS) - { + if (nErr != WSAEWOULDBLOCK && nErr != WSAEMSGSIZE && nErr != WSAEINTR && nErr != WSAEINPROGRESS) { printf("socket send error %d\n", nErr); pnode->CloseSocketDisconnect(); } } - // couldn't send anything at all break; } +#endif } if (it == pnode->vSendMsg.end()) { @@ -1221,7 +1283,7 @@ void ThreadSocketHandler2(void* parg) if (fShutdown) return; - MilliSleep(10); + MilliSleep(IsInitialBlockDownload() ? 1 : 10); } } diff --git a/src/netbase.cpp b/src/netbase.cpp index 0873a64..a80e9d0 100644 --- a/src/netbase.cpp +++ b/src/netbase.cpp @@ -10,6 +10,7 @@ #ifndef WIN32 #include +#include #endif #include @@ -457,6 +458,19 @@ bool static ConnectSocketDirectly(const CService &addrConnect, SOCKET& hSocketRe } } + // TCP_NODELAY: disable Nagle's algorithm for low-latency P2P messaging. + // SO_KEEPALIVE: detect dead connections faster (important for Tor/I2P + // tunnels that can silently drop without RST/FIN). + { + int one = 1; +#ifdef WIN32 + setsockopt(hSocket, IPPROTO_TCP, TCP_NODELAY, (char*)&one, sizeof(one)); +#else + setsockopt(hSocket, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); +#endif + setsockopt(hSocket, SOL_SOCKET, SO_KEEPALIVE, (char*)&one, sizeof(one)); + } + // this isn't even strictly necessary // CNode::ConnectNode immediately turns the socket back to non-blocking // but we'll turn it back to blocking just in case diff --git a/src/syncmanager.cpp b/src/syncmanager.cpp index c12ea57..c78668a 100644 --- a/src/syncmanager.cpp +++ b/src/syncmanager.cpp @@ -44,6 +44,11 @@ static const int HEADER_FRONT_MAX_AHEAD = 32768; std::map mapHeaders; uint256 hashBestHeader = 0; int64_t nLastNewHeaderTime = 0; + +// O(1) in-flight counter — replaces the O(n) scan in CountInFlight(). +// Incremented when fRequested transitions false→true; decremented when an +// entry with fRequested==true is erased from mapHeaders. +static size_t g_nInFlight = 0; } CSyncManager g_syncManager; @@ -151,6 +156,8 @@ void CSyncManager::PruneHeaders() if (it->second.nHeight > nProtectFloor && nNow - it->second.nInsertTime >= HEADER_SYNC_TTL_MICROS) { + if (it->second.fRequested) + --g_nInFlight; it = mapHeaders.erase(it); ++nEvicted; } @@ -188,7 +195,11 @@ void CSyncManager::PruneHeaders() const size_t nTarget = (size_t)MAX_HEADER_SYNC_CACHE * 3 / 4; size_t i = 0; while (mapHeaders.size() > nTarget && i < vEvictable.size()) + { + if (vEvictable[i]->second.fRequested) + --g_nInFlight; mapHeaders.erase(vEvictable[i++]); + } RecomputeBestHeader(); } @@ -287,14 +298,7 @@ bool CSyncManager::PathReachesChain(const std::vector& vPath) const unsigned int CSyncManager::CountInFlight() const { - const int64_t nNow = GetTime() * 1000000; - unsigned int nInFlight = 0; - for (std::map::const_iterator it = mapHeaders.begin(); it != mapHeaders.end(); ++it) - { - if (it->second.fRequested && nNow - it->second.nLastRequestTime < HEADER_REQUEST_TIMEOUT_MICROS) - ++nInFlight; - } - return nInFlight; + return (unsigned int)g_nInFlight; } unsigned int CSyncManager::GetPlannerDepth() const @@ -331,6 +335,8 @@ void CSyncManager::BlockAccepted(const uint256& hashBlock) if (mi == mapHeaders.end()) return; + if (mi->second.fRequested) + --g_nInFlight; mapHeaders.erase(mi); if (hashBestHeader == hashBlock) RecomputeBestHeader(); @@ -602,7 +608,10 @@ unsigned int CSyncManager::QueueBlocksParallel(unsigned int nWindow) if (!mi->second.fRequested || nNow - mi->second.nLastRequestTime >= HEADER_REQUEST_TIMEOUT_MICROS) { if (!mi->second.fRequested) + { mi->second.nFirstRequestTime = nNow; + ++g_nInFlight; + } mi->second.fRequested = true; mi->second.nLastRequestTime = nNow; } diff --git a/src/txdb-base.cpp b/src/txdb-base.cpp index cbc1592..cb2dfe3 100644 --- a/src/txdb-base.cpp +++ b/src/txdb-base.cpp @@ -9,6 +9,7 @@ #include "main.h" #include "sync.h" +#include #include using namespace std; @@ -320,14 +321,42 @@ struct COutPointHasher { struct CUtxoCacheEntry { CUtxoEntry utxo; bool fPresent; // true = exists, false = known absent (negative cache) + std::list::iterator lruIt; // Position in g_utxoLruList CUtxoCacheEntry() : fPresent(false) {} CUtxoCacheEntry(const CUtxoEntry& u, bool p) : utxo(u), fPresent(p) {} }; +// Access-order list for LRU eviction. Front = most recently used, back = LRU. +std::list g_utxoLruList; std::unordered_map g_mapUtxoCache; CCriticalSection g_cs_utxoCache; const size_t UTXO_CACHE_MAX_ENTRIES = 2000000; // ~400MB at ~200 bytes each +// Promote an existing cache entry to most-recently-used. +inline void TouchUtxoEntry( + std::unordered_map::iterator it) +{ + g_utxoLruList.splice(g_utxoLruList.begin(), g_utxoLruList, it->second.lruIt); +} + +// Insert or update a cache entry, promoting to most-recently-used. +inline void PutUtxoCacheEntry(const COutPoint& outpoint, + const CUtxoEntry& utxo, bool fPresent) +{ + auto it = g_mapUtxoCache.find(outpoint); + if (it != g_mapUtxoCache.end()) { + it->second.utxo = utxo; + it->second.fPresent = fPresent; + g_utxoLruList.splice(g_utxoLruList.begin(), g_utxoLruList, it->second.lruIt); + } else { + g_utxoLruList.push_front(outpoint); + CUtxoCacheEntry& e = g_mapUtxoCache[outpoint]; + e.utxo = utxo; + e.fPresent = fPresent; + e.lruIt = g_utxoLruList.begin(); + } +} + } // anonymous namespace bool CTxDBBase::ReadUtxo(const uint256& hash, unsigned int n, CUtxoEntry& entry) @@ -340,6 +369,7 @@ bool CTxDBBase::ReadUtxo(const uint256& hash, unsigned int n, CUtxoEntry& entry) auto it = g_mapUtxoCache.find(outpoint); if (it != g_mapUtxoCache.end()) { + TouchUtxoEntry(it); if (it->second.fPresent) { entry = it->second.utxo; return true; @@ -353,12 +383,7 @@ bool CTxDBBase::ReadUtxo(const uint256& hash, unsigned int n, CUtxoEntry& entry) { LOCK(g_cs_utxoCache); if (g_mapUtxoCache.size() < UTXO_CACHE_MAX_ENTRIES) - { - if (fFound) - g_mapUtxoCache[outpoint] = CUtxoCacheEntry(entry, true); - else - g_mapUtxoCache[outpoint] = CUtxoCacheEntry(CUtxoEntry(), false); - } + PutUtxoCacheEntry(outpoint, entry, fFound); } return fFound; @@ -370,16 +395,17 @@ bool CTxDBBase::WriteUtxo(const uint256& hash, unsigned int n, const CUtxoEntry& { LOCK(g_cs_utxoCache); - g_mapUtxoCache[outpoint] = CUtxoCacheEntry(entry, true); + PutUtxoCacheEntry(outpoint, entry, true); - // Periodic eviction: clear half when over the limit. Simple but - // effective — the cache repopulates with the hot working set. + // LRU eviction: evict least-recently-used entries when over the limit. if (g_mapUtxoCache.size() > UTXO_CACHE_MAX_ENTRIES) { size_t nTarget = UTXO_CACHE_MAX_ENTRIES / 2; - auto it = g_mapUtxoCache.begin(); - while (g_mapUtxoCache.size() > nTarget && it != g_mapUtxoCache.end()) - it = g_mapUtxoCache.erase(it); + while (g_mapUtxoCache.size() > nTarget) + { + g_mapUtxoCache.erase(g_utxoLruList.back()); + g_utxoLruList.pop_back(); + } } } @@ -392,7 +418,7 @@ bool CTxDBBase::EraseUtxo(const uint256& hash, unsigned int n) { LOCK(g_cs_utxoCache); - g_mapUtxoCache[outpoint] = CUtxoCacheEntry(CUtxoEntry(), false); + PutUtxoCacheEntry(outpoint, CUtxoEntry(), false); } return Erase(make_pair(string("u"), make_pair(hash, n))); @@ -405,8 +431,10 @@ bool CTxDBBase::HaveUtxo(const uint256& hash, unsigned int n) { LOCK(g_cs_utxoCache); auto it = g_mapUtxoCache.find(outpoint); - if (it != g_mapUtxoCache.end()) + if (it != g_mapUtxoCache.end()) { + TouchUtxoEntry(it); return it->second.fPresent; + } } if (Exists(make_pair(string("u"), make_pair(hash, n)))) diff --git a/src/txdb-rocksdb.cpp b/src/txdb-rocksdb.cpp index 9cfcb03..986ad73 100644 --- a/src/txdb-rocksdb.cpp +++ b/src/txdb-rocksdb.cpp @@ -32,6 +32,15 @@ namespace fs = std::filesystem; // the same way the LevelDB backend shares its txdb singleton. static rocksdb::DB* g_rocksdb = nullptr; +// Non-batched writes bypass WAL fsync. The TxnCommit path handles durability; +// crash recovery replays from block files anyway. Default WriteOptions may +// vary across RocksDB versions, so we pin sync=false explicitly. +static const rocksdb::WriteOptions g_fastWriteOpts = []{ + rocksdb::WriteOptions wo; + wo.sync = false; + return wo; +}(); + namespace { // rocksdb::DB::Open shipped a raw DB** overload for years; newer releases @@ -71,8 +80,9 @@ static rocksdb::Options GetRocksOptions() rocksdb::Options opts; opts.create_if_missing = false; opts.compression = rocksdb::kSnappyCompression; - opts.max_open_files = 1000; - opts.write_buffer_size = 64 * 1048576; + opts.max_open_files = -1; + opts.write_buffer_size = 256 * 1048576; + opts.max_write_buffer_number = 4; opts.IncreaseParallelism(); // Multi-threaded compaction. opts.OptimizeLevelStyleCompaction(); // Sensible defaults for a LSM workload. @@ -263,7 +273,7 @@ bool CRocksTxDB::WriteRaw(const std::string& key, const std::string& value) pendingBatch[key] = value; return true; } - rocksdb::Status status = pdb->Put(rocksdb::WriteOptions(), key, value); + rocksdb::Status status = pdb->Put(g_fastWriteOpts, key, value); if (!status.ok()) { printf("RocksDB write failure: %s\n", status.ToString().c_str()); return false; diff --git a/src/txdb-rocksdb.h b/src/txdb-rocksdb.h index eec41da..8e2a303 100644 --- a/src/txdb-rocksdb.h +++ b/src/txdb-rocksdb.h @@ -10,6 +10,7 @@ #include #include #include +#include #include #include @@ -76,7 +77,7 @@ private: // active batch?" without iterating the WriteBatch via Handler — Ubuntu's // librocksdb-dev hides typeinfo for rocksdb::WriteBatch::Handler so a // subclass-based scan fails to link there. - std::map> pendingBatch; + std::unordered_map> pendingBatch; bool ScanBatch(const std::string& key, std::string* value, bool* deleted) const; };