Compare commits

..

41 Commits

Author SHA1 Message Date
Denis Angell
cac563038d [fold] rename rwdb 2024-11-11 23:08:51 +01:00
Denis Angell
d33ee9605e [fold] clang-format 2024-11-11 21:31:26 +01:00
Denis Angell
8c00fd10d1 [fold] rename MemoryFactory -> MemDBFactory 2024-11-11 21:26:27 +01:00
Denis Angell
febdb6a997 [fold] revert log level 2024-11-11 21:09:41 +01:00
Denis Angell
d4f97ec974 [fold] remove tech debt 2024-11-11 21:03:01 +01:00
Denis Angell
a126286b9a [fold] add memory tests 2024-11-11 15:57:39 +01:00
Richard Holland
c90c732336 clang 2024-11-11 06:40:43 +11:00
Richard Holland
0dd02d8f8a remove tagged cache hack 2024-11-11 06:33:50 +11:00
Richard Holland
a3b1bbf1fe clang 2024-11-10 16:25:21 +11:00
Richard Holland
715fa840ff ensure db rotate doesnt happen on testmem nodedb 2024-11-10 16:21:08 +11:00
Richard Holland
470b9db5c3 whoops 2024-11-10 15:08:33 +11:00
Richard Holland
d88902f687 add legacy memory type called testmemory 2024-11-10 15:08:15 +11:00
Richard Holland
e031e50678 make memory db backend and flatmap backend different classes and selectable at runtime 2024-11-10 11:09:27 +11:00
Richard Holland
dd124124d6 clang 2024-11-10 08:41:57 +11:00
RichardAH
b54b01c3a6 Merge branch 'dev' into memdb_concurrent 2024-11-09 15:17:24 +10:00
Denis Angell
58b22901cb Fix: float_divide rounding error (#351)
Co-authored-by: RichardAH <richard.holland@starstone.co.nz>
2024-11-09 15:17:00 +10:00
Denis Angell
8ba37a3138 Add Script for SfCode generation (#358)
Co-authored-by: RichardAH <richard.holland@starstone.co.nz>
Co-authored-by: tequ <git@tequ.dev>
2024-11-09 14:17:49 +10:00
tequ
8cffd3054d add trace message to exception on etxn_fee_base (#387) 2024-11-09 14:00:59 +10:00
Richard Holland
59ceedc815 remove excess print statements and change spinlock to 100ns 2024-11-09 12:27:53 +11:00
Richard Holland
8520d95a2d fix empty account_tx bug 2024-11-09 12:09:06 +11:00
Richard Holland
2cc022c1df clang 2024-11-09 09:57:42 +11:00
Richard Holland
1986192e40 reorg the rdb mem backend to mirror how the nodestore mem backend handles concurrent vs nonconcurrent impl 2024-11-09 09:46:50 +11:00
Richard Holland
b95c4dc4ab fix potential mem leak in rdb backend 2024-11-09 09:41:14 +11:00
Richard Holland
b8c57e07d8 simplify memdb, use serder, attempt to address mem leak 2024-11-08 19:10:40 +11:00
Richard Holland
f8ec3f20dc restore the std::map mutex version of the rdb mem db, since concurrent might still have issues 2024-11-06 12:25:36 +11:00
Richard Holland
8b0d42785c define macro option to use std::map in nodestore... concurrent flatmap has a race condition on ledger cleaning cycle 2024-11-06 10:38:02 +11:00
Denis Angell
6b26045cbc Update settings.json (#342) 2024-10-25 11:56:16 +10:00
Wietse Wind
08f13b7cfe Fix account_tx sluggishness as per https://github.com/XRPLF/rippled/commit/2e9261cb (#308) 2024-10-25 11:13:42 +10:00
Richard Holland
006bd729d8 attempted crashbug fix 2024-10-25 12:02:45 +11:00
tequ
766f5d7ee1 Update macro.h (#366) 2024-10-25 10:10:43 +10:00
Wietse Wind
287c01ad04 Improve Admin command RPC Post (#384)
* Improve ADMIN HTTP POST RPC notifications: no queue limit, shorter HTTP call TTL
2024-10-25 10:10:14 +10:00
tequ
4239124750 Update amendments for rippled-standalone.cfg (#385) 2024-10-25 09:10:45 +10:00
Richard Holland
80efd28eba allow path= to be omitted in memdb config 2024-10-23 13:59:48 +11:00
Richard Holland
2bc5a8b1e8 disable online_delete for memdb, enforce ledger_history 2024-10-22 19:22:01 +11:00
Richard Holland
019d0ee527 clang 2024-10-21 18:11:56 +11:00
Richard Holland
c66857321c concurrent peerfinder 2024-10-21 18:04:39 +11:00
Richard Holland
c7c4fed461 peer finder memory store 2024-10-21 17:22:35 +11:00
Richard Holland
1639039fec concurrent flatmap 2024-10-21 16:18:08 +11:00
Richard Holland
23630dfac0 mem db implemented (not concurrent yet) 2024-10-21 14:56:40 +11:00
Richard Holland
26fba854e2 memory database not concurrent 2024-10-20 17:52:22 +11:00
Richard Holland
1dcef64626 use concurrent flatmap for memdb 2024-10-20 12:34:09 +11:00
35 changed files with 3779 additions and 131 deletions

View File

@@ -3,7 +3,7 @@
"C_Cpp.clang_format_path": ".clang-format",
"C_Cpp.clang_format_fallbackStyle": "{ ColumnLimit: 0 }",
"[cpp]":{
"editor.wordBasedSuggestions": false,
"editor.wordBasedSuggestions": "off",
"editor.suggest.insertMode": "replace",
"editor.semanticHighlighting.enabled": true,
"editor.tabSize": 4,

View File

@@ -31,6 +31,7 @@ target_sources (xrpl_core PRIVATE
src/ripple/beast/clock/basic_seconds_clock.cpp
src/ripple/beast/core/CurrentThreadName.cpp
src/ripple/beast/core/SemanticVersion.cpp
src/ripple/beast/hash/impl/xxhash.cpp
src/ripple/beast/insight/impl/Collector.cpp
src/ripple/beast/insight/impl/Groups.cpp
src/ripple/beast/insight/impl/Hook.cpp
@@ -148,8 +149,7 @@ target_link_libraries (xrpl_core
NIH::secp256k1
NIH::ed25519-donna
date::date
Ripple::opts
NIH::xxhash)
Ripple::opts)
#[=================================[
main/core headers installation
#]=================================]
@@ -283,6 +283,9 @@ install (
src/ripple/beast/hash/uhash.h
src/ripple/beast/hash/xxhasher.h
DESTINATION include/ripple/beast/hash)
install (
FILES src/ripple/beast/hash/impl/xxhash.h
DESTINATION include/ripple/beast/hash/impl)
install (
FILES
src/ripple/beast/rfc2616.h
@@ -535,7 +538,9 @@ target_sources (rippled PRIVATE
subdir: nodestore
#]===============================]
src/ripple/nodestore/backend/CassandraFactory.cpp
src/ripple/nodestore/backend/RWDBFactory.cpp
src/ripple/nodestore/backend/MemoryFactory.cpp
src/ripple/nodestore/backend/FlatmapFactory.cpp
src/ripple/nodestore/backend/NuDBFactory.cpp
src/ripple/nodestore/backend/NullFactory.cpp
src/ripple/nodestore/backend/RocksDBFactory.cpp

View File

@@ -1,51 +0,0 @@
#[===================================================================[
NIH dep: xxhash:
#]===================================================================]
# find_package(xxHash 0.8.2 CONFIG REQUIRED)
# target_link_libraries(ripple_libs INTERFACE xxHash::xxhash)
# add_library(NIH::xxhash ALIAS xxHash::xxhash)
include(FetchContent)
ExternalProject_Add(
xxhash_src
PREFIX ${nih_cache_path}
GIT_REPOSITORY https://github.com/Cyan4973/xxHash.git
GIT_TAG v0.8.2
SOURCE_SUBDIR cmake_unofficial
CMAKE_ARGS
-DBUILD_SHARED_LIBS=OFF
-DXXHASH_BUILD_XXHSUM=ON
LOG_BUILD ON
LOG_CONFIGURE ON
BUILD_COMMAND
${CMAKE_COMMAND}
--build .
--config $<CONFIG>
$<$<VERSION_GREATER_EQUAL:${CMAKE_VERSION},3.12>:--parallel ${ep_procs}>
INSTALL_COMMAND ""
BUILD_BYPRODUCTS
<BINARY_DIR>/libxxhash.a
<BINARY_DIR>/xxhsum
)
add_library(xxHash::xxhash STATIC IMPORTED GLOBAL)
ExternalProject_Get_Property (xxhash_src BINARY_DIR)
ExternalProject_Get_Property (xxhash_src SOURCE_DIR)
set (xxhash_src_SOURCE_DIR "${SOURCE_DIR}")
set (xxhash_src_BINARY_DIR "${BINARY_DIR}")
add_dependencies(xxHash::xxhash xxhash_src)
set_target_properties (xxHash::xxhash PROPERTIES
IMPORTED_LOCATION_DEBUG
"${xxhash_src_BINARY_DIR}/libxxhash.a"
IMPORTED_LOCATION_RELEASE
"${xxhash_src_BINARY_DIR}/libxxhash.a"
INTERFACE_INCLUDE_DIRECTORIES
"${xxhash_src_SOURCE_DIR}"
)
target_link_libraries(ripple_libs INTERFACE xxHash::xxhash)
add_library(NIH::xxhash ALIAS xxHash::xxhash)

View File

@@ -75,7 +75,6 @@ include(deps/gRPC)
include(deps/cassandra)
include(deps/Postgres)
include(deps/WasmEdge)
include(deps/xxHash)
###

View File

@@ -1056,7 +1056,18 @@
# Cassandra is an alternative backend to be used only with 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
#
@@ -1112,7 +1123,8 @@
# online_delete Minimum value of 256. Enable automatic purging
# of older ledger information. Maintain at least this
# 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
# relevant if online_delete is defined and non-zero:

View File

@@ -144,4 +144,12 @@ D686F2538F410C9D0D856788E98E3579595DAF7B38D38887F81ECAC934B06040 HooksUpdate1
86E83A7D2ECE3AD5FA87AB2195AE015C950469ABF0B72EAACED318F74886AE90 CryptoConditionsSuite
3C43D9A973AA4443EF3FC38E42DD306160FBFFDAB901CD8BAA15D09F2597EB87 NonFungibleTokensV1
0285B7E5E08E1A8E4C15636F0591D87F73CB6A7B6452A932AD72BBC8E5D1CBE3 fixNFTokenDirV1
36799EA497B1369B170805C078AEFE6188345F9B3E324C21E9CA3FF574E3C3D6 fixNFTokenNegOffer
36799EA497B1369B170805C078AEFE6188345F9B3E324C21E9CA3FF574E3C3D6 fixNFTokenNegOffer
4C499D17719BB365B69010A436B64FD1A82AAB199FC1CEB06962EBD01059FB09 fixXahauV1
215181D23BF5C173314B5FDB9C872C92DE6CC918483727DE037C0C13E7E6EE9D fixXahauV2
0D8BF22FF7570D58598D1EF19EBB6E142AD46E59A223FD3816262FBB69345BEA Remit
7CA0426E7F411D39BB014E57CD9E08F61DE1750F0D41FCD428D9FB80BB7596B0 ZeroB2M
4B8466415FAB32FFA89D9DCBE166A42340115771DF611A7160F8D7439C87ECD8 fixNSDelete
EDB4EE4C524E16BDD91D9A529332DED08DCAAA51CC6DC897ACFA1A0ED131C5B6 fix240819
8063140E9260799D6716756B891CEC3E7006C4E4F277AB84670663A88F94B9C4 fixPageCap
88693F108C3CD8A967F3F4253A32DEF5E35F9406ACD2A11B88B11D90865763A9 fix240911

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

@@ -1,5 +1,5 @@
/**
* These are helper macros for writing hooks, all of them are optional as is including hookmacro.h at all
* These are helper macros for writing hooks, all of them are optional as is including macro.h at all
*/
#include <stdint.h>

View File

@@ -60,7 +60,10 @@
#define sfBurnedNFTokens ((2U << 16U) + 44U)
#define sfHookStateCount ((2U << 16U) + 45U)
#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 sfRewardLgrFirst ((2U << 16U) + 99U)
#define sfRewardLgrLast ((2U << 16U) + 100U)
@@ -80,6 +83,8 @@
#define sfHookInstructionCount ((3U << 16U) + 17U)
#define sfHookReturnCode ((3U << 16U) + 18U)
#define sfReferenceCount ((3U << 16U) + 19U)
#define sfAccountIndex ((3U << 16U) + 98U)
#define sfAccountCount ((3U << 16U) + 99U)
#define sfRewardAccumulator ((3U << 16U) + 100U)
#define sfEmailHash ((4U << 16U) + 1U)
#define sfTakerPaysCurrency ((10U << 16U) + 1U)
@@ -120,6 +125,9 @@
#define sfOfferID ((5U << 16U) + 34U)
#define sfEscrowID ((5U << 16U) + 35U)
#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 sfBalance ((6U << 16U) + 2U)
#define sfLimitAmount ((6U << 16U) + 3U)
@@ -136,6 +144,9 @@
#define sfNFTokenBrokerFee ((6U << 16U) + 19U)
#define sfHookCallbackFee ((6U << 16U) + 20U)
#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 sfMessageKey ((7U << 16U) + 2U)
#define sfSigningPubKey ((7U << 16U) + 3U)
@@ -171,11 +182,13 @@
#define sfNFTokenMinter ((8U << 16U) + 9U)
#define sfEmitCallback ((8U << 16U) + 10U)
#define sfHookAccount ((8U << 16U) + 16U)
#define sfInform ((8U << 16U) + 99U)
#define sfIndexes ((19U << 16U) + 1U)
#define sfHashes ((19U << 16U) + 2U)
#define sfAmendments ((19U << 16U) + 3U)
#define sfNFTokenOffers ((19U << 16U) + 4U)
#define sfHookNamespaces ((19U << 16U) + 5U)
#define sfURITokenIDs ((19U << 16U) + 99U)
#define sfPaths ((18U << 16U) + 1U)
#define sfTransactionMetaData ((14U << 16U) + 2U)
#define sfCreatedNode ((14U << 16U) + 3U)
@@ -198,6 +211,12 @@
#define sfHookDefinition ((14U << 16U) + 22U)
#define sfHookParameter ((14U << 16U) + 23U)
#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 sfSignerEntries ((15U << 16U) + 4U)
#define sfTemplate ((15U << 16U) + 5U)
@@ -212,4 +231,8 @@
#define sfHookExecutions ((15U << 16U) + 18U)
#define sfHookParameters ((15U << 16U) + 19U)
#define sfHookGrants ((15U << 16U) + 20U)
#define sfGenesisMints ((15U << 16U) + 96U)
#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)
{
JLOG(j.trace()) << "HookInfo[" << HC_ACC()
<< "]: etxn_fee_base exception: " << e.what();
return INVALID_TXN;
}
@@ -5402,7 +5404,7 @@ DEFINE_HOOK_FUNCTION(
const int64_t float_one_internal = make_float(1000000000000000ull, -15, false);
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(float2);
@@ -5455,8 +5457,16 @@ float_divide_internal(int64_t float1, int64_t float2)
while (man2 > 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 += 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,
// 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();
}
@@ -5495,7 +5506,9 @@ DEFINE_HOOK_FUNCTION(int64_t, float_invert, int64_t float1)
return DIVISION_BY_ZERO;
if (float1 == 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();
}

View File

@@ -219,7 +219,7 @@ private:
run()
{
beast::setCurrentThreadName("LedgerCleaner");
JLOG(j_.debug()) << "Started";
JLOG(j_.debug()) << "Started ledger cleaner";
while (true)
{
@@ -392,7 +392,8 @@ private:
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));
continue;
}
@@ -415,13 +416,15 @@ private:
bool fail = false;
if (ledgerHash.isZero())
{
JLOG(j_.info())
<< "Unable to get hash for ledger " << ledgerIndex;
JLOG(j_.warn())
<< "Ledger Cleaner: Unable to get hash for ledger "
<< ledgerIndex;
fail = true;
}
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;
}

View File

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

View File

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

@@ -1132,7 +1132,7 @@ accountTxPage(
{
sql = boost::str(
boost::format(
prefix + (R"(AccountTransactions.LedgerSeq BETWEEN '%u' AND '%u'
prefix + (R"(AccountTransactions.LedgerSeq BETWEEN %u AND %u
ORDER BY AccountTransactions.LedgerSeq %s,
AccountTransactions.TxnSeq %s
LIMIT %u;)")) %
@@ -1155,12 +1155,14 @@ accountTxPage(
FROM AccountTransactions, Transactions WHERE
(AccountTransactions.TransID = Transactions.TransID AND
AccountTransactions.Account = '%s' AND
AccountTransactions.LedgerSeq BETWEEN '%u' AND '%u')
OR
AccountTransactions.LedgerSeq BETWEEN %u AND %u)
UNION
SELECT AccountTransactions.LedgerSeq,AccountTransactions.TxnSeq,Status,RawTxn,TxnMeta
FROM AccountTransactions, Transactions WHERE
(AccountTransactions.TransID = Transactions.TransID AND
AccountTransactions.Account = '%s' AND
AccountTransactions.LedgerSeq = '%u' AND
AccountTransactions.TxnSeq %s '%u')
AccountTransactions.LedgerSeq = %u AND
AccountTransactions.TxnSeq %s %u)
ORDER BY AccountTransactions.LedgerSeq %s,
AccountTransactions.TxnSeq %s
LIMIT %u;

View File

@@ -19,6 +19,8 @@
#include <ripple/app/main/Application.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/nodestore/DatabaseShard.h>
@@ -38,6 +40,8 @@ RelationalDatabase::init(
{
bool use_sqlite = false;
bool use_postgres = false;
bool use_rwdb = false;
bool use_flatmap = false;
if (config.reporting())
{
@@ -52,6 +56,14 @@ RelationalDatabase::init(
{
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
{
Throw<std::runtime_error>(
@@ -73,6 +85,14 @@ RelationalDatabase::init(
{
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>();
}

View File

@@ -0,0 +1,996 @@
/*
xxHash - Fast Hash algorithm
Copyright (C) 2012-2014, Yann Collet.
BSD 2-Clause License (http://www.opensource.org/licenses/bsd-license.php)
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following disclaimer
in the documentation and/or other materials provided with the
distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
You can contact the author at :
- xxHash source repository : http://code.google.com/p/xxhash/
- public discussion board : https://groups.google.com/forum/#!forum/lz4c
*/
#include <ripple/beast/hash/impl/xxhash.h>
//**************************************
// Tuning parameters
//**************************************
// Unaligned memory access is automatically enabled for "common" CPU, such as
// x86. For others CPU, the compiler will be more cautious, and insert extra
// code to ensure aligned access is respected. If you know your target CPU
// supports unaligned memory access, you want to force this option manually to
// improve performance. You can also enable this parameter if you know your
// input data will always be aligned (boundaries of 4, for U32).
#if defined(__ARM_FEATURE_UNALIGNED) || defined(__i386) || defined(_M_IX86) || \
defined(__x86_64__) || defined(_M_X64)
#define XXH_USE_UNALIGNED_ACCESS 1
#endif
// XXH_ACCEPT_NULL_INPUT_POINTER :
// If the input pointer is a null pointer, xxHash default behavior is to trigger
// a memory access error, since it is a bad pointer. When this option is
// enabled, xxHash output for null input pointers will be the same as a
// null-length input. This option has a very small performance cost (only
// measurable on small inputs). By default, this option is disabled. To enable
// it, uncomment below define : #define XXH_ACCEPT_NULL_INPUT_POINTER 1
// XXH_FORCE_NATIVE_FORMAT :
// By default, xxHash library provides endian-independant Hash values, based on
// little-endian convention. Results are therefore identical for little-endian
// and big-endian CPU. This comes at a performance cost for big-endian CPU,
// since some swapping is required to emulate little-endian format. Should
// endian-independance be of no importance for your application, you may set the
// #define below to 1. It will improve speed for Big-endian CPU. This option has
// no impact on Little_Endian CPU.
#define XXH_FORCE_NATIVE_FORMAT 0
//**************************************
// Compiler Specific Options
//**************************************
// Disable some Visual warning messages
#ifdef _MSC_VER // Visual Studio
#pragma warning( \
disable : 4127) // disable: C4127: conditional expression is constant
#endif
#ifdef _MSC_VER // Visual Studio
#define FORCE_INLINE static __forceinline
#else
#ifdef __GNUC__
#define FORCE_INLINE static inline __attribute__((always_inline))
#else
#define FORCE_INLINE static inline
#endif
#endif
//**************************************
// Includes & Memory related functions
//**************************************
//#include "xxhash.h"
// Modify the local functions below should you wish to use some other memory
// routines for malloc(), free()
#include <stdlib.h>
static void*
XXH_malloc(size_t s)
{
return malloc(s);
}
static void
XXH_free(void* p)
{
free(p);
}
// for memcpy()
#include <string.h>
static void*
XXH_memcpy(void* dest, const void* src, size_t size)
{
return memcpy(dest, src, size);
}
//**************************************
// Basic Types
//**************************************
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 199901L // C99
#include <stdint.h>
typedef uint8_t BYTE;
typedef uint16_t U16;
typedef uint32_t U32;
typedef int32_t S32;
typedef uint64_t U64;
#else
typedef unsigned char BYTE;
typedef unsigned short U16;
typedef unsigned int U32;
typedef signed int S32;
typedef unsigned long long U64;
#endif
#if defined(__GNUC__) && !defined(XXH_USE_UNALIGNED_ACCESS)
#define _PACKED __attribute__((packed))
#else
#define _PACKED
#endif
#if !defined(XXH_USE_UNALIGNED_ACCESS) && !defined(__GNUC__)
#ifdef __IBMC__
#pragma pack(1)
#else
#pragma pack(push, 1)
#endif
#endif
namespace beast {
namespace detail {
typedef struct _U32_S
{
U32 v;
} _PACKED U32_S;
typedef struct _U64_S
{
U64 v;
} _PACKED U64_S;
#if !defined(XXH_USE_UNALIGNED_ACCESS) && !defined(__GNUC__)
#pragma pack(pop)
#endif
#define A32(x) (((U32_S*)(x))->v)
#define A64(x) (((U64_S*)(x))->v)
//***************************************
// Compiler-specific Functions and Macros
//***************************************
#define GCC_VERSION (__GNUC__ * 100 + __GNUC_MINOR__)
// Note : although _rotl exists for minGW (GCC under windows), performance seems
// poor
#if defined(_MSC_VER)
#define XXH_rotl32(x, r) _rotl(x, r)
#define XXH_rotl64(x, r) _rotl64(x, r)
#else
#define XXH_rotl32(x, r) ((x << r) | (x >> (32 - r)))
#define XXH_rotl64(x, r) ((x << r) | (x >> (64 - r)))
#endif
#if defined(_MSC_VER) // Visual Studio
#define XXH_swap32 _byteswap_ulong
#define XXH_swap64 _byteswap_uint64
#elif GCC_VERSION >= 403
#define XXH_swap32 __builtin_bswap32
#define XXH_swap64 __builtin_bswap64
#else
static inline U32
XXH_swap32(U32 x)
{
return ((x << 24) & 0xff000000) | ((x << 8) & 0x00ff0000) |
((x >> 8) & 0x0000ff00) | ((x >> 24) & 0x000000ff);
}
static inline U64
XXH_swap64(U64 x)
{
return ((x << 56) & 0xff00000000000000ULL) |
((x << 40) & 0x00ff000000000000ULL) |
((x << 24) & 0x0000ff0000000000ULL) |
((x << 8) & 0x000000ff00000000ULL) |
((x >> 8) & 0x00000000ff000000ULL) |
((x >> 24) & 0x0000000000ff0000ULL) |
((x >> 40) & 0x000000000000ff00ULL) |
((x >> 56) & 0x00000000000000ffULL);
}
#endif
//**************************************
// Constants
//**************************************
#define PRIME32_1 2654435761U
#define PRIME32_2 2246822519U
#define PRIME32_3 3266489917U
#define PRIME32_4 668265263U
#define PRIME32_5 374761393U
#define PRIME64_1 11400714785074694791ULL
#define PRIME64_2 14029467366897019727ULL
#define PRIME64_3 1609587929392839161ULL
#define PRIME64_4 9650029242287828579ULL
#define PRIME64_5 2870177450012600261ULL
//**************************************
// Architecture Macros
//**************************************
typedef enum { XXH_bigEndian = 0, XXH_littleEndian = 1 } XXH_endianess;
#ifndef XXH_CPU_LITTLE_ENDIAN // It is possible to define XXH_CPU_LITTLE_ENDIAN
// externally, for example using a compiler
// switch
static const int one = 1;
#define XXH_CPU_LITTLE_ENDIAN (*(char*)(&one))
#endif
//**************************************
// Macros
//**************************************
#define XXH_STATIC_ASSERT(c) \
{ \
enum { XXH_static_assert = 1 / (!!(c)) }; \
} // use only *after* variable declarations
//****************************
// Memory reads
//****************************
typedef enum { XXH_aligned, XXH_unaligned } XXH_alignment;
FORCE_INLINE U32
XXH_readLE32_align(const void* ptr, XXH_endianess endian, XXH_alignment align)
{
if (align == XXH_unaligned)
return endian == XXH_littleEndian ? A32(ptr) : XXH_swap32(A32(ptr));
else
return endian == XXH_littleEndian ? *(U32*)ptr : XXH_swap32(*(U32*)ptr);
}
FORCE_INLINE U32
XXH_readLE32(const void* ptr, XXH_endianess endian)
{
return XXH_readLE32_align(ptr, endian, XXH_unaligned);
}
FORCE_INLINE U64
XXH_readLE64_align(const void* ptr, XXH_endianess endian, XXH_alignment align)
{
if (align == XXH_unaligned)
return endian == XXH_littleEndian ? A64(ptr) : XXH_swap64(A64(ptr));
else
return endian == XXH_littleEndian ? *(U64*)ptr : XXH_swap64(*(U64*)ptr);
}
FORCE_INLINE U64
XXH_readLE64(const void* ptr, XXH_endianess endian)
{
return XXH_readLE64_align(ptr, endian, XXH_unaligned);
}
//****************************
// Simple Hash Functions
//****************************
FORCE_INLINE U32
XXH32_endian_align(
const void* input,
size_t len,
U32 seed,
XXH_endianess endian,
XXH_alignment align)
{
const BYTE* p = (const BYTE*)input;
const BYTE* bEnd = p + len;
U32 h32;
#define XXH_get32bits(p) XXH_readLE32_align(p, endian, align)
#ifdef XXH_ACCEPT_NULL_INPUT_POINTER
if (p == NULL)
{
len = 0;
bEnd = p = (const BYTE*)(size_t)16;
}
#endif
if (len >= 16)
{
const BYTE* const limit = bEnd - 16;
U32 v1 = seed + PRIME32_1 + PRIME32_2;
U32 v2 = seed + PRIME32_2;
U32 v3 = seed + 0;
U32 v4 = seed - PRIME32_1;
do
{
v1 += XXH_get32bits(p) * PRIME32_2;
v1 = XXH_rotl32(v1, 13);
v1 *= PRIME32_1;
p += 4;
v2 += XXH_get32bits(p) * PRIME32_2;
v2 = XXH_rotl32(v2, 13);
v2 *= PRIME32_1;
p += 4;
v3 += XXH_get32bits(p) * PRIME32_2;
v3 = XXH_rotl32(v3, 13);
v3 *= PRIME32_1;
p += 4;
v4 += XXH_get32bits(p) * PRIME32_2;
v4 = XXH_rotl32(v4, 13);
v4 *= PRIME32_1;
p += 4;
} while (p <= limit);
h32 = XXH_rotl32(v1, 1) + XXH_rotl32(v2, 7) + XXH_rotl32(v3, 12) +
XXH_rotl32(v4, 18);
}
else
{
h32 = seed + PRIME32_5;
}
h32 += (U32)len;
while (p + 4 <= bEnd)
{
h32 += XXH_get32bits(p) * PRIME32_3;
h32 = XXH_rotl32(h32, 17) * PRIME32_4;
p += 4;
}
while (p < bEnd)
{
h32 += (*p) * PRIME32_5;
h32 = XXH_rotl32(h32, 11) * PRIME32_1;
p++;
}
h32 ^= h32 >> 15;
h32 *= PRIME32_2;
h32 ^= h32 >> 13;
h32 *= PRIME32_3;
h32 ^= h32 >> 16;
return h32;
}
unsigned int
XXH32(const void* input, size_t len, unsigned seed)
{
#if 0
// Simple version, good for code maintenance, but unfortunately slow for small inputs
XXH32_state_t state;
XXH32_reset(&state, seed);
XXH32_update(&state, input, len);
return XXH32_digest(&state);
#else
XXH_endianess endian_detected = (XXH_endianess)XXH_CPU_LITTLE_ENDIAN;
#if !defined(XXH_USE_UNALIGNED_ACCESS)
if ((((size_t)input) & 3) ==
0) // Input is aligned, let's leverage the speed advantage
{
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH32_endian_align(
input, len, seed, XXH_littleEndian, XXH_aligned);
else
return XXH32_endian_align(
input, len, seed, XXH_bigEndian, XXH_aligned);
}
#endif
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH32_endian_align(
input, len, seed, XXH_littleEndian, XXH_unaligned);
else
return XXH32_endian_align(
input, len, seed, XXH_bigEndian, XXH_unaligned);
#endif
}
FORCE_INLINE U64
XXH64_endian_align(
const void* input,
size_t len,
U64 seed,
XXH_endianess endian,
XXH_alignment align)
{
const BYTE* p = (const BYTE*)input;
const BYTE* bEnd = p + len;
U64 h64;
#define XXH_get64bits(p) XXH_readLE64_align(p, endian, align)
#ifdef XXH_ACCEPT_NULL_INPUT_POINTER
if (p == NULL)
{
len = 0;
bEnd = p = (const BYTE*)(size_t)32;
}
#endif
if (len >= 32)
{
const BYTE* const limit = bEnd - 32;
U64 v1 = seed + PRIME64_1 + PRIME64_2;
U64 v2 = seed + PRIME64_2;
U64 v3 = seed + 0;
U64 v4 = seed - PRIME64_1;
do
{
v1 += XXH_get64bits(p) * PRIME64_2;
p += 8;
v1 = XXH_rotl64(v1, 31);
v1 *= PRIME64_1;
v2 += XXH_get64bits(p) * PRIME64_2;
p += 8;
v2 = XXH_rotl64(v2, 31);
v2 *= PRIME64_1;
v3 += XXH_get64bits(p) * PRIME64_2;
p += 8;
v3 = XXH_rotl64(v3, 31);
v3 *= PRIME64_1;
v4 += XXH_get64bits(p) * PRIME64_2;
p += 8;
v4 = XXH_rotl64(v4, 31);
v4 *= PRIME64_1;
} while (p <= limit);
h64 = XXH_rotl64(v1, 1) + XXH_rotl64(v2, 7) + XXH_rotl64(v3, 12) +
XXH_rotl64(v4, 18);
v1 *= PRIME64_2;
v1 = XXH_rotl64(v1, 31);
v1 *= PRIME64_1;
h64 ^= v1;
h64 = h64 * PRIME64_1 + PRIME64_4;
v2 *= PRIME64_2;
v2 = XXH_rotl64(v2, 31);
v2 *= PRIME64_1;
h64 ^= v2;
h64 = h64 * PRIME64_1 + PRIME64_4;
v3 *= PRIME64_2;
v3 = XXH_rotl64(v3, 31);
v3 *= PRIME64_1;
h64 ^= v3;
h64 = h64 * PRIME64_1 + PRIME64_4;
v4 *= PRIME64_2;
v4 = XXH_rotl64(v4, 31);
v4 *= PRIME64_1;
h64 ^= v4;
h64 = h64 * PRIME64_1 + PRIME64_4;
}
else
{
h64 = seed + PRIME64_5;
}
h64 += (U64)len;
while (p + 8 <= bEnd)
{
U64 k1 = XXH_get64bits(p);
k1 *= PRIME64_2;
k1 = XXH_rotl64(k1, 31);
k1 *= PRIME64_1;
h64 ^= k1;
h64 = XXH_rotl64(h64, 27) * PRIME64_1 + PRIME64_4;
p += 8;
}
if (p + 4 <= bEnd)
{
h64 ^= (U64)(XXH_get32bits(p)) * PRIME64_1;
h64 = XXH_rotl64(h64, 23) * PRIME64_2 + PRIME64_3;
p += 4;
}
while (p < bEnd)
{
h64 ^= (*p) * PRIME64_5;
h64 = XXH_rotl64(h64, 11) * PRIME64_1;
p++;
}
h64 ^= h64 >> 33;
h64 *= PRIME64_2;
h64 ^= h64 >> 29;
h64 *= PRIME64_3;
h64 ^= h64 >> 32;
return h64;
}
unsigned long long
XXH64(const void* input, size_t len, unsigned long long seed)
{
#if 0
// Simple version, good for code maintenance, but unfortunately slow for small inputs
XXH64_state_t state;
XXH64_reset(&state, seed);
XXH64_update(&state, input, len);
return XXH64_digest(&state);
#else
XXH_endianess endian_detected = (XXH_endianess)XXH_CPU_LITTLE_ENDIAN;
#if !defined(XXH_USE_UNALIGNED_ACCESS)
if ((((size_t)input) & 7) ==
0) // Input is aligned, let's leverage the speed advantage
{
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH64_endian_align(
input, len, seed, XXH_littleEndian, XXH_aligned);
else
return XXH64_endian_align(
input, len, seed, XXH_bigEndian, XXH_aligned);
}
#endif
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH64_endian_align(
input, len, seed, XXH_littleEndian, XXH_unaligned);
else
return XXH64_endian_align(
input, len, seed, XXH_bigEndian, XXH_unaligned);
#endif
}
/****************************************************
* Advanced Hash Functions
****************************************************/
/*** Allocation ***/
typedef struct
{
U64 total_len;
U32 seed;
U32 v1;
U32 v2;
U32 v3;
U32 v4;
U32 mem32[4]; /* defined as U32 for alignment */
U32 memsize;
} XXH_istate32_t;
typedef struct
{
U64 total_len;
U64 seed;
U64 v1;
U64 v2;
U64 v3;
U64 v4;
U64 mem64[4]; /* defined as U64 for alignment */
U32 memsize;
} XXH_istate64_t;
XXH32_state_t*
XXH32_createState(void)
{
static_assert(
sizeof(XXH32_state_t) >= sizeof(XXH_istate32_t),
""); // A compilation error here means XXH32_state_t is not large
// enough
return (XXH32_state_t*)XXH_malloc(sizeof(XXH32_state_t));
}
XXH_errorcode
XXH32_freeState(XXH32_state_t* statePtr)
{
XXH_free(statePtr);
return XXH_OK;
};
XXH64_state_t*
XXH64_createState(void)
{
static_assert(
sizeof(XXH64_state_t) >= sizeof(XXH_istate64_t),
""); // A compilation error here means XXH64_state_t is not large
// enough
return (XXH64_state_t*)XXH_malloc(sizeof(XXH64_state_t));
}
XXH_errorcode
XXH64_freeState(XXH64_state_t* statePtr)
{
XXH_free(statePtr);
return XXH_OK;
};
/*** Hash feed ***/
XXH_errorcode
XXH32_reset(XXH32_state_t* state_in, U32 seed)
{
XXH_istate32_t* state = (XXH_istate32_t*)state_in;
state->seed = seed;
state->v1 = seed + PRIME32_1 + PRIME32_2;
state->v2 = seed + PRIME32_2;
state->v3 = seed + 0;
state->v4 = seed - PRIME32_1;
state->total_len = 0;
state->memsize = 0;
return XXH_OK;
}
XXH_errorcode
XXH64_reset(XXH64_state_t* state_in, unsigned long long seed)
{
XXH_istate64_t* state = (XXH_istate64_t*)state_in;
state->seed = seed;
state->v1 = seed + PRIME64_1 + PRIME64_2;
state->v2 = seed + PRIME64_2;
state->v3 = seed + 0;
state->v4 = seed - PRIME64_1;
state->total_len = 0;
state->memsize = 0;
return XXH_OK;
}
FORCE_INLINE XXH_errorcode
XXH32_update_endian(
XXH32_state_t* state_in,
const void* input,
size_t len,
XXH_endianess endian)
{
XXH_istate32_t* state = (XXH_istate32_t*)state_in;
const BYTE* p = (const BYTE*)input;
const BYTE* const bEnd = p + len;
#ifdef XXH_ACCEPT_NULL_INPUT_POINTER
if (input == NULL)
return XXH_ERROR;
#endif
state->total_len += len;
if (state->memsize + len < 16) // fill in tmp buffer
{
XXH_memcpy((BYTE*)(state->mem32) + state->memsize, input, len);
state->memsize += (U32)len;
return XXH_OK;
}
if (state->memsize) // some data left from previous update
{
XXH_memcpy(
(BYTE*)(state->mem32) + state->memsize, input, 16 - state->memsize);
{
const U32* p32 = state->mem32;
state->v1 += XXH_readLE32(p32, endian) * PRIME32_2;
state->v1 = XXH_rotl32(state->v1, 13);
state->v1 *= PRIME32_1;
p32++;
state->v2 += XXH_readLE32(p32, endian) * PRIME32_2;
state->v2 = XXH_rotl32(state->v2, 13);
state->v2 *= PRIME32_1;
p32++;
state->v3 += XXH_readLE32(p32, endian) * PRIME32_2;
state->v3 = XXH_rotl32(state->v3, 13);
state->v3 *= PRIME32_1;
p32++;
state->v4 += XXH_readLE32(p32, endian) * PRIME32_2;
state->v4 = XXH_rotl32(state->v4, 13);
state->v4 *= PRIME32_1;
p32++;
}
p += 16 - state->memsize;
state->memsize = 0;
}
if (p <= bEnd - 16)
{
const BYTE* const limit = bEnd - 16;
U32 v1 = state->v1;
U32 v2 = state->v2;
U32 v3 = state->v3;
U32 v4 = state->v4;
do
{
v1 += XXH_readLE32(p, endian) * PRIME32_2;
v1 = XXH_rotl32(v1, 13);
v1 *= PRIME32_1;
p += 4;
v2 += XXH_readLE32(p, endian) * PRIME32_2;
v2 = XXH_rotl32(v2, 13);
v2 *= PRIME32_1;
p += 4;
v3 += XXH_readLE32(p, endian) * PRIME32_2;
v3 = XXH_rotl32(v3, 13);
v3 *= PRIME32_1;
p += 4;
v4 += XXH_readLE32(p, endian) * PRIME32_2;
v4 = XXH_rotl32(v4, 13);
v4 *= PRIME32_1;
p += 4;
} while (p <= limit);
state->v1 = v1;
state->v2 = v2;
state->v3 = v3;
state->v4 = v4;
}
if (p < bEnd)
{
XXH_memcpy(state->mem32, p, bEnd - p);
state->memsize = (int)(bEnd - p);
}
return XXH_OK;
}
XXH_errorcode
XXH32_update(XXH32_state_t* state_in, const void* input, size_t len)
{
XXH_endianess endian_detected = (XXH_endianess)XXH_CPU_LITTLE_ENDIAN;
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH32_update_endian(state_in, input, len, XXH_littleEndian);
else
return XXH32_update_endian(state_in, input, len, XXH_bigEndian);
}
FORCE_INLINE U32
XXH32_digest_endian(const XXH32_state_t* state_in, XXH_endianess endian)
{
XXH_istate32_t* state = (XXH_istate32_t*)state_in;
const BYTE* p = (const BYTE*)state->mem32;
BYTE* bEnd = (BYTE*)(state->mem32) + state->memsize;
U32 h32;
if (state->total_len >= 16)
{
h32 = XXH_rotl32(state->v1, 1) + XXH_rotl32(state->v2, 7) +
XXH_rotl32(state->v3, 12) + XXH_rotl32(state->v4, 18);
}
else
{
h32 = state->seed + PRIME32_5;
}
h32 += (U32)state->total_len;
while (p + 4 <= bEnd)
{
h32 += XXH_readLE32(p, endian) * PRIME32_3;
h32 = XXH_rotl32(h32, 17) * PRIME32_4;
p += 4;
}
while (p < bEnd)
{
h32 += (*p) * PRIME32_5;
h32 = XXH_rotl32(h32, 11) * PRIME32_1;
p++;
}
h32 ^= h32 >> 15;
h32 *= PRIME32_2;
h32 ^= h32 >> 13;
h32 *= PRIME32_3;
h32 ^= h32 >> 16;
return h32;
}
U32
XXH32_digest(const XXH32_state_t* state_in)
{
XXH_endianess endian_detected = (XXH_endianess)XXH_CPU_LITTLE_ENDIAN;
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH32_digest_endian(state_in, XXH_littleEndian);
else
return XXH32_digest_endian(state_in, XXH_bigEndian);
}
FORCE_INLINE XXH_errorcode
XXH64_update_endian(
XXH64_state_t* state_in,
const void* input,
size_t len,
XXH_endianess endian)
{
XXH_istate64_t* state = (XXH_istate64_t*)state_in;
const BYTE* p = (const BYTE*)input;
const BYTE* const bEnd = p + len;
#ifdef XXH_ACCEPT_NULL_INPUT_POINTER
if (input == NULL)
return XXH_ERROR;
#endif
state->total_len += len;
if (state->memsize + len < 32) // fill in tmp buffer
{
XXH_memcpy(((BYTE*)state->mem64) + state->memsize, input, len);
state->memsize += (U32)len;
return XXH_OK;
}
if (state->memsize) // some data left from previous update
{
XXH_memcpy(
((BYTE*)state->mem64) + state->memsize, input, 32 - state->memsize);
{
const U64* p64 = state->mem64;
state->v1 += XXH_readLE64(p64, endian) * PRIME64_2;
state->v1 = XXH_rotl64(state->v1, 31);
state->v1 *= PRIME64_1;
p64++;
state->v2 += XXH_readLE64(p64, endian) * PRIME64_2;
state->v2 = XXH_rotl64(state->v2, 31);
state->v2 *= PRIME64_1;
p64++;
state->v3 += XXH_readLE64(p64, endian) * PRIME64_2;
state->v3 = XXH_rotl64(state->v3, 31);
state->v3 *= PRIME64_1;
p64++;
state->v4 += XXH_readLE64(p64, endian) * PRIME64_2;
state->v4 = XXH_rotl64(state->v4, 31);
state->v4 *= PRIME64_1;
p64++;
}
p += 32 - state->memsize;
state->memsize = 0;
}
if (p + 32 <= bEnd)
{
const BYTE* const limit = bEnd - 32;
U64 v1 = state->v1;
U64 v2 = state->v2;
U64 v3 = state->v3;
U64 v4 = state->v4;
do
{
v1 += XXH_readLE64(p, endian) * PRIME64_2;
v1 = XXH_rotl64(v1, 31);
v1 *= PRIME64_1;
p += 8;
v2 += XXH_readLE64(p, endian) * PRIME64_2;
v2 = XXH_rotl64(v2, 31);
v2 *= PRIME64_1;
p += 8;
v3 += XXH_readLE64(p, endian) * PRIME64_2;
v3 = XXH_rotl64(v3, 31);
v3 *= PRIME64_1;
p += 8;
v4 += XXH_readLE64(p, endian) * PRIME64_2;
v4 = XXH_rotl64(v4, 31);
v4 *= PRIME64_1;
p += 8;
} while (p <= limit);
state->v1 = v1;
state->v2 = v2;
state->v3 = v3;
state->v4 = v4;
}
if (p < bEnd)
{
XXH_memcpy(state->mem64, p, bEnd - p);
state->memsize = (int)(bEnd - p);
}
return XXH_OK;
}
XXH_errorcode
XXH64_update(XXH64_state_t* state_in, const void* input, size_t len)
{
XXH_endianess endian_detected = (XXH_endianess)XXH_CPU_LITTLE_ENDIAN;
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH64_update_endian(state_in, input, len, XXH_littleEndian);
else
return XXH64_update_endian(state_in, input, len, XXH_bigEndian);
}
FORCE_INLINE U64
XXH64_digest_endian(const XXH64_state_t* state_in, XXH_endianess endian)
{
XXH_istate64_t* state = (XXH_istate64_t*)state_in;
const BYTE* p = (const BYTE*)state->mem64;
BYTE* bEnd = (BYTE*)state->mem64 + state->memsize;
U64 h64;
if (state->total_len >= 32)
{
U64 v1 = state->v1;
U64 v2 = state->v2;
U64 v3 = state->v3;
U64 v4 = state->v4;
h64 = XXH_rotl64(v1, 1) + XXH_rotl64(v2, 7) + XXH_rotl64(v3, 12) +
XXH_rotl64(v4, 18);
v1 *= PRIME64_2;
v1 = XXH_rotl64(v1, 31);
v1 *= PRIME64_1;
h64 ^= v1;
h64 = h64 * PRIME64_1 + PRIME64_4;
v2 *= PRIME64_2;
v2 = XXH_rotl64(v2, 31);
v2 *= PRIME64_1;
h64 ^= v2;
h64 = h64 * PRIME64_1 + PRIME64_4;
v3 *= PRIME64_2;
v3 = XXH_rotl64(v3, 31);
v3 *= PRIME64_1;
h64 ^= v3;
h64 = h64 * PRIME64_1 + PRIME64_4;
v4 *= PRIME64_2;
v4 = XXH_rotl64(v4, 31);
v4 *= PRIME64_1;
h64 ^= v4;
h64 = h64 * PRIME64_1 + PRIME64_4;
}
else
{
h64 = state->seed + PRIME64_5;
}
h64 += (U64)state->total_len;
while (p + 8 <= bEnd)
{
U64 k1 = XXH_readLE64(p, endian);
k1 *= PRIME64_2;
k1 = XXH_rotl64(k1, 31);
k1 *= PRIME64_1;
h64 ^= k1;
h64 = XXH_rotl64(h64, 27) * PRIME64_1 + PRIME64_4;
p += 8;
}
if (p + 4 <= bEnd)
{
h64 ^= (U64)(XXH_readLE32(p, endian)) * PRIME64_1;
h64 = XXH_rotl64(h64, 23) * PRIME64_2 + PRIME64_3;
p += 4;
}
while (p < bEnd)
{
h64 ^= (*p) * PRIME64_5;
h64 = XXH_rotl64(h64, 11) * PRIME64_1;
p++;
}
h64 ^= h64 >> 33;
h64 *= PRIME64_2;
h64 ^= h64 >> 29;
h64 *= PRIME64_3;
h64 ^= h64 >> 32;
return h64;
}
unsigned long long
XXH64_digest(const XXH64_state_t* state_in)
{
XXH_endianess endian_detected = (XXH_endianess)XXH_CPU_LITTLE_ENDIAN;
if ((endian_detected == XXH_littleEndian) || XXH_FORCE_NATIVE_FORMAT)
return XXH64_digest_endian(state_in, XXH_littleEndian);
else
return XXH64_digest_endian(state_in, XXH_bigEndian);
}
} // namespace detail
} // namespace beast

View File

@@ -0,0 +1,170 @@
/*
xxHash - Extremely Fast Hash algorithm
Header File
Copyright (C) 2012-2014, Yann Collet.
BSD 2-Clause License (http://www.opensource.org/licenses/bsd-license.php)
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following disclaimer
in the documentation and/or other materials provided with the
distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
You can contact the author at :
- xxHash source repository : http://code.google.com/p/xxhash/
*/
/* Notice extracted from xxHash homepage :
xxHash is an extremely fast Hash algorithm, running at RAM speed limits.
It also successfully passes all tests from the SMHasher suite.
Comparison (single thread, Windows Seven 32 bits, using SMHasher on a Core 2 Duo
@3GHz)
Name Speed Q.Score Author
xxHash 5.4 GB/s 10
CrapWow 3.2 GB/s 2 Andrew
MumurHash 3a 2.7 GB/s 10 Austin Appleby
SpookyHash 2.0 GB/s 10 Bob Jenkins
SBox 1.4 GB/s 9 Bret Mulvey
Lookup3 1.2 GB/s 9 Bob Jenkins
SuperFastHash 1.2 GB/s 1 Paul Hsieh
CityHash64 1.05 GB/s 10 Pike & Alakuijala
FNV 0.55 GB/s 5 Fowler, Noll, Vo
CRC32 0.43 GB/s 9
MD5-32 0.33 GB/s 10 Ronald L. Rivest
SHA1-32 0.28 GB/s 10
Q.Score is a measure of quality of the hash function.
It depends on successfully passing SMHasher test set.
10 is a perfect score.
*/
#ifndef BEAST_HASH_XXHASH_H_INCLUDED
#define BEAST_HASH_XXHASH_H_INCLUDED
/*****************************
Includes
*****************************/
#include <stddef.h> /* size_t */
namespace beast {
namespace detail {
/*****************************
Type
*****************************/
typedef enum { XXH_OK = 0, XXH_ERROR } XXH_errorcode;
/*****************************
Simple Hash Functions
*****************************/
unsigned int
XXH32(const void* input, size_t length, unsigned seed);
unsigned long long
XXH64(const void* input, size_t length, unsigned long long seed);
/*
XXH32() :
Calculate the 32-bits hash of sequence "length" bytes stored at memory
address "input". The memory between input & input+length must be valid
(allocated and read-accessible). "seed" can be used to alter the result
predictably. This function successfully passes all SMHasher tests. Speed on Core
2 Duo @ 3 GHz (single thread, SMHasher benchmark) : 5.4 GB/s XXH64() : Calculate
the 64-bits hash of sequence of length "len" stored at memory address "input".
*/
/*****************************
Advanced Hash Functions
*****************************/
typedef struct
{
long long ll[6];
} XXH32_state_t;
typedef struct
{
long long ll[11];
} XXH64_state_t;
/*
These structures allow static allocation of XXH states.
States must then be initialized using XXHnn_reset() before first use.
If you prefer dynamic allocation, please refer to functions below.
*/
XXH32_state_t*
XXH32_createState(void);
XXH_errorcode
XXH32_freeState(XXH32_state_t* statePtr);
XXH64_state_t*
XXH64_createState(void);
XXH_errorcode
XXH64_freeState(XXH64_state_t* statePtr);
/*
These functions create and release memory for XXH state.
States must then be initialized using XXHnn_reset() before first use.
*/
XXH_errorcode
XXH32_reset(XXH32_state_t* statePtr, unsigned seed);
XXH_errorcode
XXH32_update(XXH32_state_t* statePtr, const void* input, size_t length);
unsigned int
XXH32_digest(const XXH32_state_t* statePtr);
XXH_errorcode
XXH64_reset(XXH64_state_t* statePtr, unsigned long long seed);
XXH_errorcode
XXH64_update(XXH64_state_t* statePtr, const void* input, size_t length);
unsigned long long
XXH64_digest(const XXH64_state_t* statePtr);
/*
These functions calculate the xxHash of an input provided in multiple smaller
packets, as opposed to an input provided as a single block.
XXH state space must first be allocated, using either static or dynamic method
provided above.
Start a new hash by initializing state with a seed, using XXHnn_reset().
Then, feed the hash state by calling XXHnn_update() as many times as necessary.
Obviously, input must be valid, meaning allocated and read accessible.
The function returns an error code, with 0 meaning OK, and any other value
meaning there is an error.
Finally, you can produce a hash anytime, by using XXHnn_digest().
This function returns the final nn-bits hash.
You can nonetheless continue feeding the hash state with more input,
and therefore get some new hashes, by calling again XXHnn_digest().
When you are done, don't forget to free XXH state space, using typically
XXHnn_freeState().
*/
} // namespace detail
} // namespace beast
#endif

View File

@@ -20,11 +20,9 @@
#ifndef BEAST_HASH_XXHASHER_H_INCLUDED
#define BEAST_HASH_XXHASHER_H_INCLUDED
#include <ripple/beast/hash/impl/xxhash.h>
#include <boost/endian/conversion.hpp>
#include <xxhash.h>
#include <cstddef>
#include <new>
#include <type_traits>
namespace beast {
@@ -35,35 +33,16 @@ private:
// requires 64-bit std::size_t
static_assert(sizeof(std::size_t) == 8, "");
XXH3_state_t* state_;
static XXH3_state_t*
allocState()
{
auto ret = XXH3_createState();
if (ret == nullptr)
throw std::bad_alloc();
return ret;
}
detail::XXH64_state_t state_;
public:
using result_type = std::size_t;
static constexpr auto const endian = boost::endian::order::native;
xxhasher(xxhasher const&) = delete;
xxhasher&
operator=(xxhasher const&) = delete;
xxhasher()
xxhasher() noexcept
{
state_ = allocState();
XXH3_64bits_reset(state_);
}
~xxhasher() noexcept
{
XXH3_freeState(state_);
detail::XXH64_reset(&state_, 1);
}
template <
@@ -71,8 +50,7 @@ public:
std::enable_if_t<std::is_unsigned<Seed>::value>* = nullptr>
explicit xxhasher(Seed seed)
{
state_ = allocState();
XXH3_64bits_reset_withSeed(state_, seed);
detail::XXH64_reset(&state_, seed);
}
template <
@@ -80,19 +58,18 @@ public:
std::enable_if_t<std::is_unsigned<Seed>::value>* = nullptr>
xxhasher(Seed seed, Seed)
{
state_ = allocState();
XXH3_64bits_reset_withSeed(state_, seed);
detail::XXH64_reset(&state_, seed);
}
void
operator()(void const* key, std::size_t len) noexcept
{
XXH3_64bits_update(state_, key, len);
detail::XXH64_update(&state_, key, len);
}
explicit operator std::size_t() noexcept
{
return XXH3_64bits_digest(state_);
return detail::XXH64_digest(&state_);
}
};

View File

@@ -25,6 +25,7 @@
#include <ripple/basics/base_uint.h>
#include <ripple/beast/net/IPEndpoint.h>
#include <ripple/beast/utility/Journal.h>
#include <ripple/core/ConfigSections.h>
#include <ripple/protocol/PublicKey.h>
#include <ripple/protocol/SystemParameters.h> // VFALCO Breaks levelization
#include <boost/beast/core/string.hpp>
@@ -350,6 +351,21 @@ public:
{
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
useTxTables() const

View File

@@ -1894,7 +1894,9 @@ fromNetwork(
constexpr auto RPC_REPLY_MAX_BYTES = megabytes(256);
using namespace std::chrono_literals;
auto constexpr RPC_NOTIFY = 10min;
// auto constexpr RPC_NOTIFY = 10min; // Wietse: lolwut 10 minutes for one
// HTTP call?
auto constexpr RPC_NOTIFY = 30s;
HTTPClient::request(
bSSL,

View File

@@ -78,12 +78,14 @@ public:
{
std::lock_guard sl(mLock);
if (mDeque.size() >= eventQueueMax)
{
// Drop the previous event.
JLOG(j_.warn()) << "RPCCall::fromNetwork drop";
mDeque.pop_back();
}
// Wietse: we're not going to limit this, this is admin-port only, scale
// accordingly Dropping events just like this results in inconsistent
// data on the receiving end if (mDeque.size() >= eventQueueMax)
// {
// // Drop the previous event.
// JLOG(j_.warn()) << "RPCCall::fromNetwork drop";
// mDeque.pop_back();
// }
auto jm = broadcast ? j_.debug() : j_.info();
JLOG(jm) << "RPCCall::fromNetwork push: " << jvObj;
@@ -182,7 +184,8 @@ private:
}
private:
enum { eventQueueMax = 32 };
// Wietse: we're not going to limit this, this is admin-port only, scale
// accordingly enum { eventQueueMax = 32 };
boost::asio::io_service& m_io_service;
JobQueue& m_jobQueue;

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.
if (name_.empty())
Throw<std::runtime_error>("Missing path in Memory backend");
Throw<std::runtime_error>("Missing path in TestMemory backend");
}
~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/server/SimpleWriter.h>
#include <ripple/core/ConfigSections.h>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/utility/in_place_factory.hpp>
@@ -136,7 +137,11 @@ OverlayImpl::OverlayImpl(
stopwatch(),
app_.journal("PeerFinder"),
config,
collector))
collector,
app.config().section(SECTION_RELATIONAL_DB).empty() ||
!boost::iequals(
get(app.config().section(SECTION_RELATIONAL_DB), "backend"),
"rwdb")))
, m_resolver(resolver)
, next_id_(1)
, 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/impl/Checker.h>
#include <ripple/peerfinder/impl/InMemoryStore.h>
#include <ripple/peerfinder/impl/Logic.h>
#include <ripple/peerfinder/impl/SourceStrings.h>
#include <ripple/peerfinder/impl/StoreSqdb.h>
@@ -38,7 +40,7 @@ public:
std::optional<boost::asio::io_service::work> work_;
clock_type& m_clock;
beast::Journal m_journal;
StoreSqdb m_store;
std::unique_ptr<Store> m_store;
Checker<boost::asio::ip::tcp> checker_;
Logic<decltype(checker_)> m_logic;
BasicConfig const& m_config;
@@ -50,15 +52,18 @@ public:
clock_type& clock,
beast::Journal journal,
BasicConfig const& config,
beast::insight::Collector::ptr const& collector)
beast::insight::Collector::ptr const& collector,
bool useSqLiteStore)
: Manager()
, io_service_(io_service)
, work_(std::in_place, std::ref(io_service_))
, m_clock(clock)
, m_journal(journal)
, m_store(journal)
, m_store(
useSqLiteStore ? static_cast<Store*>(new StoreSqdb(journal))
: static_cast<Store*>(new InMemoryStore()))
, checker_(io_service_)
, m_logic(clock, m_store, checker_, journal)
, m_logic(clock, *m_store, checker_, journal)
, m_config(config)
, m_stats(std::bind(&ManagerImp::collect_metrics, this), collector)
{
@@ -215,7 +220,8 @@ public:
void
start() override
{
m_store.open(m_config);
if (auto sqdb = dynamic_cast<StoreSqdb*>(m_store.get()))
sqdb->open(m_config);
m_logic.load();
}
@@ -275,10 +281,11 @@ make_Manager(
clock_type& clock,
beast::Journal journal,
BasicConfig const& config,
beast::insight::Collector::ptr const& collector)
beast::insight::Collector::ptr const& collector,
bool useSqLiteStore)
{
return std::make_unique<ManagerImp>(
io_service, clock, journal, config, collector);
io_service, clock, journal, config, collector, useSqLiteStore);
}
} // namespace PeerFinder

View File

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

View File

@@ -74,7 +74,7 @@ namespace detail {
// 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
// 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.
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 fixPageCap;
extern uint256 const fix240911;
extern uint256 const fixFloatDivide;
} // namespace ripple

View File

@@ -467,6 +467,7 @@ REGISTER_FIX (fixNSDelete, Supported::yes, VoteBehavior::De
REGISTER_FIX (fix240819, Supported::yes, VoteBehavior::DefaultYes);
REGISTER_FIX (fixPageCap, 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
// because they could potentially get enabled.

View File

@@ -244,6 +244,7 @@ SHAMap::checkFilter(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
// Get a node without throwing
// Used on maps where missing nodes are expected
/*
std::shared_ptr<SHAMapTreeNode>
SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
{
@@ -266,6 +267,49 @@ SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
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>
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
std::unique_ptr<Backend> backend = Manager::instance().make_Backend(
@@ -105,6 +107,8 @@ public:
{
std::uint64_t const seedValue = 50;
testBackend("memory", seedValue);
testBackend("rwdb", seedValue);
testBackend("nudb", seedValue);
#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 ripple

View File

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