Merge branch 'dev' into sync-rippled

This commit is contained in:
RichardAH
2024-11-12 08:57:53 +10:00
committed by GitHub
24 changed files with 2569 additions and 30 deletions

View File

@@ -538,7 +538,9 @@ target_sources (rippled PRIVATE
subdir: nodestore subdir: nodestore
#]===============================] #]===============================]
src/ripple/nodestore/backend/CassandraFactory.cpp src/ripple/nodestore/backend/CassandraFactory.cpp
src/ripple/nodestore/backend/RWDBFactory.cpp
src/ripple/nodestore/backend/MemoryFactory.cpp src/ripple/nodestore/backend/MemoryFactory.cpp
src/ripple/nodestore/backend/FlatmapFactory.cpp
src/ripple/nodestore/backend/NuDBFactory.cpp src/ripple/nodestore/backend/NuDBFactory.cpp
src/ripple/nodestore/backend/NullFactory.cpp src/ripple/nodestore/backend/NullFactory.cpp
src/ripple/nodestore/backend/RocksDBFactory.cpp src/ripple/nodestore/backend/RocksDBFactory.cpp

View File

@@ -1056,7 +1056,18 @@
# Cassandra is an alternative backend to be used only with Reporting Mode. # Cassandra is an alternative backend to be used only with Reporting Mode.
# See the Reporting Mode section for more details about Reporting Mode. # See the Reporting Mode section for more details about Reporting Mode.
# #
# Required keys for NuDB and RocksDB: # type = RWDB
#
# RWDB is a high-performance memory store written by XRPL-Labs and optimized
# for xahaud. RWDB is NOT persistent and the data will be lost on restart.
# RWDB is recommended for Validator and Peer nodes that are not required to
# store history.
#
# RWDB maintains its high speed regardless of the amount of history
# stored. Online delete should NOT be used instead RWDB will use the
# ledger_history config value to determine how many ledgers to keep in memory.
#
# Required keys for NuDB, RWDB and RocksDB:
# #
# path Location to store the database # path Location to store the database
# #
@@ -1112,7 +1123,8 @@
# online_delete Minimum value of 256. Enable automatic purging # online_delete Minimum value of 256. Enable automatic purging
# of older ledger information. Maintain at least this # of older ledger information. Maintain at least this
# number of ledger records online. Must be greater # number of ledger records online. Must be greater
# than or equal to ledger_history. # than or equal to ledger_history. If using RWDB
# this value is ignored.
# #
# These keys modify the behavior of online_delete, and thus are only # These keys modify the behavior of online_delete, and thus are only
# relevant if online_delete is defined and non-zero: # relevant if online_delete is defined and non-zero:

29
hook/generate_sfcodes.sh Executable file
View File

@@ -0,0 +1,29 @@
#/bin/bash
RIPPLED_ROOT="../src/ripple"
echo '// For documentation please see: https://xrpl-hooks.readme.io/reference/'
echo '// Generated using generate_sfcodes.sh'
cat $RIPPLED_ROOT/protocol/impl/SField.cpp | grep -E '^CONSTRUCT_' |
sed 's/UINT16/1/g' |
sed 's/UINT32/2/g' |
sed 's/UINT64/3/g' |
sed 's/HASH128/4/g' |
sed 's/HASH256/5/g' |
sed 's/UINT128/4/g' |
sed 's/UINT256/5/g' |
sed 's/AMOUNT/6/g' |
sed 's/VL/7/g' | sed 's/Import7/ImportVL/g' |
sed 's/ACCOUNT/8/g' |
sed 's/OBJECT/14/g' |
sed 's/ARRAY/15/g' |
sed 's/UINT8/16/g' |
sed 's/HASH160/17/g' |
sed 's/UINT160/17/g' |
sed 's/PATHSET/18/g' |
sed 's/VECTOR256/19/g' |
sed 's/UINT96/20/g' |
sed 's/UINT192/21/g' |
sed 's/UINT384/22/g' |
sed 's/UINT512/23/g' |
grep -Eo '"([^"]+)", *([0-9]+), *([0-9]+)' |
sed 's/"//g' | sed 's/ *//g' | sed 's/,/ /g' |
awk '{print ("#define sf"$1" (("$2"U << 16U) + "$3"U)")}'

View File

@@ -60,7 +60,10 @@
#define sfBurnedNFTokens ((2U << 16U) + 44U) #define sfBurnedNFTokens ((2U << 16U) + 44U)
#define sfHookStateCount ((2U << 16U) + 45U) #define sfHookStateCount ((2U << 16U) + 45U)
#define sfEmitGeneration ((2U << 16U) + 46U) #define sfEmitGeneration ((2U << 16U) + 46U)
#define sfLockCount ((2U << 16U) + 47U) #define sfLockCount ((2U << 16U) + 49U)
#define sfFirstNFTokenSequence ((2U << 16U) + 50U)
#define sfXahauActivationLgrSeq ((2U << 16U) + 96U)
#define sfImportSequence ((2U << 16U) + 97U)
#define sfRewardTime ((2U << 16U) + 98U) #define sfRewardTime ((2U << 16U) + 98U)
#define sfRewardLgrFirst ((2U << 16U) + 99U) #define sfRewardLgrFirst ((2U << 16U) + 99U)
#define sfRewardLgrLast ((2U << 16U) + 100U) #define sfRewardLgrLast ((2U << 16U) + 100U)
@@ -80,6 +83,8 @@
#define sfHookInstructionCount ((3U << 16U) + 17U) #define sfHookInstructionCount ((3U << 16U) + 17U)
#define sfHookReturnCode ((3U << 16U) + 18U) #define sfHookReturnCode ((3U << 16U) + 18U)
#define sfReferenceCount ((3U << 16U) + 19U) #define sfReferenceCount ((3U << 16U) + 19U)
#define sfAccountIndex ((3U << 16U) + 98U)
#define sfAccountCount ((3U << 16U) + 99U)
#define sfRewardAccumulator ((3U << 16U) + 100U) #define sfRewardAccumulator ((3U << 16U) + 100U)
#define sfEmailHash ((4U << 16U) + 1U) #define sfEmailHash ((4U << 16U) + 1U)
#define sfTakerPaysCurrency ((10U << 16U) + 1U) #define sfTakerPaysCurrency ((10U << 16U) + 1U)
@@ -120,6 +125,9 @@
#define sfOfferID ((5U << 16U) + 34U) #define sfOfferID ((5U << 16U) + 34U)
#define sfEscrowID ((5U << 16U) + 35U) #define sfEscrowID ((5U << 16U) + 35U)
#define sfURITokenID ((5U << 16U) + 36U) #define sfURITokenID ((5U << 16U) + 36U)
#define sfGovernanceFlags ((5U << 16U) + 99U)
#define sfGovernanceMarks ((5U << 16U) + 98U)
#define sfEmittedTxnID ((5U << 16U) + 97U)
#define sfAmount ((6U << 16U) + 1U) #define sfAmount ((6U << 16U) + 1U)
#define sfBalance ((6U << 16U) + 2U) #define sfBalance ((6U << 16U) + 2U)
#define sfLimitAmount ((6U << 16U) + 3U) #define sfLimitAmount ((6U << 16U) + 3U)
@@ -136,6 +144,9 @@
#define sfNFTokenBrokerFee ((6U << 16U) + 19U) #define sfNFTokenBrokerFee ((6U << 16U) + 19U)
#define sfHookCallbackFee ((6U << 16U) + 20U) #define sfHookCallbackFee ((6U << 16U) + 20U)
#define sfLockedBalance ((6U << 16U) + 21U) #define sfLockedBalance ((6U << 16U) + 21U)
#define sfBaseFeeDrops ((6U << 16U) + 22U)
#define sfReserveBaseDrops ((6U << 16U) + 23U)
#define sfReserveIncrementDrops ((6U << 16U) + 24U)
#define sfPublicKey ((7U << 16U) + 1U) #define sfPublicKey ((7U << 16U) + 1U)
#define sfMessageKey ((7U << 16U) + 2U) #define sfMessageKey ((7U << 16U) + 2U)
#define sfSigningPubKey ((7U << 16U) + 3U) #define sfSigningPubKey ((7U << 16U) + 3U)
@@ -171,11 +182,13 @@
#define sfNFTokenMinter ((8U << 16U) + 9U) #define sfNFTokenMinter ((8U << 16U) + 9U)
#define sfEmitCallback ((8U << 16U) + 10U) #define sfEmitCallback ((8U << 16U) + 10U)
#define sfHookAccount ((8U << 16U) + 16U) #define sfHookAccount ((8U << 16U) + 16U)
#define sfInform ((8U << 16U) + 99U)
#define sfIndexes ((19U << 16U) + 1U) #define sfIndexes ((19U << 16U) + 1U)
#define sfHashes ((19U << 16U) + 2U) #define sfHashes ((19U << 16U) + 2U)
#define sfAmendments ((19U << 16U) + 3U) #define sfAmendments ((19U << 16U) + 3U)
#define sfNFTokenOffers ((19U << 16U) + 4U) #define sfNFTokenOffers ((19U << 16U) + 4U)
#define sfHookNamespaces ((19U << 16U) + 5U) #define sfHookNamespaces ((19U << 16U) + 5U)
#define sfURITokenIDs ((19U << 16U) + 99U)
#define sfPaths ((18U << 16U) + 1U) #define sfPaths ((18U << 16U) + 1U)
#define sfTransactionMetaData ((14U << 16U) + 2U) #define sfTransactionMetaData ((14U << 16U) + 2U)
#define sfCreatedNode ((14U << 16U) + 3U) #define sfCreatedNode ((14U << 16U) + 3U)
@@ -198,6 +211,12 @@
#define sfHookDefinition ((14U << 16U) + 22U) #define sfHookDefinition ((14U << 16U) + 22U)
#define sfHookParameter ((14U << 16U) + 23U) #define sfHookParameter ((14U << 16U) + 23U)
#define sfHookGrant ((14U << 16U) + 24U) #define sfHookGrant ((14U << 16U) + 24U)
#define sfGenesisMint ((14U << 16U) + 96U)
#define sfActiveValidator ((14U << 16U) + 95U)
#define sfImportVLKey ((14U << 16U) + 94U)
#define sfHookEmission ((14U << 16U) + 93U)
#define sfMintURIToken ((14U << 16U) + 92U)
#define sfAmountEntry ((14U << 16U) + 91U)
#define sfSigners ((15U << 16U) + 3U) #define sfSigners ((15U << 16U) + 3U)
#define sfSignerEntries ((15U << 16U) + 4U) #define sfSignerEntries ((15U << 16U) + 4U)
#define sfTemplate ((15U << 16U) + 5U) #define sfTemplate ((15U << 16U) + 5U)
@@ -212,4 +231,8 @@
#define sfHookExecutions ((15U << 16U) + 18U) #define sfHookExecutions ((15U << 16U) + 18U)
#define sfHookParameters ((15U << 16U) + 19U) #define sfHookParameters ((15U << 16U) + 19U)
#define sfHookGrants ((15U << 16U) + 20U) #define sfHookGrants ((15U << 16U) + 20U)
#define sfGenesisMints ((15U << 16U) + 96U)
#define sfActiveValidators ((15U << 16U) + 95U) #define sfActiveValidators ((15U << 16U) + 95U)
#define sfImportVLKeys ((15U << 16U) + 94U)
#define sfHookEmissions ((15U << 16U) + 93U)
#define sfAmounts ((15U << 16U) + 92U)

View File

@@ -4617,6 +4617,8 @@ DEFINE_HOOK_FUNCTION(
} }
catch (std::exception& e) catch (std::exception& e)
{ {
JLOG(j.trace()) << "HookInfo[" << HC_ACC()
<< "]: etxn_fee_base exception: " << e.what();
return INVALID_TXN; return INVALID_TXN;
} }
@@ -5402,7 +5404,7 @@ DEFINE_HOOK_FUNCTION(
const int64_t float_one_internal = make_float(1000000000000000ull, -15, false); const int64_t float_one_internal = make_float(1000000000000000ull, -15, false);
inline int64_t inline int64_t
float_divide_internal(int64_t float1, int64_t float2) float_divide_internal(int64_t float1, int64_t float2, bool hasFix)
{ {
RETURN_IF_INVALID_FLOAT(float1); RETURN_IF_INVALID_FLOAT(float1);
RETURN_IF_INVALID_FLOAT(float2); RETURN_IF_INVALID_FLOAT(float2);
@@ -5455,8 +5457,16 @@ float_divide_internal(int64_t float1, int64_t float2)
while (man2 > 0) while (man2 > 0)
{ {
int i = 0; int i = 0;
for (; man1 > man2; man1 -= man2, ++i) if (hasFix)
; {
for (; man1 >= man2; man1 -= man2, ++i)
;
}
else
{
for (; man1 > man2; man1 -= man2, ++i)
;
}
man3 *= 10; man3 *= 10;
man3 += i; man3 += i;
@@ -5476,7 +5486,8 @@ DEFINE_HOOK_FUNCTION(int64_t, float_divide, int64_t float1, int64_t float2)
HOOK_SETUP(); // populates memory_ctx, memory, memory_length, applyCtx, HOOK_SETUP(); // populates memory_ctx, memory, memory_length, applyCtx,
// hookCtx on current stack // hookCtx on current stack
return float_divide_internal(float1, float2); bool const hasFix = view.rules().enabled(fixFloatDivide);
return float_divide_internal(float1, float2, hasFix);
HOOK_TEARDOWN(); HOOK_TEARDOWN();
} }
@@ -5495,7 +5506,9 @@ DEFINE_HOOK_FUNCTION(int64_t, float_invert, int64_t float1)
return DIVISION_BY_ZERO; return DIVISION_BY_ZERO;
if (float1 == float_one_internal) if (float1 == float_one_internal)
return float_one_internal; return float_one_internal;
return float_divide_internal(float_one_internal, float1);
bool const fixV3 = view.rules().enabled(fixFloatDivide);
return float_divide_internal(float_one_internal, float1, fixV3);
HOOK_TEARDOWN(); HOOK_TEARDOWN();
} }

View File

@@ -219,7 +219,7 @@ private:
run() run()
{ {
beast::setCurrentThreadName("LedgerCleaner"); beast::setCurrentThreadName("LedgerCleaner");
JLOG(j_.debug()) << "Started"; JLOG(j_.debug()) << "Started ledger cleaner";
while (true) while (true)
{ {
@@ -392,7 +392,8 @@ private:
if (app_.getFeeTrack().isLoadedLocal()) if (app_.getFeeTrack().isLoadedLocal())
{ {
JLOG(j_.debug()) << "Waiting for load to subside"; JLOG(j_.debug())
<< "Ledger Cleaner: Waiting for load to subside";
std::this_thread::sleep_for(std::chrono::seconds(5)); std::this_thread::sleep_for(std::chrono::seconds(5));
continue; continue;
} }
@@ -415,13 +416,15 @@ private:
bool fail = false; bool fail = false;
if (ledgerHash.isZero()) if (ledgerHash.isZero())
{ {
JLOG(j_.info()) JLOG(j_.warn())
<< "Unable to get hash for ledger " << ledgerIndex; << "Ledger Cleaner: Unable to get hash for ledger "
<< ledgerIndex;
fail = true; fail = true;
} }
else if (!doLedger(ledgerIndex, ledgerHash, doNodes, doTxns)) else if (!doLedger(ledgerIndex, ledgerHash, doNodes, doTxns))
{ {
JLOG(j_.info()) << "Failed to process ledger " << ledgerIndex; JLOG(j_.warn()) << "Ledger Cleaner: Failed to process ledger "
<< ledgerIndex;
fail = true; fail = true;
} }

View File

@@ -19,7 +19,6 @@
#ifndef RIPPLE_APP_MAIN_APPLICATION_H_INCLUDED #ifndef RIPPLE_APP_MAIN_APPLICATION_H_INCLUDED
#define RIPPLE_APP_MAIN_APPLICATION_H_INCLUDED #define RIPPLE_APP_MAIN_APPLICATION_H_INCLUDED
#include <ripple/basics/TaggedCache.h> #include <ripple/basics/TaggedCache.h>
#include <ripple/beast/utility/PropertyStream.h> #include <ripple/beast/utility/PropertyStream.h>
#include <ripple/core/Config.h> #include <ripple/core/Config.h>

View File

@@ -118,7 +118,9 @@ SHAMapStoreImp::SHAMapStoreImp(
get_if_exists(section, "online_delete", deleteInterval_); get_if_exists(section, "online_delete", deleteInterval_);
if (deleteInterval_) bool const isMem = config.mem_backend();
if (deleteInterval_ || isMem)
{ {
if (app_.config().reporting()) if (app_.config().reporting())
{ {
@@ -127,6 +129,9 @@ SHAMapStoreImp::SHAMapStoreImp(
"online_delete info from config"); "online_delete info from config");
} }
if (isMem)
deleteInterval_ = config.LEDGER_HISTORY;
// Configuration that affects the behavior of online delete // Configuration that affects the behavior of online delete
get_if_exists(section, "delete_batch", deleteBatch_); get_if_exists(section, "delete_batch", deleteBatch_);
std::uint32_t temp; std::uint32_t temp;
@@ -162,7 +167,8 @@ SHAMapStoreImp::SHAMapStoreImp(
} }
state_db_.init(config, dbName_); state_db_.init(config, dbName_);
dbPaths(); if (!isMem)
dbPaths();
} }
} }
@@ -195,6 +201,7 @@ SHAMapStoreImp::makeNodeStore(int readThreads)
"online_delete info from config"); "online_delete info from config");
} }
SavedState state = state_db_.getState(); SavedState state = state_db_.getState();
auto writableBackend = makeBackendRotating(state.writableDb); auto writableBackend = makeBackendRotating(state.writableDb);
auto archiveBackend = makeBackendRotating(state.archiveDb); auto archiveBackend = makeBackendRotating(state.archiveDb);
if (!state.writableDb.size()) if (!state.writableDb.size())
@@ -293,6 +300,8 @@ SHAMapStoreImp::run()
fullBelowCache_ = &(*app_.getNodeFamily().getFullBelowCache(0)); fullBelowCache_ = &(*app_.getNodeFamily().getFullBelowCache(0));
treeNodeCache_ = &(*app_.getNodeFamily().getTreeNodeCache(0)); treeNodeCache_ = &(*app_.getNodeFamily().getTreeNodeCache(0));
bool const isMem = app_.config().mem_backend();
if (advisoryDelete_) if (advisoryDelete_)
canDelete_ = state_db_.getCanDelete(); canDelete_ = state_db_.getCanDelete();
@@ -351,7 +360,7 @@ SHAMapStoreImp::run()
// will delete up to (not including) lastRotated // will delete up to (not including) lastRotated
if (readyToRotate && !waitForImport) if (readyToRotate && !waitForImport)
{ {
JLOG(journal_.warn()) JLOG(journal_.debug())
<< "rotating validatedSeq " << validatedSeq << " lastRotated " << "rotating validatedSeq " << validatedSeq << " lastRotated "
<< lastRotated << " deleteInterval " << deleteInterval_ << lastRotated << " deleteInterval " << deleteInterval_
<< " canDelete_ " << canDelete_ << " state " << " canDelete_ " << canDelete_ << " state "
@@ -395,7 +404,7 @@ SHAMapStoreImp::run()
// Only log if we completed without a "health" abort // Only log if we completed without a "health" abort
JLOG(journal_.debug()) << validatedSeq << " freshened caches"; JLOG(journal_.debug()) << validatedSeq << " freshened caches";
JLOG(journal_.trace()) << "Making a new backend"; JLOG(journal_.debug()) << "Making a new backend";
auto newBackend = makeBackendRotating(); auto newBackend = makeBackendRotating();
JLOG(journal_.debug()) JLOG(journal_.debug())
<< validatedSeq << " new backend " << newBackend->getName(); << validatedSeq << " new backend " << newBackend->getName();

View File

@@ -0,0 +1,853 @@
#ifndef RIPPLE_APP_RDB_BACKEND_FLATMAPDATABASE_H_INCLUDED
#define RIPPLE_APP_RDB_BACKEND_FLATMAPDATABASE_H_INCLUDED
#include <ripple/app/ledger/AcceptedLedger.h>
#include <ripple/app/ledger/LedgerMaster.h>
#include <ripple/app/ledger/TransactionMaster.h>
#include <ripple/app/rdb/backend/SQLiteDatabase.h>
#include <algorithm>
#include <map>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <vector>
#include <boost/unordered/concurrent_flat_map.hpp>
namespace ripple {
struct base_uint_hasher
{
using result_type = std::size_t;
result_type
operator()(base_uint<256> const& value) const
{
return hardened_hash<>{}(value);
}
result_type
operator()(AccountID const& value) const
{
return hardened_hash<>{}(value);
}
};
class FlatmapDatabase : public SQLiteDatabase
{
private:
struct LedgerData
{
LedgerInfo info;
boost::unordered::
concurrent_flat_map<uint256, AccountTx, base_uint_hasher>
transactions;
};
struct AccountTxData
{
boost::unordered::
concurrent_flat_map<std::pair<uint32_t, uint32_t>, AccountTx>
transactions;
};
Application& app_;
Config const& config_;
JobQueue& jobQueue_;
boost::unordered::concurrent_flat_map<LedgerIndex, LedgerData> ledgers_;
boost::unordered::
concurrent_flat_map<uint256, LedgerIndex, base_uint_hasher>
ledgerHashToSeq_;
boost::unordered::concurrent_flat_map<uint256, AccountTx, base_uint_hasher>
transactionMap_;
boost::unordered::
concurrent_flat_map<AccountID, AccountTxData, base_uint_hasher>
accountTxMap_;
public:
FlatmapDatabase(Application& app, Config const& config, JobQueue& jobQueue)
: app_(app), config_(config), jobQueue_(jobQueue)
{
}
std::optional<LedgerIndex>
getMinLedgerSeq() override
{
std::optional<LedgerIndex> minSeq;
ledgers_.visit_all([&minSeq](auto const& pair) {
if (!minSeq || pair.first < *minSeq)
{
minSeq = pair.first;
}
});
return minSeq;
}
std::optional<LedgerIndex>
getTransactionsMinLedgerSeq() override
{
std::optional<LedgerIndex> minSeq;
transactionMap_.visit_all([&minSeq](auto const& pair) {
LedgerIndex seq = pair.second.second->getLgrSeq();
if (!minSeq || seq < *minSeq)
{
minSeq = seq;
}
});
return minSeq;
}
std::optional<LedgerIndex>
getAccountTransactionsMinLedgerSeq() override
{
std::optional<LedgerIndex> minSeq;
accountTxMap_.visit_all([&minSeq](auto const& pair) {
pair.second.transactions.visit_all([&minSeq](auto const& tx) {
if (!minSeq || tx.first.first < *minSeq)
{
minSeq = tx.first.first;
}
});
});
return minSeq;
}
std::optional<LedgerIndex>
getMaxLedgerSeq() override
{
std::optional<LedgerIndex> maxSeq;
ledgers_.visit_all([&maxSeq](auto const& pair) {
if (!maxSeq || pair.first > *maxSeq)
{
maxSeq = pair.first;
}
});
return maxSeq;
}
void
deleteTransactionByLedgerSeq(LedgerIndex ledgerSeq) override
{
ledgers_.visit(ledgerSeq, [this](auto& item) {
item.second.transactions.visit_all([this](auto const& txPair) {
transactionMap_.erase(txPair.first);
});
item.second.transactions.clear();
});
accountTxMap_.visit_all([ledgerSeq](auto& item) {
item.second.transactions.erase_if([ledgerSeq](auto const& tx) {
return tx.first.first == ledgerSeq;
});
});
}
void
deleteBeforeLedgerSeq(LedgerIndex ledgerSeq) override
{
ledgers_.erase_if([this, ledgerSeq](auto const& item) {
if (item.first < ledgerSeq)
{
item.second.transactions.visit_all([this](auto const& txPair) {
transactionMap_.erase(txPair.first);
});
ledgerHashToSeq_.erase(item.second.info.hash);
return true;
}
return false;
});
accountTxMap_.visit_all([ledgerSeq](auto& item) {
item.second.transactions.erase_if([ledgerSeq](auto const& tx) {
return tx.first.first < ledgerSeq;
});
});
}
void
deleteTransactionsBeforeLedgerSeq(LedgerIndex ledgerSeq) override
{
ledgers_.visit_all([this, ledgerSeq](auto& item) {
if (item.first < ledgerSeq)
{
item.second.transactions.visit_all([this](auto const& txPair) {
transactionMap_.erase(txPair.first);
});
item.second.transactions.clear();
}
});
accountTxMap_.visit_all([ledgerSeq](auto& item) {
item.second.transactions.erase_if([ledgerSeq](auto const& tx) {
return tx.first.first < ledgerSeq;
});
});
}
void
deleteAccountTransactionsBeforeLedgerSeq(LedgerIndex ledgerSeq) override
{
accountTxMap_.visit_all([ledgerSeq](auto& item) {
item.second.transactions.erase_if([ledgerSeq](auto const& tx) {
return tx.first.first < ledgerSeq;
});
});
}
std::size_t
getTransactionCount() override
{
return transactionMap_.size();
}
std::size_t
getAccountTransactionCount() override
{
std::size_t count = 0;
accountTxMap_.visit_all([&count](auto const& item) {
count += item.second.transactions.size();
});
return count;
}
CountMinMax
getLedgerCountMinMax() override
{
CountMinMax result{0, 0, 0};
ledgers_.visit_all([&result](auto const& item) {
result.numberOfRows++;
if (result.minLedgerSequence == 0 ||
item.first < result.minLedgerSequence)
{
result.minLedgerSequence = item.first;
}
if (item.first > result.maxLedgerSequence)
{
result.maxLedgerSequence = item.first;
}
});
return result;
}
bool
saveValidatedLedger(
std::shared_ptr<Ledger const> const& ledger,
bool current) override
{
try
{
LedgerData ledgerData;
ledgerData.info = ledger->info();
auto aLedger = std::make_shared<AcceptedLedger>(ledger, app_);
for (auto const& acceptedLedgerTx : *aLedger)
{
auto const& txn = acceptedLedgerTx->getTxn();
auto const& meta = acceptedLedgerTx->getMeta();
auto const& id = txn->getTransactionID();
std::string reason;
auto accTx = std::make_pair(
std::make_shared<ripple::Transaction>(txn, reason, app_),
std::make_shared<ripple::TxMeta>(meta));
ledgerData.transactions.emplace(id, accTx);
transactionMap_.emplace(id, accTx);
for (auto const& account : meta.getAffectedAccounts())
{
accountTxMap_.visit(account, [&](auto& data) {
data.second.transactions.emplace(
std::make_pair(
ledger->info().seq,
acceptedLedgerTx->getTxnSeq()),
accTx);
});
}
}
ledgers_.emplace(ledger->info().seq, std::move(ledgerData));
ledgerHashToSeq_.emplace(ledger->info().hash, ledger->info().seq);
if (current)
{
auto const cutoffSeq =
ledger->info().seq > app_.config().LEDGER_HISTORY
? ledger->info().seq - app_.config().LEDGER_HISTORY
: 0;
if (cutoffSeq > 0)
{
const std::size_t BATCH_SIZE = 128;
std::size_t deleted = 0;
ledgers_.erase_if([&](auto const& item) {
if (deleted >= BATCH_SIZE)
return false;
if (item.first < cutoffSeq)
{
item.second.transactions.visit_all(
[this](auto const& txPair) {
transactionMap_.erase(txPair.first);
});
ledgerHashToSeq_.erase(item.second.info.hash);
deleted++;
return true;
}
return false;
});
if (deleted > 0)
{
accountTxMap_.visit_all([cutoffSeq](auto& item) {
item.second.transactions.erase_if(
[cutoffSeq](auto const& tx) {
return tx.first.first < cutoffSeq;
});
});
}
app_.getLedgerMaster().clearPriorLedgers(cutoffSeq);
}
}
return true;
}
catch (std::exception const&)
{
deleteTransactionByLedgerSeq(ledger->info().seq);
return false;
}
}
std::optional<LedgerInfo>
getLedgerInfoByIndex(LedgerIndex ledgerSeq) override
{
std::optional<LedgerInfo> result;
ledgers_.visit(ledgerSeq, [&result](auto const& item) {
result = item.second.info;
});
return result;
}
std::optional<LedgerInfo>
getNewestLedgerInfo() override
{
std::optional<LedgerInfo> result;
ledgers_.visit_all([&result](auto const& item) {
if (!result || item.second.info.seq > result->seq)
{
result = item.second.info;
}
});
return result;
}
std::optional<LedgerInfo>
getLimitedOldestLedgerInfo(LedgerIndex ledgerFirstIndex) override
{
std::optional<LedgerInfo> result;
ledgers_.visit_all([&](auto const& item) {
if (item.first >= ledgerFirstIndex &&
(!result || item.first < result->seq))
{
result = item.second.info;
}
});
return result;
}
std::optional<LedgerInfo>
getLimitedNewestLedgerInfo(LedgerIndex ledgerFirstIndex) override
{
std::optional<LedgerInfo> result;
ledgers_.visit_all([&](auto const& item) {
if (item.first >= ledgerFirstIndex &&
(!result || item.first > result->seq))
{
result = item.second.info;
}
});
return result;
}
std::optional<LedgerInfo>
getLedgerInfoByHash(uint256 const& ledgerHash) override
{
std::optional<LedgerInfo> result;
ledgerHashToSeq_.visit(ledgerHash, [this, &result](auto const& item) {
ledgers_.visit(item.second, [&result](auto const& item) {
result = item.second.info;
});
});
return result;
}
uint256
getHashByIndex(LedgerIndex ledgerIndex) override
{
uint256 result;
ledgers_.visit(ledgerIndex, [&result](auto const& item) {
result = item.second.info.hash;
});
return result;
}
std::optional<LedgerHashPair>
getHashesByIndex(LedgerIndex ledgerIndex) override
{
std::optional<LedgerHashPair> result;
ledgers_.visit(ledgerIndex, [&result](auto const& item) {
result = LedgerHashPair{
item.second.info.hash, item.second.info.parentHash};
});
return result;
}
std::map<LedgerIndex, LedgerHashPair>
getHashesByIndex(LedgerIndex minSeq, LedgerIndex maxSeq) override
{
std::map<LedgerIndex, LedgerHashPair> result;
ledgers_.visit_all([&](auto const& item) {
if (item.first >= minSeq && item.first <= maxSeq)
{
result[item.first] = LedgerHashPair{
item.second.info.hash, item.second.info.parentHash};
}
});
return result;
}
std::variant<AccountTx, TxSearched>
getTransaction(
uint256 const& id,
std::optional<ClosedInterval<std::uint32_t>> const& range,
error_code_i& ec) override
{
std::variant<AccountTx, TxSearched> result = TxSearched::unknown;
transactionMap_.visit(id, [&](auto const& item) {
auto const& tx = item.second;
if (!range ||
(range->lower() <= tx.second->getLgrSeq() &&
tx.second->getLgrSeq() <= range->upper()))
{
result = tx;
}
else
{
result = TxSearched::all;
}
});
return result;
}
bool
ledgerDbHasSpace(Config const& config) override
{
return true; // In-memory database always has space
}
bool
transactionDbHasSpace(Config const& config) override
{
return true; // In-memory database always has space
}
std::uint32_t
getKBUsedAll() override
{
std::uint32_t size = sizeof(*this);
size += ledgers_.size() * (sizeof(LedgerIndex) + sizeof(LedgerData));
size +=
ledgerHashToSeq_.size() * (sizeof(uint256) + sizeof(LedgerIndex));
size += transactionMap_.size() * (sizeof(uint256) + sizeof(AccountTx));
accountTxMap_.visit_all([&size](auto const& item) {
size += sizeof(AccountID) + sizeof(AccountTxData);
size += item.second.transactions.size() * sizeof(AccountTx);
});
return size / 1024; // Convert to KB
}
std::uint32_t
getKBUsedLedger() override
{
std::uint32_t size =
ledgers_.size() * (sizeof(LedgerIndex) + sizeof(LedgerData));
size +=
ledgerHashToSeq_.size() * (sizeof(uint256) + sizeof(LedgerIndex));
return size / 1024;
}
std::uint32_t
getKBUsedTransaction() override
{
std::uint32_t size =
transactionMap_.size() * (sizeof(uint256) + sizeof(AccountTx));
accountTxMap_.visit_all([&size](auto const& item) {
size += sizeof(AccountID) + sizeof(AccountTxData);
size += item.second.transactions.size() * sizeof(AccountTx);
});
return size / 1024;
}
void
closeLedgerDB() override
{
// No-op for in-memory database
}
void
closeTransactionDB() override
{
// No-op for in-memory database
}
~FlatmapDatabase()
{
// Concurrent maps need visit_all
accountTxMap_.visit_all(
[](auto& pair) { pair.second.transactions.clear(); });
accountTxMap_.clear();
transactionMap_.clear();
ledgers_.visit_all(
[](auto& pair) { pair.second.transactions.clear(); });
ledgers_.clear();
ledgerHashToSeq_.clear();
}
std::vector<std::shared_ptr<Transaction>>
getTxHistory(LedgerIndex startIndex) override
{
std::vector<std::shared_ptr<Transaction>> result;
transactionMap_.visit_all([&](auto const& item) {
if (item.second.second->getLgrSeq() >= startIndex)
{
result.push_back(item.second.first);
}
});
std::sort(
result.begin(), result.end(), [](auto const& a, auto const& b) {
return a->getLedger() > b->getLedger();
});
if (result.size() > 20)
{
result.resize(20);
}
return result;
}
// Helper function to handle limits
template <typename Container>
void
applyLimit(Container& container, std::size_t limit, bool bUnlimited)
{
if (!bUnlimited && limit > 0 && container.size() > limit)
{
container.resize(limit);
}
}
AccountTxs
getOldestAccountTxs(AccountTxOptions const& options) override
{
AccountTxs result;
accountTxMap_.visit(options.account, [&](auto const& item) {
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
result.push_back(tx.second);
}
});
});
std::sort(
result.begin(), result.end(), [](auto const& a, auto const& b) {
return a.second->getLgrSeq() < b.second->getLgrSeq();
});
applyLimit(result, options.limit, options.bUnlimited);
return result;
}
AccountTxs
getNewestAccountTxs(AccountTxOptions const& options) override
{
AccountTxs result;
accountTxMap_.visit(options.account, [&](auto const& item) {
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
result.push_back(tx.second);
}
});
});
std::sort(
result.begin(), result.end(), [](auto const& a, auto const& b) {
return a.second->getLgrSeq() > b.second->getLgrSeq();
});
applyLimit(result, options.limit, options.bUnlimited);
return result;
}
MetaTxsList
getOldestAccountTxsB(AccountTxOptions const& options) override
{
MetaTxsList result;
accountTxMap_.visit(options.account, [&](auto const& item) {
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
result.emplace_back(
tx.second.first->getSTransaction()
->getSerializer()
.peekData(),
tx.second.second->getAsObject()
.getSerializer()
.peekData(),
tx.first.first);
}
});
});
std::sort(
result.begin(), result.end(), [](auto const& a, auto const& b) {
return std::get<2>(a) < std::get<2>(b);
});
applyLimit(result, options.limit, options.bUnlimited);
return result;
}
MetaTxsList
getNewestAccountTxsB(AccountTxOptions const& options) override
{
MetaTxsList result;
accountTxMap_.visit(options.account, [&](auto const& item) {
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
result.emplace_back(
tx.second.first->getSTransaction()
->getSerializer()
.peekData(),
tx.second.second->getAsObject()
.getSerializer()
.peekData(),
tx.first.first);
}
});
});
std::sort(
result.begin(), result.end(), [](auto const& a, auto const& b) {
return std::get<2>(a) > std::get<2>(b);
});
applyLimit(result, options.limit, options.bUnlimited);
return result;
}
std::pair<AccountTxs, std::optional<AccountTxMarker>>
oldestAccountTxPage(AccountTxPageOptions const& options) override
{
AccountTxs result;
std::optional<AccountTxMarker> marker;
accountTxMap_.visit(options.account, [&](auto const& item) {
std::vector<std::pair<std::pair<uint32_t, uint32_t>, AccountTx>>
txs;
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
txs.emplace_back(tx);
}
});
std::sort(txs.begin(), txs.end(), [](auto const& a, auto const& b) {
return a.first < b.first;
});
auto it = txs.begin();
if (options.marker)
{
it = std::find_if(txs.begin(), txs.end(), [&](auto const& tx) {
return tx.first.first == options.marker->ledgerSeq &&
tx.first.second == options.marker->txnSeq;
});
if (it != txs.end())
++it;
}
for (; it != txs.end() &&
(options.limit == 0 || result.size() < options.limit);
++it)
{
result.push_back(it->second);
}
if (it != txs.end())
{
marker = AccountTxMarker{it->first.first, it->first.second};
}
});
return {result, marker};
}
std::pair<AccountTxs, std::optional<AccountTxMarker>>
newestAccountTxPage(AccountTxPageOptions const& options) override
{
AccountTxs result;
std::optional<AccountTxMarker> marker;
accountTxMap_.visit(options.account, [&](auto const& item) {
std::vector<std::pair<std::pair<uint32_t, uint32_t>, AccountTx>>
txs;
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
txs.emplace_back(tx);
}
});
std::sort(txs.begin(), txs.end(), [](auto const& a, auto const& b) {
return a.first > b.first;
});
auto it = txs.begin();
if (options.marker)
{
it = std::find_if(txs.begin(), txs.end(), [&](auto const& tx) {
return tx.first.first == options.marker->ledgerSeq &&
tx.first.second == options.marker->txnSeq;
});
if (it != txs.end())
++it;
}
for (; it != txs.end() &&
(options.limit == 0 || result.size() < options.limit);
++it)
{
result.push_back(it->second);
}
if (it != txs.end())
{
marker = AccountTxMarker{it->first.first, it->first.second};
}
});
return {result, marker};
}
std::pair<MetaTxsList, std::optional<AccountTxMarker>>
oldestAccountTxPageB(AccountTxPageOptions const& options) override
{
MetaTxsList result;
std::optional<AccountTxMarker> marker;
accountTxMap_.visit(options.account, [&](auto const& item) {
std::vector<std::tuple<uint32_t, uint32_t, AccountTx>> txs;
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
txs.emplace_back(
tx.first.first, tx.first.second, tx.second);
}
});
std::sort(txs.begin(), txs.end());
auto it = txs.begin();
if (options.marker)
{
it = std::find_if(txs.begin(), txs.end(), [&](auto const& tx) {
return std::get<0>(tx) == options.marker->ledgerSeq &&
std::get<1>(tx) == options.marker->txnSeq;
});
if (it != txs.end())
++it;
}
for (; it != txs.end() &&
(options.limit == 0 || result.size() < options.limit);
++it)
{
const auto& [_, __, tx] = *it;
result.emplace_back(
tx.first->getSTransaction()->getSerializer().peekData(),
tx.second->getAsObject().getSerializer().peekData(),
std::get<0>(*it));
}
if (it != txs.end())
{
marker = AccountTxMarker{std::get<0>(*it), std::get<1>(*it)};
}
});
return {result, marker};
}
std::pair<MetaTxsList, std::optional<AccountTxMarker>>
newestAccountTxPageB(AccountTxPageOptions const& options) override
{
MetaTxsList result;
std::optional<AccountTxMarker> marker;
accountTxMap_.visit(options.account, [&](auto const& item) {
std::vector<std::tuple<uint32_t, uint32_t, AccountTx>> txs;
item.second.transactions.visit_all([&](auto const& tx) {
if (tx.first.first >= options.minLedger &&
tx.first.first <= options.maxLedger)
{
txs.emplace_back(
tx.first.first, tx.first.second, tx.second);
}
});
std::sort(txs.begin(), txs.end(), std::greater<>());
auto it = txs.begin();
if (options.marker)
{
it = std::find_if(txs.begin(), txs.end(), [&](auto const& tx) {
return std::get<0>(tx) == options.marker->ledgerSeq &&
std::get<1>(tx) == options.marker->txnSeq;
});
if (it != txs.end())
++it;
}
for (; it != txs.end() &&
(options.limit == 0 || result.size() < options.limit);
++it)
{
const auto& [_, __, tx] = *it;
result.emplace_back(
tx.first->getSTransaction()->getSerializer().peekData(),
tx.second->getAsObject().getSerializer().peekData(),
std::get<0>(*it));
}
if (it != txs.end())
{
marker = AccountTxMarker{std::get<0>(*it), std::get<1>(*it)};
}
});
return {result, marker};
}
};
// Factory function
std::unique_ptr<SQLiteDatabase>
getFlatmapDatabase(Application& app, Config const& config, JobQueue& jobQueue)
{
return std::make_unique<FlatmapDatabase>(app, config, jobQueue);
}
} // namespace ripple
#endif // RIPPLE_APP_RDB_BACKEND_FLATMAPDATABASE_H_INCLUDED

View File

@@ -0,0 +1,964 @@
#ifndef RIPPLE_APP_RDB_BACKEND_MEMORYDATABASE_H_INCLUDED
#define RIPPLE_APP_RDB_BACKEND_MEMORYDATABASE_H_INCLUDED
#include <ripple/app/ledger/AcceptedLedger.h>
#include <ripple/app/ledger/LedgerMaster.h>
#include <ripple/app/ledger/TransactionMaster.h>
#include <ripple/app/rdb/backend/SQLiteDatabase.h>
#include <algorithm>
#include <map>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <vector>
namespace ripple {
class RWDBDatabase : public SQLiteDatabase
{
private:
struct LedgerData
{
LedgerInfo info;
std::map<uint256, AccountTx> transactions;
};
struct AccountTxData
{
AccountTxs transactions;
std::map<uint32_t, std::map<uint32_t, size_t>>
ledgerTxMap; // ledgerSeq -> txSeq -> index in transactions
};
Application& app_;
mutable std::shared_mutex mutex_;
std::map<LedgerIndex, LedgerData> ledgers_;
std::map<uint256, LedgerIndex> ledgerHashToSeq_;
std::map<uint256, AccountTx> transactionMap_;
std::map<AccountID, AccountTxData> accountTxMap_;
public:
RWDBDatabase(Application& app, Config const& config, JobQueue& jobQueue)
: app_(app)
{
}
std::optional<LedgerIndex>
getMinLedgerSeq() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
if (ledgers_.empty())
return std::nullopt;
return ledgers_.begin()->first;
}
std::optional<LedgerIndex>
getTransactionsMinLedgerSeq() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
if (transactionMap_.empty())
return std::nullopt;
return transactionMap_.begin()->second.second->getLgrSeq();
}
std::optional<LedgerIndex>
getAccountTransactionsMinLedgerSeq() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
if (accountTxMap_.empty())
return std::nullopt;
LedgerIndex minSeq = std::numeric_limits<LedgerIndex>::max();
for (const auto& [_, accountData] : accountTxMap_)
{
if (!accountData.ledgerTxMap.empty())
minSeq =
std::min(minSeq, accountData.ledgerTxMap.begin()->first);
}
return minSeq == std::numeric_limits<LedgerIndex>::max()
? std::nullopt
: std::optional<LedgerIndex>(minSeq);
}
std::optional<LedgerIndex>
getMaxLedgerSeq() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
if (ledgers_.empty())
return std::nullopt;
return ledgers_.rbegin()->first;
}
void
deleteTransactionByLedgerSeq(LedgerIndex ledgerSeq) override
{
std::unique_lock<std::shared_mutex> lock(mutex_);
auto it = ledgers_.find(ledgerSeq);
if (it != ledgers_.end())
{
for (const auto& [txHash, _] : it->second.transactions)
{
transactionMap_.erase(txHash);
}
it->second.transactions.clear();
}
for (auto& [_, accountData] : accountTxMap_)
{
accountData.ledgerTxMap.erase(ledgerSeq);
accountData.transactions.erase(
std::remove_if(
accountData.transactions.begin(),
accountData.transactions.end(),
[ledgerSeq](const AccountTx& tx) {
return tx.second->getLgrSeq() == ledgerSeq;
}),
accountData.transactions.end());
}
}
void
deleteBeforeLedgerSeq(LedgerIndex ledgerSeq) override
{
std::unique_lock<std::shared_mutex> lock(mutex_);
auto it = ledgers_.begin();
while (it != ledgers_.end() && it->first < ledgerSeq)
{
for (const auto& [txHash, _] : it->second.transactions)
{
transactionMap_.erase(txHash);
}
ledgerHashToSeq_.erase(it->second.info.hash);
it = ledgers_.erase(it);
}
for (auto& [_, accountData] : accountTxMap_)
{
auto txIt = accountData.ledgerTxMap.begin();
while (txIt != accountData.ledgerTxMap.end() &&
txIt->first < ledgerSeq)
{
txIt = accountData.ledgerTxMap.erase(txIt);
}
accountData.transactions.erase(
std::remove_if(
accountData.transactions.begin(),
accountData.transactions.end(),
[ledgerSeq](const AccountTx& tx) {
return tx.second->getLgrSeq() < ledgerSeq;
}),
accountData.transactions.end());
}
}
void
deleteTransactionsBeforeLedgerSeq(LedgerIndex ledgerSeq) override
{
std::unique_lock<std::shared_mutex> lock(mutex_);
for (auto& [seq, ledgerData] : ledgers_)
{
if (seq < ledgerSeq)
{
for (const auto& [txHash, _] : ledgerData.transactions)
{
transactionMap_.erase(txHash);
}
ledgerData.transactions.clear();
}
}
for (auto& [_, accountData] : accountTxMap_)
{
auto txIt = accountData.ledgerTxMap.begin();
while (txIt != accountData.ledgerTxMap.end() &&
txIt->first < ledgerSeq)
{
txIt = accountData.ledgerTxMap.erase(txIt);
}
accountData.transactions.erase(
std::remove_if(
accountData.transactions.begin(),
accountData.transactions.end(),
[ledgerSeq](const AccountTx& tx) {
return tx.second->getLgrSeq() < ledgerSeq;
}),
accountData.transactions.end());
}
}
void
deleteAccountTransactionsBeforeLedgerSeq(LedgerIndex ledgerSeq) override
{
std::unique_lock<std::shared_mutex> lock(mutex_);
for (auto& [_, accountData] : accountTxMap_)
{
auto txIt = accountData.ledgerTxMap.begin();
while (txIt != accountData.ledgerTxMap.end() &&
txIt->first < ledgerSeq)
{
txIt = accountData.ledgerTxMap.erase(txIt);
}
accountData.transactions.erase(
std::remove_if(
accountData.transactions.begin(),
accountData.transactions.end(),
[ledgerSeq](const AccountTx& tx) {
return tx.second->getLgrSeq() < ledgerSeq;
}),
accountData.transactions.end());
}
}
std::size_t
getTransactionCount() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
return transactionMap_.size();
}
std::size_t
getAccountTransactionCount() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
std::size_t count = 0;
for (const auto& [_, accountData] : accountTxMap_)
{
count += accountData.transactions.size();
}
return count;
}
CountMinMax
getLedgerCountMinMax() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
if (ledgers_.empty())
return {0, 0, 0};
return {
ledgers_.size(), ledgers_.begin()->first, ledgers_.rbegin()->first};
}
bool
saveValidatedLedger(
std::shared_ptr<Ledger const> const& ledger,
bool current) override
{
std::unique_lock<std::shared_mutex> lock(mutex_);
LedgerData ledgerData;
ledgerData.info = ledger->info();
auto aLedger = std::make_shared<AcceptedLedger>(ledger, app_);
for (auto const& acceptedLedgerTx : *aLedger)
{
auto const& txn = acceptedLedgerTx->getTxn();
auto const& meta = acceptedLedgerTx->getMeta();
auto const& id = txn->getTransactionID();
std::string reason;
auto accTx = std::make_pair(
std::make_shared<ripple::Transaction>(txn, reason, app_),
std::make_shared<ripple::TxMeta>(meta));
ledgerData.transactions.emplace(id, accTx);
transactionMap_.emplace(id, accTx);
for (auto const& account : meta.getAffectedAccounts())
{
if (accountTxMap_.find(account) == accountTxMap_.end())
accountTxMap_[account] = AccountTxData();
auto& accountData = accountTxMap_[account];
accountData.transactions.push_back(accTx);
accountData.ledgerTxMap[ledger->info().seq]
[acceptedLedgerTx->getTxnSeq()] =
accountData.transactions.size() - 1;
}
}
ledgers_[ledger->info().seq] = std::move(ledgerData);
ledgerHashToSeq_[ledger->info().hash] = ledger->info().seq;
if (current)
{
auto const cutoffSeq =
ledger->info().seq > app_.config().LEDGER_HISTORY
? ledger->info().seq - app_.config().LEDGER_HISTORY
: 0;
if (cutoffSeq > 0)
{
const std::size_t BATCH_SIZE = 128;
std::size_t deleted = 0;
std::vector<std::uint32_t> ledgersToDelete;
for (const auto& item : ledgers_)
{
if (deleted >= BATCH_SIZE)
break;
if (item.first < cutoffSeq)
{
ledgersToDelete.push_back(item.first);
deleted++;
}
}
for (auto seq : ledgersToDelete)
{
auto& ledgerToDelete = ledgers_[seq];
for (const auto& txPair : ledgerToDelete.transactions)
{
transactionMap_.erase(txPair.first);
}
ledgerHashToSeq_.erase(ledgerToDelete.info.hash);
ledgers_.erase(seq);
}
if (deleted > 0)
{
for (auto& [account, data] : accountTxMap_)
{
auto it = data.ledgerTxMap.begin();
while (it != data.ledgerTxMap.end())
{
if (it->first < cutoffSeq)
{
for (const auto& seqPair : it->second)
{
if (seqPair.second <
data.transactions.size())
{
auto& txPair =
data.transactions[seqPair.second];
txPair.first.reset();
txPair.second.reset();
}
}
it = data.ledgerTxMap.erase(it);
}
else
{
++it;
}
}
data.transactions.erase(
std::remove_if(
data.transactions.begin(),
data.transactions.end(),
[](const auto& tx) {
return !tx.first && !tx.second;
}),
data.transactions.end());
for (auto& [ledgerSeq, txMap] : data.ledgerTxMap)
{
for (auto& [txSeq, index] : txMap)
{
auto newIndex = std::distance(
data.transactions.begin(),
std::find(
data.transactions.begin(),
data.transactions.end(),
data.transactions[index]));
index = newIndex;
}
}
}
app_.getLedgerMaster().clearPriorLedgers(cutoffSeq);
}
}
}
return true;
}
std::optional<LedgerInfo>
getLedgerInfoByIndex(LedgerIndex ledgerSeq) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = ledgers_.find(ledgerSeq);
if (it != ledgers_.end())
return it->second.info;
return std::nullopt;
}
std::optional<LedgerInfo>
getNewestLedgerInfo() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
if (ledgers_.empty())
return std::nullopt;
return ledgers_.rbegin()->second.info;
}
std::optional<LedgerInfo>
getLimitedOldestLedgerInfo(LedgerIndex ledgerFirstIndex) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = ledgers_.lower_bound(ledgerFirstIndex);
if (it != ledgers_.end())
return it->second.info;
return std::nullopt;
}
std::optional<LedgerInfo>
getLimitedNewestLedgerInfo(LedgerIndex ledgerFirstIndex) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = ledgers_.lower_bound(ledgerFirstIndex);
if (it == ledgers_.end())
return std::nullopt;
return ledgers_.rbegin()->second.info;
}
std::optional<LedgerInfo>
getLedgerInfoByHash(uint256 const& ledgerHash) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = ledgerHashToSeq_.find(ledgerHash);
if (it != ledgerHashToSeq_.end())
return ledgers_.at(it->second).info;
return std::nullopt;
}
uint256
getHashByIndex(LedgerIndex ledgerIndex) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = ledgers_.find(ledgerIndex);
if (it != ledgers_.end())
return it->second.info.hash;
return uint256();
}
std::optional<LedgerHashPair>
getHashesByIndex(LedgerIndex ledgerIndex) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = ledgers_.find(ledgerIndex);
if (it != ledgers_.end())
{
return LedgerHashPair{
it->second.info.hash, it->second.info.parentHash};
}
return std::nullopt;
}
std::map<LedgerIndex, LedgerHashPair>
getHashesByIndex(LedgerIndex minSeq, LedgerIndex maxSeq) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
std::map<LedgerIndex, LedgerHashPair> result;
auto it = ledgers_.lower_bound(minSeq);
auto end = ledgers_.upper_bound(maxSeq);
for (; it != end; ++it)
{
result[it->first] = LedgerHashPair{
it->second.info.hash, it->second.info.parentHash};
}
return result;
}
std::variant<AccountTx, TxSearched>
getTransaction(
uint256 const& id,
std::optional<ClosedInterval<std::uint32_t>> const& range,
error_code_i& ec) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = transactionMap_.find(id);
if (it != transactionMap_.end())
{
const auto& [txn, txMeta] = it->second;
if (!range ||
(range->lower() <= txMeta->getLgrSeq() &&
txMeta->getLgrSeq() <= range->upper()))
return it->second;
}
if (range)
{
bool allPresent = true;
for (LedgerIndex seq = range->lower(); seq <= range->upper(); ++seq)
{
if (ledgers_.find(seq) == ledgers_.end())
{
allPresent = false;
break;
}
}
return allPresent ? TxSearched::all : TxSearched::some;
}
return TxSearched::unknown;
}
bool
ledgerDbHasSpace(Config const& config) override
{
return true; // In-memory database always has space
}
bool
transactionDbHasSpace(Config const& config) override
{
return true; // In-memory database always has space
}
std::uint32_t
getKBUsedAll() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
std::uint32_t size = sizeof(*this);
size += ledgers_.size() * (sizeof(LedgerIndex) + sizeof(LedgerData));
size +=
ledgerHashToSeq_.size() * (sizeof(uint256) + sizeof(LedgerIndex));
size += transactionMap_.size() * (sizeof(uint256) + sizeof(AccountTx));
for (const auto& [_, accountData] : accountTxMap_)
{
size += sizeof(AccountID) + sizeof(AccountTxData);
size += accountData.transactions.size() * sizeof(AccountTx);
for (const auto& [_, innerMap] : accountData.ledgerTxMap)
{
size += sizeof(uint32_t) +
innerMap.size() * (sizeof(uint32_t) + sizeof(size_t));
}
}
return size / 1024;
}
std::uint32_t
getKBUsedLedger() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
std::uint32_t size = 0;
size += ledgers_.size() * (sizeof(LedgerIndex) + sizeof(LedgerData));
size +=
ledgerHashToSeq_.size() * (sizeof(uint256) + sizeof(LedgerIndex));
return size / 1024;
}
std::uint32_t
getKBUsedTransaction() override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
std::uint32_t size = 0;
size += transactionMap_.size() * (sizeof(uint256) + sizeof(AccountTx));
for (const auto& [_, accountData] : accountTxMap_)
{
size += sizeof(AccountID) + sizeof(AccountTxData);
size += accountData.transactions.size() * sizeof(AccountTx);
for (const auto& [_, innerMap] : accountData.ledgerTxMap)
{
size += sizeof(uint32_t) +
innerMap.size() * (sizeof(uint32_t) + sizeof(size_t));
}
}
return size / 1024;
}
void
closeLedgerDB() override
{
// No-op for in-memory database
}
void
closeTransactionDB() override
{
// No-op for in-memory database
}
~RWDBDatabase()
{
// Regular maps can use standard clear
accountTxMap_.clear();
transactionMap_.clear();
for (auto& ledger : ledgers_)
{
ledger.second.transactions.clear();
}
ledgers_.clear();
ledgerHashToSeq_.clear();
}
std::vector<std::shared_ptr<Transaction>>
getTxHistory(LedgerIndex startIndex) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
std::vector<std::shared_ptr<Transaction>> result;
auto it = ledgers_.lower_bound(startIndex);
int count = 0;
while (it != ledgers_.end() && count < 20)
{
for (const auto& [txHash, accountTx] : it->second.transactions)
{
result.push_back(accountTx.first);
if (++count >= 20)
break;
}
++it;
}
return result;
}
// Helper function to handle limits
template <typename Container>
void
applyLimit(Container& container, std::size_t limit, bool bUnlimited)
{
if (!bUnlimited && limit > 0 && container.size() > limit)
{
container.resize(limit);
}
}
AccountTxs
getOldestAccountTxs(AccountTxOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {};
AccountTxs result;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
std::size_t skipped = 0;
for (; txIt != txEnd &&
(options.bUnlimited || result.size() < options.limit);
++txIt)
{
for (const auto& [txSeq, txIndex] : txIt->second)
{
if (skipped < options.offset)
{
++skipped;
continue;
}
result.push_back(accountData.transactions[txIndex]);
if (!options.bUnlimited && result.size() >= options.limit)
break;
}
}
return result;
}
AccountTxs
getNewestAccountTxs(AccountTxOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {};
AccountTxs result;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
std::size_t skipped = 0;
for (auto rIt = std::make_reverse_iterator(txEnd);
rIt != std::make_reverse_iterator(txIt) &&
(options.bUnlimited || result.size() < options.limit);
++rIt)
{
for (auto innerRIt = rIt->second.rbegin();
innerRIt != rIt->second.rend();
++innerRIt)
{
if (skipped < options.offset)
{
++skipped;
continue;
}
result.push_back(accountData.transactions[innerRIt->second]);
if (!options.bUnlimited && result.size() >= options.limit)
break;
}
}
return result;
}
MetaTxsList
getOldestAccountTxsB(AccountTxOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {};
MetaTxsList result;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
std::size_t skipped = 0;
for (; txIt != txEnd &&
(options.bUnlimited || result.size() < options.limit);
++txIt)
{
for (const auto& [txSeq, txIndex] : txIt->second)
{
if (skipped < options.offset)
{
++skipped;
continue;
}
const auto& [txn, txMeta] = accountData.transactions[txIndex];
result.emplace_back(
txn->getSTransaction()->getSerializer().peekData(),
txMeta->getAsObject().getSerializer().peekData(),
txIt->first);
if (!options.bUnlimited && result.size() >= options.limit)
break;
}
}
return result;
}
MetaTxsList
getNewestAccountTxsB(AccountTxOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {};
MetaTxsList result;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
std::size_t skipped = 0;
for (auto rIt = std::make_reverse_iterator(txEnd);
rIt != std::make_reverse_iterator(txIt) &&
(options.bUnlimited || result.size() < options.limit);
++rIt)
{
for (auto innerRIt = rIt->second.rbegin();
innerRIt != rIt->second.rend();
++innerRIt)
{
if (skipped < options.offset)
{
++skipped;
continue;
}
const auto& [txn, txMeta] =
accountData.transactions[innerRIt->second];
result.emplace_back(
txn->getSTransaction()->getSerializer().peekData(),
txMeta->getAsObject().getSerializer().peekData(),
rIt->first);
if (!options.bUnlimited && result.size() >= options.limit)
break;
}
}
return result;
}
std::pair<AccountTxs, std::optional<AccountTxMarker>>
oldestAccountTxPage(AccountTxPageOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {{}, std::nullopt};
AccountTxs result;
std::optional<AccountTxMarker> marker;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
bool lookingForMarker = options.marker.has_value();
std::size_t count = 0;
for (; txIt != txEnd && (options.limit == 0 || count < options.limit);
++txIt)
{
for (const auto& [txSeq, txIndex] : txIt->second)
{
if (lookingForMarker)
{
if (txIt->first == options.marker->ledgerSeq &&
txSeq == options.marker->txnSeq)
lookingForMarker = false;
continue;
}
result.push_back(accountData.transactions[txIndex]);
++count;
if (options.limit > 0 && count >= options.limit)
{
marker = AccountTxMarker{txIt->first, txSeq};
break;
}
}
}
return {result, marker};
}
std::pair<AccountTxs, std::optional<AccountTxMarker>>
newestAccountTxPage(AccountTxPageOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {{}, std::nullopt};
AccountTxs result;
std::optional<AccountTxMarker> marker;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
bool lookingForMarker = options.marker.has_value();
std::size_t count = 0;
for (auto rIt = std::make_reverse_iterator(txEnd);
rIt != std::make_reverse_iterator(txIt) &&
(options.limit == 0 || count < options.limit);
++rIt)
{
for (auto innerRIt = rIt->second.rbegin();
innerRIt != rIt->second.rend();
++innerRIt)
{
if (lookingForMarker)
{
if (rIt->first == options.marker->ledgerSeq &&
innerRIt->first == options.marker->txnSeq)
lookingForMarker = false;
continue;
}
result.push_back(accountData.transactions[innerRIt->second]);
++count;
if (options.limit > 0 && count >= options.limit)
{
marker = AccountTxMarker{rIt->first, innerRIt->first};
break;
}
}
}
return {result, marker};
}
std::pair<MetaTxsList, std::optional<AccountTxMarker>>
oldestAccountTxPageB(AccountTxPageOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {{}, std::nullopt};
MetaTxsList result;
std::optional<AccountTxMarker> marker;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
bool lookingForMarker = options.marker.has_value();
std::size_t count = 0;
for (; txIt != txEnd && (options.limit == 0 || count < options.limit);
++txIt)
{
for (const auto& [txSeq, txIndex] : txIt->second)
{
if (lookingForMarker)
{
if (txIt->first == options.marker->ledgerSeq &&
txSeq == options.marker->txnSeq)
lookingForMarker = false;
continue;
}
const auto& [txn, txMeta] = accountData.transactions[txIndex];
result.emplace_back(
txn->getSTransaction()->getSerializer().peekData(),
txMeta->getAsObject().getSerializer().peekData(),
txIt->first);
++count;
if (options.limit > 0 && count >= options.limit)
{
marker = AccountTxMarker{txIt->first, txSeq};
break;
}
}
}
return {result, marker};
}
std::pair<MetaTxsList, std::optional<AccountTxMarker>>
newestAccountTxPageB(AccountTxPageOptions const& options) override
{
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = accountTxMap_.find(options.account);
if (it == accountTxMap_.end())
return {{}, std::nullopt};
MetaTxsList result;
std::optional<AccountTxMarker> marker;
const auto& accountData = it->second;
auto txIt = accountData.ledgerTxMap.lower_bound(options.minLedger);
auto txEnd = accountData.ledgerTxMap.upper_bound(options.maxLedger);
bool lookingForMarker = options.marker.has_value();
std::size_t count = 0;
for (auto rIt = std::make_reverse_iterator(txEnd);
rIt != std::make_reverse_iterator(txIt) &&
(options.limit == 0 || count < options.limit);
++rIt)
{
for (auto innerRIt = rIt->second.rbegin();
innerRIt != rIt->second.rend();
++innerRIt)
{
if (lookingForMarker)
{
if (rIt->first == options.marker->ledgerSeq &&
innerRIt->first == options.marker->txnSeq)
lookingForMarker = false;
continue;
}
const auto& [txn, txMeta] =
accountData.transactions[innerRIt->second];
result.emplace_back(
txn->getSTransaction()->getSerializer().peekData(),
txMeta->getAsObject().getSerializer().peekData(),
rIt->first);
++count;
if (options.limit > 0 && count >= options.limit)
{
marker = AccountTxMarker{rIt->first, innerRIt->first};
break;
}
}
}
return {result, marker};
}
};
// Factory function
std::unique_ptr<SQLiteDatabase>
getRWDBDatabase(Application& app, Config const& config, JobQueue& jobQueue)
{
return std::make_unique<RWDBDatabase>(app, config, jobQueue);
}
} // namespace ripple
#endif // RIPPLE_APP_RDB_BACKEND_MEMORYDATABASE_H_INCLUDED

View File

@@ -19,6 +19,8 @@
#include <ripple/app/main/Application.h> #include <ripple/app/main/Application.h>
#include <ripple/app/rdb/RelationalDatabase.h> #include <ripple/app/rdb/RelationalDatabase.h>
#include <ripple/app/rdb/backend/FlatmapDatabase.h>
#include <ripple/app/rdb/backend/RWDBDatabase.h>
#include <ripple/core/ConfigSections.h> #include <ripple/core/ConfigSections.h>
#include <ripple/nodestore/DatabaseShard.h> #include <ripple/nodestore/DatabaseShard.h>
@@ -38,6 +40,8 @@ RelationalDatabase::init(
{ {
bool use_sqlite = false; bool use_sqlite = false;
bool use_postgres = false; bool use_postgres = false;
bool use_rwdb = false;
bool use_flatmap = false;
if (config.reporting()) if (config.reporting())
{ {
@@ -52,6 +56,14 @@ RelationalDatabase::init(
{ {
use_sqlite = true; use_sqlite = true;
} }
else if (boost::iequals(get(rdb_section, "backend"), "rwdb"))
{
use_rwdb = true;
}
else if (boost::iequals(get(rdb_section, "backend"), "flatmap"))
{
use_flatmap = true;
}
else else
{ {
Throw<std::runtime_error>( Throw<std::runtime_error>(
@@ -73,6 +85,14 @@ RelationalDatabase::init(
{ {
return getPostgresDatabase(app, config, jobQueue); return getPostgresDatabase(app, config, jobQueue);
} }
else if (use_rwdb)
{
return getRWDBDatabase(app, config, jobQueue);
}
else if (use_flatmap)
{
return getFlatmapDatabase(app, config, jobQueue);
}
return std::unique_ptr<RelationalDatabase>(); return std::unique_ptr<RelationalDatabase>();
} }

View File

@@ -25,6 +25,7 @@
#include <ripple/basics/base_uint.h> #include <ripple/basics/base_uint.h>
#include <ripple/beast/net/IPEndpoint.h> #include <ripple/beast/net/IPEndpoint.h>
#include <ripple/beast/utility/Journal.h> #include <ripple/beast/utility/Journal.h>
#include <ripple/core/ConfigSections.h>
#include <ripple/protocol/PublicKey.h> #include <ripple/protocol/PublicKey.h>
#include <ripple/protocol/SystemParameters.h> // VFALCO Breaks levelization #include <ripple/protocol/SystemParameters.h> // VFALCO Breaks levelization
#include <boost/beast/core/string.hpp> #include <boost/beast/core/string.hpp>
@@ -350,6 +351,21 @@ public:
{ {
return RUN_REPORTING; return RUN_REPORTING;
} }
bool
mem_backend() const
{
static bool const isMem =
(!section(SECTION_RELATIONAL_DB).empty() &&
boost::beast::iequals(
get(section(SECTION_RELATIONAL_DB), "backend"), "rwdb")) ||
(!section("node_db").empty() &&
(boost::beast::iequals(get(section("node_db"), "type"), "rwdb") ||
boost::beast::iequals(
get(section("node_db"), "type"), "flatmap")));
// RHNOTE: memory type is not selected for here because it breaks
// tests
return isMem;
}
bool bool
useTxTables() const useTxTables() const

View File

@@ -0,0 +1,235 @@
#include <ripple/basics/contract.h>
#include <ripple/nodestore/Factory.h>
#include <ripple/nodestore/Manager.h>
#include <ripple/nodestore/impl/DecodedBlob.h>
#include <ripple/nodestore/impl/EncodedBlob.h>
#include <ripple/nodestore/impl/codec.h>
#include <boost/beast/core/string.hpp>
#include <boost/core/ignore_unused.hpp>
#include <boost/unordered/concurrent_flat_map.hpp>
#include <memory>
#include <mutex>
namespace ripple {
namespace NodeStore {
class FlatmapBackend : public Backend
{
private:
std::string name_;
beast::Journal journal_;
bool isOpen_{false};
struct base_uint_hasher
{
using result_type = std::size_t;
result_type
operator()(base_uint<256> const& value) const
{
return hardened_hash<>{}(value);
}
};
using DataStore = boost::unordered::concurrent_flat_map<
uint256,
std::vector<std::uint8_t>, // Store compressed blob data
base_uint_hasher>;
DataStore table_;
public:
FlatmapBackend(
size_t keyBytes,
Section const& keyValues,
beast::Journal journal)
: name_(get(keyValues, "path")), journal_(journal)
{
boost::ignore_unused(journal_);
if (name_.empty())
name_ = "node_db";
}
~FlatmapBackend() override
{
close();
}
std::string
getName() override
{
return name_;
}
void
open(bool createIfMissing) override
{
if (isOpen_)
Throw<std::runtime_error>("already open");
isOpen_ = true;
}
bool
isOpen() override
{
return isOpen_;
}
void
close() override
{
table_.clear();
isOpen_ = false;
}
Status
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
{
if (!isOpen_)
return notFound;
uint256 const hash(uint256::fromVoid(key));
bool found = table_.visit(hash, [&](const auto& key_value_pair) {
nudb::detail::buffer bf;
auto const result = nodeobject_decompress(
key_value_pair.second.data(), key_value_pair.second.size(), bf);
DecodedBlob decoded(hash.data(), result.first, result.second);
if (!decoded.wasOk())
{
*pObject = nullptr;
return;
}
*pObject = decoded.createObject();
});
return found ? (*pObject ? ok : dataCorrupt) : notFound;
}
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256 const*> const& hashes) override
{
std::vector<std::shared_ptr<NodeObject>> results;
results.reserve(hashes.size());
for (auto const& h : hashes)
{
std::shared_ptr<NodeObject> nObj;
Status status = fetch(h->begin(), &nObj);
if (status != ok)
results.push_back({});
else
results.push_back(nObj);
}
return {results, ok};
}
void
store(std::shared_ptr<NodeObject> const& object) override
{
if (!isOpen_)
return;
if (!object)
return;
EncodedBlob encoded(object);
nudb::detail::buffer bf;
auto const result =
nodeobject_compress(encoded.getData(), encoded.getSize(), bf);
std::vector<std::uint8_t> compressed(
static_cast<const std::uint8_t*>(result.first),
static_cast<const std::uint8_t*>(result.first) + result.second);
table_.insert_or_assign(object->getHash(), std::move(compressed));
}
void
storeBatch(Batch const& batch) override
{
for (auto const& e : batch)
store(e);
}
void
sync() override
{
}
void
for_each(std::function<void(std::shared_ptr<NodeObject>)> f) override
{
if (!isOpen_)
return;
table_.visit_all([&f](const auto& entry) {
nudb::detail::buffer bf;
auto const result = nodeobject_decompress(
entry.second.data(), entry.second.size(), bf);
DecodedBlob decoded(
entry.first.data(), result.first, result.second);
if (decoded.wasOk())
f(decoded.createObject());
});
}
int
getWriteLoad() override
{
return 0;
}
void
setDeletePath() override
{
close();
}
int
fdRequired() const override
{
return 0;
}
private:
size_t
size() const
{
return table_.size();
}
};
class FlatmapFactory : public Factory
{
public:
FlatmapFactory()
{
Manager::instance().insert(*this);
}
~FlatmapFactory() override
{
Manager::instance().erase(*this);
}
std::string
getName() const override
{
return "Flatmap";
}
std::unique_ptr<Backend>
createInstance(
size_t keyBytes,
Section const& keyValues,
std::size_t burstSize,
Scheduler& scheduler,
beast::Journal journal) override
{
return std::make_unique<FlatmapBackend>(keyBytes, keyValues, journal);
}
};
static FlatmapFactory flatmapFactory;
} // namespace NodeStore
} // namespace ripple

View File

@@ -94,7 +94,7 @@ public:
{ {
boost::ignore_unused(journal_); // Keep unused journal_ just in case. boost::ignore_unused(journal_); // Keep unused journal_ just in case.
if (name_.empty()) if (name_.empty())
Throw<std::runtime_error>("Missing path in Memory backend"); Throw<std::runtime_error>("Missing path in TestMemory backend");
} }
~MemoryBackend() override ~MemoryBackend() override

View File

@@ -0,0 +1,242 @@
#include <ripple/basics/contract.h>
#include <ripple/nodestore/Factory.h>
#include <ripple/nodestore/Manager.h>
#include <ripple/nodestore/impl/DecodedBlob.h>
#include <ripple/nodestore/impl/EncodedBlob.h>
#include <ripple/nodestore/impl/codec.h>
#include <boost/beast/core/string.hpp>
#include <boost/core/ignore_unused.hpp>
#include <boost/unordered/concurrent_flat_map.hpp>
#include <memory>
#include <mutex>
namespace ripple {
namespace NodeStore {
class RWDBBackend : public Backend
{
private:
std::string name_;
beast::Journal journal_;
bool isOpen_{false};
struct base_uint_hasher
{
using result_type = std::size_t;
result_type
operator()(base_uint<256> const& value) const
{
return hardened_hash<>{}(value);
}
};
using DataStore =
std::map<uint256, std::vector<std::uint8_t>>; // Store compressed blob
// data
mutable std::recursive_mutex
mutex_; // Only needed for std::map implementation
DataStore table_;
public:
RWDBBackend(
size_t keyBytes,
Section const& keyValues,
beast::Journal journal)
: name_(get(keyValues, "path")), journal_(journal)
{
boost::ignore_unused(journal_);
if (name_.empty())
name_ = "node_db";
}
~RWDBBackend() override
{
close();
}
std::string
getName() override
{
return name_;
}
void
open(bool createIfMissing) override
{
std::lock_guard lock(mutex_);
if (isOpen_)
Throw<std::runtime_error>("already open");
isOpen_ = true;
}
bool
isOpen() override
{
return isOpen_;
}
void
close() override
{
std::lock_guard lock(mutex_);
table_.clear();
isOpen_ = false;
}
Status
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
{
if (!isOpen_)
return notFound;
uint256 const hash(uint256::fromVoid(key));
std::lock_guard lock(mutex_);
auto it = table_.find(hash);
if (it == table_.end())
return notFound;
nudb::detail::buffer bf;
auto const result =
nodeobject_decompress(it->second.data(), it->second.size(), bf);
DecodedBlob decoded(hash.data(), result.first, result.second);
if (!decoded.wasOk())
return dataCorrupt;
*pObject = decoded.createObject();
return ok;
}
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256 const*> const& hashes) override
{
std::vector<std::shared_ptr<NodeObject>> results;
results.reserve(hashes.size());
for (auto const& h : hashes)
{
std::shared_ptr<NodeObject> nObj;
Status status = fetch(h->begin(), &nObj);
if (status != ok)
results.push_back({});
else
results.push_back(nObj);
}
return {results, ok};
}
void
store(std::shared_ptr<NodeObject> const& object) override
{
if (!isOpen_)
return;
if (!object)
return;
EncodedBlob encoded(object);
nudb::detail::buffer bf;
auto const result =
nodeobject_compress(encoded.getData(), encoded.getSize(), bf);
std::vector<std::uint8_t> compressed(
static_cast<const std::uint8_t*>(result.first),
static_cast<const std::uint8_t*>(result.first) + result.second);
std::lock_guard lock(mutex_);
table_[object->getHash()] = std::move(compressed);
}
void
storeBatch(Batch const& batch) override
{
for (auto const& e : batch)
store(e);
}
void
sync() override
{
}
void
for_each(std::function<void(std::shared_ptr<NodeObject>)> f) override
{
if (!isOpen_)
return;
std::lock_guard lock(mutex_);
for (const auto& entry : table_)
{
nudb::detail::buffer bf;
auto const result = nodeobject_decompress(
entry.second.data(), entry.second.size(), bf);
DecodedBlob decoded(
entry.first.data(), result.first, result.second);
if (decoded.wasOk())
f(decoded.createObject());
}
}
int
getWriteLoad() override
{
return 0;
}
void
setDeletePath() override
{
close();
}
int
fdRequired() const override
{
return 0;
}
private:
size_t
size() const
{
std::lock_guard lock(mutex_);
return table_.size();
}
};
class RWDBFactory : public Factory
{
public:
RWDBFactory()
{
Manager::instance().insert(*this);
}
~RWDBFactory() override
{
Manager::instance().erase(*this);
}
std::string
getName() const override
{
return "RWDB";
}
std::unique_ptr<Backend>
createInstance(
size_t keyBytes,
Section const& keyValues,
std::size_t burstSize,
Scheduler& scheduler,
beast::Journal journal) override
{
return std::make_unique<RWDBBackend>(keyBytes, keyValues, journal);
}
};
static RWDBFactory rwDBFactory;
} // namespace NodeStore
} // namespace ripple

View File

@@ -38,6 +38,7 @@
#include <ripple/rpc/json_body.h> #include <ripple/rpc/json_body.h>
#include <ripple/server/SimpleWriter.h> #include <ripple/server/SimpleWriter.h>
#include <ripple/core/ConfigSections.h>
#include <boost/algorithm/string/predicate.hpp> #include <boost/algorithm/string/predicate.hpp>
#include <boost/utility/in_place_factory.hpp> #include <boost/utility/in_place_factory.hpp>
@@ -136,7 +137,11 @@ OverlayImpl::OverlayImpl(
stopwatch(), stopwatch(),
app_.journal("PeerFinder"), app_.journal("PeerFinder"),
config, config,
collector)) collector,
app.config().section(SECTION_RELATIONAL_DB).empty() ||
!boost::iequals(
get(app.config().section(SECTION_RELATIONAL_DB), "backend"),
"rwdb")))
, m_resolver(resolver) , m_resolver(resolver)
, next_id_(1) , next_id_(1)
, timer_count_(0) , timer_count_(0)

View File

@@ -0,0 +1,54 @@
#ifndef RIPPLE_PEERFINDER_INMEMORYSTORE_H_INCLUDED
#define RIPPLE_PEERFINDER_INMEMORYSTORE_H_INCLUDED
#include <ripple/beast/net/IPEndpoint.h>
#include <ripple/peerfinder/impl/Store.h>
#include <boost/functional/hash.hpp>
#include <boost/unordered/concurrent_flat_map.hpp>
namespace ripple {
namespace PeerFinder {
struct EndpointHasher
{
std::size_t
operator()(beast::IP::Endpoint const& endpoint) const
{
std::size_t seed = 0;
boost::hash_combine(seed, endpoint.address().to_string());
boost::hash_combine(seed, endpoint.port());
return seed;
}
};
class InMemoryStore : public Store
{
private:
boost::concurrent_flat_map<beast::IP::Endpoint, int, EndpointHasher>
entries;
public:
std::size_t
load(load_callback const& cb) override
{
std::size_t count = 0;
entries.visit_all([&](auto const& entry) {
cb(entry.first, entry.second);
++count;
});
return count;
}
void
save(std::vector<Entry> const& v) override
{
entries.clear();
for (auto const& entry : v)
entries.emplace(entry.endpoint, entry.valence);
}
};
} // namespace PeerFinder
} // namespace ripple
#endif

View File

@@ -17,8 +17,10 @@
*/ */
//============================================================================== //==============================================================================
#include <ripple/core/ConfigSections.h>
#include <ripple/peerfinder/PeerfinderManager.h> #include <ripple/peerfinder/PeerfinderManager.h>
#include <ripple/peerfinder/impl/Checker.h> #include <ripple/peerfinder/impl/Checker.h>
#include <ripple/peerfinder/impl/InMemoryStore.h>
#include <ripple/peerfinder/impl/Logic.h> #include <ripple/peerfinder/impl/Logic.h>
#include <ripple/peerfinder/impl/SourceStrings.h> #include <ripple/peerfinder/impl/SourceStrings.h>
#include <ripple/peerfinder/impl/StoreSqdb.h> #include <ripple/peerfinder/impl/StoreSqdb.h>
@@ -38,7 +40,7 @@ public:
std::optional<boost::asio::io_service::work> work_; std::optional<boost::asio::io_service::work> work_;
clock_type& m_clock; clock_type& m_clock;
beast::Journal m_journal; beast::Journal m_journal;
StoreSqdb m_store; std::unique_ptr<Store> m_store;
Checker<boost::asio::ip::tcp> checker_; Checker<boost::asio::ip::tcp> checker_;
Logic<decltype(checker_)> m_logic; Logic<decltype(checker_)> m_logic;
BasicConfig const& m_config; BasicConfig const& m_config;
@@ -50,15 +52,18 @@ public:
clock_type& clock, clock_type& clock,
beast::Journal journal, beast::Journal journal,
BasicConfig const& config, BasicConfig const& config,
beast::insight::Collector::ptr const& collector) beast::insight::Collector::ptr const& collector,
bool useSqLiteStore)
: Manager() : Manager()
, io_service_(io_service) , io_service_(io_service)
, work_(std::in_place, std::ref(io_service_)) , work_(std::in_place, std::ref(io_service_))
, m_clock(clock) , m_clock(clock)
, m_journal(journal) , m_journal(journal)
, m_store(journal) , m_store(
useSqLiteStore ? static_cast<Store*>(new StoreSqdb(journal))
: static_cast<Store*>(new InMemoryStore()))
, checker_(io_service_) , checker_(io_service_)
, m_logic(clock, m_store, checker_, journal) , m_logic(clock, *m_store, checker_, journal)
, m_config(config) , m_config(config)
, m_stats(std::bind(&ManagerImp::collect_metrics, this), collector) , m_stats(std::bind(&ManagerImp::collect_metrics, this), collector)
{ {
@@ -215,7 +220,8 @@ public:
void void
start() override start() override
{ {
m_store.open(m_config); if (auto sqdb = dynamic_cast<StoreSqdb*>(m_store.get()))
sqdb->open(m_config);
m_logic.load(); m_logic.load();
} }
@@ -275,10 +281,11 @@ make_Manager(
clock_type& clock, clock_type& clock,
beast::Journal journal, beast::Journal journal,
BasicConfig const& config, BasicConfig const& config,
beast::insight::Collector::ptr const& collector) beast::insight::Collector::ptr const& collector,
bool useSqLiteStore)
{ {
return std::make_unique<ManagerImp>( return std::make_unique<ManagerImp>(
io_service, clock, journal, config, collector); io_service, clock, journal, config, collector, useSqLiteStore);
} }
} // namespace PeerFinder } // namespace PeerFinder

View File

@@ -34,7 +34,8 @@ make_Manager(
clock_type& clock, clock_type& clock,
beast::Journal journal, beast::Journal journal,
BasicConfig const& config, BasicConfig const& config,
beast::insight::Collector::ptr const& collector); beast::insight::Collector::ptr const& collector,
bool useSqliteStore);
} // namespace PeerFinder } // namespace PeerFinder
} // namespace ripple } // namespace ripple

View File

@@ -74,7 +74,7 @@ namespace detail {
// Feature.cpp. Because it's only used to reserve storage, and determine how // Feature.cpp. Because it's only used to reserve storage, and determine how
// large to make the FeatureBitset, it MAY be larger. It MUST NOT be less than // large to make the FeatureBitset, it MAY be larger. It MUST NOT be less than
// the actual number of amendments. A LogicError on startup will verify this. // the actual number of amendments. A LogicError on startup will verify this.
static constexpr std::size_t numFeatures = 73; static constexpr std::size_t numFeatures = 74;
/** Amendments that this server supports and the default voting behavior. /** Amendments that this server supports and the default voting behavior.
Whether they are enabled depends on the Rules defined in the validated Whether they are enabled depends on the Rules defined in the validated
@@ -361,6 +361,7 @@ extern uint256 const fixNSDelete;
extern uint256 const fix240819; extern uint256 const fix240819;
extern uint256 const fixPageCap; extern uint256 const fixPageCap;
extern uint256 const fix240911; extern uint256 const fix240911;
extern uint256 const fixFloatDivide;
} // namespace ripple } // namespace ripple

View File

@@ -467,6 +467,7 @@ REGISTER_FIX (fixNSDelete, Supported::yes, VoteBehavior::De
REGISTER_FIX (fix240819, Supported::yes, VoteBehavior::DefaultYes); REGISTER_FIX (fix240819, Supported::yes, VoteBehavior::DefaultYes);
REGISTER_FIX (fixPageCap, Supported::yes, VoteBehavior::DefaultYes); REGISTER_FIX (fixPageCap, Supported::yes, VoteBehavior::DefaultYes);
REGISTER_FIX (fix240911, Supported::yes, VoteBehavior::DefaultYes); REGISTER_FIX (fix240911, Supported::yes, VoteBehavior::DefaultYes);
REGISTER_FIX (fixFloatDivide, Supported::yes, VoteBehavior::DefaultYes);
// The following amendments are obsolete, but must remain supported // The following amendments are obsolete, but must remain supported
// because they could potentially get enabled. // because they could potentially get enabled.

View File

@@ -239,6 +239,7 @@ SHAMap::checkFilter(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
// Get a node without throwing // Get a node without throwing
// Used on maps where missing nodes are expected // Used on maps where missing nodes are expected
/*
std::shared_ptr<SHAMapTreeNode> std::shared_ptr<SHAMapTreeNode>
SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
{ {
@@ -261,6 +262,49 @@ SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
return node; return node;
} }
*/
std::shared_ptr<SHAMapTreeNode>
SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
{
using namespace std::chrono;
auto start = high_resolution_clock::now();
auto timeout = nanoseconds(50);
while (true)
{
// Try to fetch from cache first
auto node = cacheLookup(hash);
if (node)
return node;
if (backed_)
{
node = fetchNodeFromDB(hash);
if (node)
{
canonicalize(hash, node);
return node;
}
}
if (filter)
node = checkFilter(hash, filter);
if (node)
return node;
// Check if we've exceeded timeout
auto elapsed = high_resolution_clock::now() - start;
if (elapsed >= timeout)
break;
// Short yield to avoid overwhelming CPU
std::this_thread::yield();
}
return nullptr;
}
std::shared_ptr<SHAMapTreeNode> std::shared_ptr<SHAMapTreeNode>
SHAMap::fetchNodeNT(SHAMapHash const& hash) const SHAMap::fetchNodeNT(SHAMapHash const& hash) const

View File

@@ -82,6 +82,8 @@ public:
} }
} }
// rwdb backend does not keep table/data after close
if (type != "rwdb")
{ {
// Re-open the backend // Re-open the backend
std::unique_ptr<Backend> backend = Manager::instance().make_Backend( std::unique_ptr<Backend> backend = Manager::instance().make_Backend(
@@ -105,6 +107,8 @@ public:
{ {
std::uint64_t const seedValue = 50; std::uint64_t const seedValue = 50;
testBackend("memory", seedValue);
testBackend("rwdb", seedValue);
testBackend("nudb", seedValue); testBackend("nudb", seedValue);
#if RIPPLE_ROCKSDB_AVAILABLE #if RIPPLE_ROCKSDB_AVAILABLE
@@ -117,7 +121,7 @@ public:
} }
}; };
BEAST_DEFINE_TESTSUITE(Backend, ripple_core, ripple); BEAST_DEFINE_TESTSUITE(Backend, NodeStore, ripple);
} // namespace NodeStore } // namespace NodeStore
} // namespace ripple } // namespace ripple

View File

@@ -661,6 +661,8 @@ public:
testNodeStore("memory", false, seedValue); testNodeStore("memory", false, seedValue);
testNodeStore("rwdb", false, seedValue);
// Persistent backend tests // Persistent backend tests
{ {
testNodeStore("nudb", true, seedValue); testNodeStore("nudb", true, seedValue);