mirror of
https://github.com/XRPLF/rippled.git
synced 2026-02-03 13:35:25 +00:00
Compare commits
139 Commits
bthomee/ke
...
ximinez/on
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1a2ee706eb | ||
|
|
2a981357ba | ||
|
|
1ae475e724 | ||
|
|
a3e9401fbc | ||
|
|
9091469f9e | ||
|
|
17fa54f1f9 | ||
|
|
8fb5347c2d | ||
|
|
6739bf998f | ||
|
|
6eea38ba67 | ||
|
|
e9cf88b359 | ||
|
|
645b203476 | ||
|
|
be2aff1f4c | ||
|
|
56ed237e82 | ||
|
|
fd7b0fd135 | ||
|
|
e700994891 | ||
|
|
c76f7029ac | ||
|
|
d535c5fb2a | ||
|
|
54f860463e | ||
|
|
950434b8ff | ||
|
|
ee365e876d | ||
|
|
c6c59834b9 | ||
|
|
63b47914b8 | ||
|
|
9e02e5be2e | ||
|
|
093cd70fa1 | ||
|
|
376d65a483 | ||
|
|
a0d9a2458e | ||
|
|
456f639cf7 | ||
|
|
2c559ec2f3 | ||
|
|
619c81f463 | ||
|
|
f1490df960 | ||
|
|
7bdf74de98 | ||
|
|
1743d6fb98 | ||
|
|
ca7a5bb926 | ||
|
|
ce8b1a3f1e | ||
|
|
486fa75a10 | ||
|
|
f8d68cd3d3 | ||
|
|
ef7a3f5606 | ||
|
|
4f84ed7490 | ||
|
|
d534103131 | ||
|
|
82dff3c2ce | ||
|
|
30d73eb5ba | ||
|
|
1b2754bac2 | ||
|
|
cf80cafc75 | ||
|
|
b8897d51de | ||
|
|
3ff25eeb65 | ||
|
|
2bbfc4e786 | ||
|
|
2b1eb052e6 | ||
|
|
360e214e54 | ||
|
|
2618afed94 | ||
|
|
698ba2c788 | ||
|
|
b614e99588 | ||
|
|
fe8e4af2fa | ||
|
|
0a897f1528 | ||
|
|
cf8a3f5779 | ||
|
|
db39a39868 | ||
|
|
37a03d28c2 | ||
|
|
19d275425a | ||
|
|
88e9045602 | ||
|
|
5adbc536b6 | ||
|
|
e27af94ba9 | ||
|
|
43fe1e7e9c | ||
|
|
f456a858c8 | ||
|
|
084c3aa88e | ||
|
|
34f9b63921 | ||
|
|
bd3de79817 | ||
|
|
304eee2259 | ||
|
|
9e729b7f59 | ||
|
|
dd141468c4 | ||
|
|
933147c21f | ||
|
|
9201a4f591 | ||
|
|
5adb1e9b8b | ||
|
|
4df84d7988 | ||
|
|
cd87c0968b | ||
|
|
8a8e7c90bf | ||
|
|
e806069065 | ||
|
|
ce948cbec0 | ||
|
|
6ed34b3294 | ||
|
|
7161a235ca | ||
|
|
71463810de | ||
|
|
e997219a85 | ||
|
|
895cc13fa6 | ||
|
|
8d3c3ca29a | ||
|
|
9829553807 | ||
|
|
e551f9731a | ||
|
|
fd827bf58b | ||
|
|
5a3baba34d | ||
|
|
c78f5b160f | ||
|
|
485f78761a | ||
|
|
23cd2f7b21 | ||
|
|
5753266c43 | ||
|
|
4722d2607d | ||
|
|
85b5b4f855 | ||
|
|
a16f492f0f | ||
|
|
3633dc632c | ||
|
|
b3b30c3a86 | ||
|
|
c78a7684f4 | ||
|
|
cf83d92630 | ||
|
|
a56b1274d8 | ||
|
|
ae4bdd0492 | ||
|
|
e90102dd3b | ||
|
|
71f0e8db3d | ||
|
|
638929373a | ||
|
|
8440654377 | ||
|
|
9fa66c4741 | ||
|
|
38a9235145 | ||
|
|
c7a3cc9108 | ||
|
|
248337908d | ||
|
|
3d003619fd | ||
|
|
f163dca12c | ||
|
|
6e0ce458e5 | ||
|
|
5fae8480f1 | ||
|
|
e6587d374a | ||
|
|
376cc404e0 | ||
|
|
9898ca638f | ||
|
|
34b46d8f7c | ||
|
|
fe7d0798a7 | ||
|
|
0cecc09d71 | ||
|
|
e091d55561 | ||
|
|
69cf18158b | ||
|
|
6513c53817 | ||
|
|
e13baa58a5 | ||
|
|
951056fe9b | ||
|
|
67700ea6bd | ||
|
|
e5442cf3f1 | ||
|
|
da68076f04 | ||
|
|
b24116a118 | ||
|
|
f67398c6bf | ||
|
|
43d3eb1a24 | ||
|
|
0993315ed5 | ||
|
|
0bc383ada9 | ||
|
|
1841ceca43 | ||
|
|
2714cebabd | ||
|
|
e184db4ce2 | ||
|
|
ac6dc6943c | ||
|
|
ddd53806df | ||
|
|
e629a1f70e | ||
|
|
68076d969c | ||
|
|
d3009d3e1c | ||
|
|
54f7f3c894 |
@@ -940,7 +940,23 @@
|
||||
#
|
||||
# path Location to store the database
|
||||
#
|
||||
# Optional keys for NuDB and RocksDB:
|
||||
# Optional keys
|
||||
#
|
||||
# cache_size Size of cache for database records. Default is 16384.
|
||||
# Setting this value to 0 will use the default value.
|
||||
#
|
||||
# cache_age Length of time in minutes to keep database records
|
||||
# cached. Default is 5 minutes. Setting this value to
|
||||
# 0 will use the default value.
|
||||
#
|
||||
# Note: if neither cache_size nor cache_age is
|
||||
# specified, the cache for database records will not
|
||||
# be created. If only one of cache_size or cache_age
|
||||
# is specified, the cache will be created using the
|
||||
# default value for the unspecified parameter.
|
||||
#
|
||||
# Note: the cache will not be created if online_delete
|
||||
# is specified.
|
||||
#
|
||||
# fast_load Boolean. If set, load the last persisted ledger
|
||||
# from disk upon process start before syncing to
|
||||
@@ -948,6 +964,8 @@
|
||||
# if sufficient IOPS capacity is available.
|
||||
# Default 0.
|
||||
#
|
||||
# Optional keys for NuDB or RocksDB:
|
||||
#
|
||||
# earliest_seq The default is 32570 to match the XRP ledger
|
||||
# network's earliest allowed sequence. Alternate
|
||||
# networks may set this value. Minimum value of 1.
|
||||
@@ -1033,10 +1051,11 @@
|
||||
# The online delete process checks periodically
|
||||
# that xrpld is still in sync with the network,
|
||||
# and that the validated ledger is less than
|
||||
# 'age_threshold_seconds' old. If not, then continue
|
||||
# 'age_threshold_seconds' old, and that all
|
||||
# recent ledgers are available. If not, then continue
|
||||
# sleeping for this number of seconds and
|
||||
# checking until healthy.
|
||||
# Default is 5.
|
||||
# Default is 1.
|
||||
#
|
||||
# Notes:
|
||||
# The 'node_db' entry configures the primary, persistent storage.
|
||||
|
||||
@@ -77,16 +77,16 @@ public:
|
||||
If the object is not found or an error is encountered, the
|
||||
result will indicate the condition.
|
||||
@note This will be called concurrently.
|
||||
@param hash The hash of the object.
|
||||
@param key A pointer to the key data.
|
||||
@param pObject [out] The created object if successful.
|
||||
@return The result of the operation.
|
||||
*/
|
||||
virtual Status
|
||||
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) = 0;
|
||||
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) = 0;
|
||||
|
||||
/** Fetch a batch synchronously. */
|
||||
virtual std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
|
||||
fetchBatch(std::vector<uint256> const& hashes) = 0;
|
||||
fetchBatch(std::vector<uint256 const*> const& hashes) = 0;
|
||||
|
||||
/** Store a single object.
|
||||
Depending on the implementation this may happen immediately
|
||||
|
||||
@@ -24,6 +24,32 @@ public:
|
||||
beast::Journal j)
|
||||
: Database(scheduler, readThreads, config, j), backend_(std::move(backend))
|
||||
{
|
||||
std::optional<int> cacheSize, cacheAge;
|
||||
|
||||
if (config.exists("cache_size"))
|
||||
{
|
||||
cacheSize = get<int>(config, "cache_size");
|
||||
if (cacheSize.value() < 0)
|
||||
{
|
||||
Throw<std::runtime_error>("Specified negative value for cache_size");
|
||||
}
|
||||
}
|
||||
|
||||
if (config.exists("cache_age"))
|
||||
{
|
||||
cacheAge = get<int>(config, "cache_age");
|
||||
if (cacheAge.value() < 0)
|
||||
{
|
||||
Throw<std::runtime_error>("Specified negative value for cache_age");
|
||||
}
|
||||
}
|
||||
|
||||
if (cacheSize != 0 || cacheAge != 0)
|
||||
{
|
||||
cache_ = std::make_shared<TaggedCache<uint256, NodeObject>>(
|
||||
"DatabaseNodeImp", cacheSize.value_or(0), std::chrono::minutes(cacheAge.value_or(0)), stopwatch(), j);
|
||||
}
|
||||
|
||||
XRPL_ASSERT(
|
||||
backend_,
|
||||
"xrpl::NodeStore::DatabaseNodeImp::DatabaseNodeImp : non-null "
|
||||
@@ -82,6 +108,9 @@ public:
|
||||
sweep() override;
|
||||
|
||||
private:
|
||||
// Cache for database objects. This cache is not always initialized. Check
|
||||
// for null before using.
|
||||
std::shared_ptr<TaggedCache<uint256, NodeObject>> cache_;
|
||||
// Persistent key/value storage
|
||||
std::shared_ptr<Backend> backend_;
|
||||
|
||||
|
||||
@@ -10,6 +10,11 @@ DatabaseNodeImp::store(NodeObjectType type, Blob&& data, uint256 const& hash, st
|
||||
|
||||
auto obj = NodeObject::createObject(type, std::move(data), hash);
|
||||
backend_->store(obj);
|
||||
if (cache_)
|
||||
{
|
||||
// After the store, replace a negative cache entry if there is one
|
||||
cache_->canonicalize(hash, obj, [](std::shared_ptr<NodeObject> const& n) { return n->getType() == hotDUMMY; });
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
@@ -18,41 +23,77 @@ DatabaseNodeImp::asyncFetch(
|
||||
std::uint32_t ledgerSeq,
|
||||
std::function<void(std::shared_ptr<NodeObject> const&)>&& callback)
|
||||
{
|
||||
if (cache_)
|
||||
{
|
||||
std::shared_ptr<NodeObject> obj = cache_->fetch(hash);
|
||||
if (obj)
|
||||
{
|
||||
callback(obj->getType() == hotDUMMY ? nullptr : obj);
|
||||
return;
|
||||
}
|
||||
}
|
||||
Database::asyncFetch(hash, ledgerSeq, std::move(callback));
|
||||
}
|
||||
|
||||
void
|
||||
DatabaseNodeImp::sweep()
|
||||
{
|
||||
if (cache_)
|
||||
cache_->sweep();
|
||||
}
|
||||
|
||||
std::shared_ptr<NodeObject>
|
||||
DatabaseNodeImp::fetchNodeObject(uint256 const& hash, std::uint32_t, FetchReport& fetchReport, bool duplicate)
|
||||
{
|
||||
std::shared_ptr<NodeObject> nodeObject = nullptr;
|
||||
Status status;
|
||||
std::shared_ptr<NodeObject> nodeObject = cache_ ? cache_->fetch(hash) : nullptr;
|
||||
|
||||
try
|
||||
if (!nodeObject)
|
||||
{
|
||||
status = backend_->fetch(hash, &nodeObject);
|
||||
}
|
||||
catch (std::exception const& e)
|
||||
{
|
||||
JLOG(j_.fatal()) << "fetchNodeObject " << hash << ": Exception fetching from backend: " << e.what();
|
||||
Rethrow();
|
||||
}
|
||||
JLOG(j_.trace()) << "fetchNodeObject " << hash << ": record not " << (cache_ ? "cached" : "found");
|
||||
|
||||
switch (status)
|
||||
Status status;
|
||||
|
||||
try
|
||||
{
|
||||
status = backend_->fetch(hash.data(), &nodeObject);
|
||||
}
|
||||
catch (std::exception const& e)
|
||||
{
|
||||
JLOG(j_.fatal()) << "fetchNodeObject " << hash << ": Exception fetching from backend: " << e.what();
|
||||
Rethrow();
|
||||
}
|
||||
|
||||
switch (status)
|
||||
{
|
||||
case ok:
|
||||
if (cache_)
|
||||
{
|
||||
if (nodeObject)
|
||||
cache_->canonicalize_replace_client(hash, nodeObject);
|
||||
else
|
||||
{
|
||||
auto notFound = NodeObject::createObject(hotDUMMY, {}, hash);
|
||||
cache_->canonicalize_replace_client(hash, notFound);
|
||||
if (notFound->getType() != hotDUMMY)
|
||||
nodeObject = notFound;
|
||||
}
|
||||
}
|
||||
break;
|
||||
case notFound:
|
||||
break;
|
||||
case dataCorrupt:
|
||||
JLOG(j_.fatal()) << "fetchNodeObject " << hash << ": nodestore data is corrupted";
|
||||
break;
|
||||
default:
|
||||
JLOG(j_.warn()) << "fetchNodeObject " << hash << ": backend returns unknown result " << status;
|
||||
break;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
case ok:
|
||||
case notFound:
|
||||
break;
|
||||
case dataCorrupt:
|
||||
JLOG(j_.fatal()) << "fetchNodeObject " << hash << ": nodestore data is corrupted";
|
||||
break;
|
||||
default:
|
||||
JLOG(j_.warn()) << "fetchNodeObject " << hash << ": backend returns unknown result " << status;
|
||||
break;
|
||||
JLOG(j_.trace()) << "fetchNodeObject " << hash << ": record found in cache";
|
||||
if (nodeObject->getType() == hotDUMMY)
|
||||
nodeObject.reset();
|
||||
}
|
||||
|
||||
if (nodeObject)
|
||||
@@ -64,22 +105,66 @@ DatabaseNodeImp::fetchNodeObject(uint256 const& hash, std::uint32_t, FetchReport
|
||||
std::vector<std::shared_ptr<NodeObject>>
|
||||
DatabaseNodeImp::fetchBatch(std::vector<uint256> const& hashes)
|
||||
{
|
||||
std::vector<std::shared_ptr<NodeObject>> results{hashes.size()};
|
||||
using namespace std::chrono;
|
||||
auto const before = steady_clock::now();
|
||||
|
||||
std::vector<std::shared_ptr<NodeObject>> results{hashes.size()};
|
||||
results = backend_->fetchBatch(hashes).first;
|
||||
for (size_t i = 0; i < results.size(); ++i)
|
||||
std::unordered_map<uint256 const*, size_t> indexMap;
|
||||
std::vector<uint256 const*> cacheMisses;
|
||||
uint64_t hits = 0;
|
||||
uint64_t fetches = 0;
|
||||
for (size_t i = 0; i < hashes.size(); ++i)
|
||||
{
|
||||
if (!results[i])
|
||||
auto const& hash = hashes[i];
|
||||
// See if the object already exists in the cache
|
||||
auto nObj = cache_ ? cache_->fetch(hash) : nullptr;
|
||||
++fetches;
|
||||
if (!nObj)
|
||||
{
|
||||
JLOG(j_.error()) << "fetchBatch - "
|
||||
<< "record not found in db. hash = " << strHex(hashes[i]);
|
||||
// Try the database
|
||||
indexMap[&hash] = i;
|
||||
cacheMisses.push_back(&hash);
|
||||
}
|
||||
else
|
||||
{
|
||||
results[i] = nObj->getType() == hotDUMMY ? nullptr : nObj;
|
||||
// It was in the cache.
|
||||
++hits;
|
||||
}
|
||||
}
|
||||
|
||||
JLOG(j_.debug()) << "fetchBatch - cache hits = " << (hashes.size() - cacheMisses.size())
|
||||
<< " - cache misses = " << cacheMisses.size();
|
||||
auto dbResults = backend_->fetchBatch(cacheMisses).first;
|
||||
|
||||
for (size_t i = 0; i < dbResults.size(); ++i)
|
||||
{
|
||||
auto nObj = std::move(dbResults[i]);
|
||||
size_t index = indexMap[cacheMisses[i]];
|
||||
auto const& hash = hashes[index];
|
||||
|
||||
if (nObj)
|
||||
{
|
||||
// Ensure all threads get the same object
|
||||
if (cache_)
|
||||
cache_->canonicalize_replace_client(hash, nObj);
|
||||
}
|
||||
else
|
||||
{
|
||||
JLOG(j_.error()) << "fetchBatch - "
|
||||
<< "record not found in db or cache. hash = " << strHex(hash);
|
||||
if (cache_)
|
||||
{
|
||||
auto notFound = NodeObject::createObject(hotDUMMY, {}, hash);
|
||||
cache_->canonicalize_replace_client(hash, notFound);
|
||||
if (notFound->getType() != hotDUMMY)
|
||||
nObj = std::move(notFound);
|
||||
}
|
||||
}
|
||||
results[index] = std::move(nObj);
|
||||
}
|
||||
|
||||
auto fetchDurationUs = std::chrono::duration_cast<std::chrono::microseconds>(steady_clock::now() - before).count();
|
||||
updateFetchMetrics(hashes.size(), 0, fetchDurationUs);
|
||||
updateFetchMetrics(fetches, hits, fetchDurationUs);
|
||||
return results;
|
||||
}
|
||||
|
||||
|
||||
@@ -107,7 +107,7 @@ DatabaseRotatingImp::fetchNodeObject(uint256 const& hash, std::uint32_t, FetchRe
|
||||
std::shared_ptr<NodeObject> nodeObject;
|
||||
try
|
||||
{
|
||||
status = backend->fetch(hash, &nodeObject);
|
||||
status = backend->fetch(hash.data(), &nodeObject);
|
||||
}
|
||||
catch (std::exception const& e)
|
||||
{
|
||||
|
||||
@@ -115,9 +115,10 @@ public:
|
||||
//--------------------------------------------------------------------------
|
||||
|
||||
Status
|
||||
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) override
|
||||
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
|
||||
{
|
||||
XRPL_ASSERT(db_, "xrpl::NodeStore::MemoryBackend::fetch : non-null database");
|
||||
uint256 const hash(uint256::fromVoid(key));
|
||||
|
||||
std::lock_guard _(db_->mutex);
|
||||
|
||||
@@ -132,14 +133,14 @@ public:
|
||||
}
|
||||
|
||||
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
|
||||
fetchBatch(std::vector<uint256> const& hashes) override
|
||||
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, &nObj);
|
||||
Status status = fetch(h->begin(), &nObj);
|
||||
if (status != ok)
|
||||
results.push_back({});
|
||||
else
|
||||
|
||||
@@ -177,17 +177,17 @@ public:
|
||||
}
|
||||
|
||||
Status
|
||||
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pno) override
|
||||
fetch(void const* key, std::shared_ptr<NodeObject>* pno) override
|
||||
{
|
||||
Status status;
|
||||
pno->reset();
|
||||
nudb::error_code ec;
|
||||
db_.fetch(
|
||||
hash.data(),
|
||||
[hash, pno, &status](void const* data, std::size_t size) {
|
||||
key,
|
||||
[key, pno, &status](void const* data, std::size_t size) {
|
||||
nudb::detail::buffer bf;
|
||||
auto const result = nodeobject_decompress(data, size, bf);
|
||||
DecodedBlob decoded(hash.data(), result.first, result.second);
|
||||
DecodedBlob decoded(key, result.first, result.second);
|
||||
if (!decoded.wasOk())
|
||||
{
|
||||
status = dataCorrupt;
|
||||
@@ -205,14 +205,14 @@ public:
|
||||
}
|
||||
|
||||
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
|
||||
fetchBatch(std::vector<uint256> const& hashes) override
|
||||
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, &nObj);
|
||||
Status status = fetch(h->begin(), &nObj);
|
||||
if (status != ok)
|
||||
results.push_back({});
|
||||
else
|
||||
|
||||
@@ -36,13 +36,13 @@ public:
|
||||
}
|
||||
|
||||
Status
|
||||
fetch(uint256 const&, std::shared_ptr<NodeObject>*) override
|
||||
fetch(void const*, std::shared_ptr<NodeObject>*) override
|
||||
{
|
||||
return notFound;
|
||||
}
|
||||
|
||||
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
|
||||
fetchBatch(std::vector<uint256> const& hashes) override
|
||||
fetchBatch(std::vector<uint256 const*> const& hashes) override
|
||||
{
|
||||
return {};
|
||||
}
|
||||
|
||||
@@ -237,7 +237,7 @@ public:
|
||||
//--------------------------------------------------------------------------
|
||||
|
||||
Status
|
||||
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) override
|
||||
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
|
||||
{
|
||||
XRPL_ASSERT(m_db, "xrpl::NodeStore::RocksDBBackend::fetch : non-null database");
|
||||
pObject->reset();
|
||||
@@ -245,7 +245,7 @@ public:
|
||||
Status status(ok);
|
||||
|
||||
rocksdb::ReadOptions const options;
|
||||
rocksdb::Slice const slice(reinterpret_cast<char const*>(hash.data()), m_keyBytes);
|
||||
rocksdb::Slice const slice(static_cast<char const*>(key), m_keyBytes);
|
||||
|
||||
std::string string;
|
||||
|
||||
@@ -253,7 +253,7 @@ public:
|
||||
|
||||
if (getStatus.ok())
|
||||
{
|
||||
DecodedBlob decoded(hash.data(), string.data(), string.size());
|
||||
DecodedBlob decoded(key, string.data(), string.size());
|
||||
|
||||
if (decoded.wasOk())
|
||||
{
|
||||
@@ -288,14 +288,14 @@ public:
|
||||
}
|
||||
|
||||
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
|
||||
fetchBatch(std::vector<uint256> const& hashes) override
|
||||
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, &nObj);
|
||||
Status status = fetch(h->begin(), &nObj);
|
||||
if (status != ok)
|
||||
results.push_back({});
|
||||
else
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
#include <test/jtx/Env.h>
|
||||
|
||||
#include <xrpld/app/ledger/LedgerMaster.h>
|
||||
#include <xrpld/app/misc/SHAMapStore.h>
|
||||
|
||||
namespace xrpl {
|
||||
namespace test {
|
||||
@@ -93,6 +94,70 @@ class LedgerMaster_test : public beast::unit_test::suite
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
testCompleteLedgerRange(FeatureBitset features)
|
||||
{
|
||||
// Note that this test is intentionally very similar to
|
||||
// SHAMapStore_test::testLedgerGaps, but has a different
|
||||
// focus.
|
||||
|
||||
testcase("Complete Ledger operations");
|
||||
|
||||
using namespace test::jtx;
|
||||
|
||||
auto const deleteInterval = 8;
|
||||
|
||||
Env env{*this, envconfig([](auto cfg) { return online_delete(std::move(cfg), deleteInterval); })};
|
||||
|
||||
auto const alice = Account("alice");
|
||||
env.fund(XRP(1000), alice);
|
||||
env.close();
|
||||
|
||||
auto& lm = env.app().getLedgerMaster();
|
||||
LedgerIndex minSeq = 2;
|
||||
LedgerIndex maxSeq = env.closed()->header().seq;
|
||||
auto& store = env.app().getSHAMapStore();
|
||||
store.rendezvous();
|
||||
LedgerIndex lastRotated = store.getLastRotated();
|
||||
BEAST_EXPECTS(maxSeq == 3, to_string(maxSeq));
|
||||
BEAST_EXPECTS(lm.getCompleteLedgers() == "2-3", lm.getCompleteLedgers());
|
||||
BEAST_EXPECTS(lastRotated == 3, to_string(lastRotated));
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
|
||||
|
||||
// Close enough ledgers to rotate a few times
|
||||
for (int i = 0; i < 24; ++i)
|
||||
{
|
||||
for (int t = 0; t < 3; ++t)
|
||||
{
|
||||
env(noop(alice));
|
||||
}
|
||||
env.close();
|
||||
store.rendezvous();
|
||||
|
||||
++maxSeq;
|
||||
|
||||
if (maxSeq == lastRotated + deleteInterval)
|
||||
{
|
||||
minSeq = lastRotated;
|
||||
lastRotated = maxSeq;
|
||||
}
|
||||
BEAST_EXPECTS(env.closed()->header().seq == maxSeq, to_string(env.closed()->header().seq));
|
||||
BEAST_EXPECTS(store.getLastRotated() == lastRotated, to_string(store.getLastRotated()));
|
||||
std::stringstream expectedRange;
|
||||
expectedRange << minSeq << "-" << maxSeq;
|
||||
BEAST_EXPECTS(lm.getCompleteLedgers() == expectedRange.str(), lm.getCompleteLedgers());
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
|
||||
}
|
||||
}
|
||||
|
||||
public:
|
||||
void
|
||||
run() override
|
||||
@@ -106,6 +171,7 @@ public:
|
||||
testWithFeats(FeatureBitset features)
|
||||
{
|
||||
testTxnIdFromIndex(features);
|
||||
testCompleteLedgerRange(features);
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
#include <test/jtx.h>
|
||||
#include <test/jtx/envconfig.h>
|
||||
|
||||
#include <xrpld/app/ledger/LedgerMaster.h>
|
||||
#include <xrpld/app/main/Application.h>
|
||||
#include <xrpld/app/main/NodeStoreScheduler.h>
|
||||
#include <xrpld/app/misc/SHAMapStore.h>
|
||||
@@ -10,6 +11,8 @@
|
||||
#include <xrpl/nodestore/detail/DatabaseRotatingImp.h>
|
||||
#include <xrpl/protocol/jss.h>
|
||||
|
||||
#include <thread>
|
||||
|
||||
namespace xrpl {
|
||||
namespace test {
|
||||
|
||||
@@ -20,10 +23,8 @@ class SHAMapStore_test : public beast::unit_test::suite
|
||||
static auto
|
||||
onlineDelete(std::unique_ptr<Config> cfg)
|
||||
{
|
||||
cfg->LEDGER_HISTORY = deleteInterval;
|
||||
auto& section = cfg->section(ConfigSection::nodeDatabase());
|
||||
section.set("online_delete", std::to_string(deleteInterval));
|
||||
return cfg;
|
||||
using namespace jtx;
|
||||
return online_delete(std::move(cfg), deleteInterval);
|
||||
}
|
||||
|
||||
static auto
|
||||
@@ -490,8 +491,19 @@ public:
|
||||
Env env(*this, envconfig(onlineDelete));
|
||||
|
||||
/////////////////////////////////////////////////////////////
|
||||
// Create NodeStore with two backends to allow online deletion of data.
|
||||
// Normally, SHAMapStoreImp handles all these details.
|
||||
// Create the backend. Normally, SHAMapStoreImp handles all these
|
||||
// details
|
||||
auto nscfg = env.app().config().section(ConfigSection::nodeDatabase());
|
||||
|
||||
// Provide default values:
|
||||
if (!nscfg.exists("cache_size"))
|
||||
nscfg.set(
|
||||
"cache_size", std::to_string(env.app().config().getValueFor(SizedItem::treeCacheSize, std::nullopt)));
|
||||
|
||||
if (!nscfg.exists("cache_age"))
|
||||
nscfg.set(
|
||||
"cache_age", std::to_string(env.app().config().getValueFor(SizedItem::treeCacheAge, std::nullopt)));
|
||||
|
||||
NodeStoreScheduler scheduler(env.app().getJobQueue());
|
||||
|
||||
std::string const writableDb = "write";
|
||||
@@ -499,8 +511,9 @@ public:
|
||||
auto writableBackend = makeBackendRotating(env, scheduler, writableDb);
|
||||
auto archiveBackend = makeBackendRotating(env, scheduler, archiveDb);
|
||||
|
||||
// Create NodeStore with two backends to allow online deletion of
|
||||
// data
|
||||
constexpr int readThreads = 4;
|
||||
auto nscfg = env.app().config().section(ConfigSection::nodeDatabase());
|
||||
auto dbr = std::make_unique<NodeStore::DatabaseRotatingImp>(
|
||||
scheduler,
|
||||
readThreads,
|
||||
@@ -555,6 +568,151 @@ public:
|
||||
BEAST_EXPECT(dbr->getName() == "3");
|
||||
}
|
||||
|
||||
void
|
||||
testLedgerGaps()
|
||||
{
|
||||
// Note that this test is intentionally very similar to
|
||||
// LedgerMaster_test::testCompleteLedgerRange, but has a different
|
||||
// focus.
|
||||
|
||||
testcase("Wait for ledger gaps to fill in");
|
||||
|
||||
using namespace test::jtx;
|
||||
|
||||
Env env{*this, envconfig(onlineDelete)};
|
||||
|
||||
std::map<LedgerIndex, uint256> hashes;
|
||||
|
||||
auto failureMessage = [&](char const* label, auto expected, auto actual) {
|
||||
std::stringstream ss;
|
||||
ss << label << ": Expected: " << expected << ", Got: " << actual;
|
||||
return ss.str();
|
||||
};
|
||||
|
||||
auto const alice = Account("alice");
|
||||
env.fund(XRP(1000), alice);
|
||||
env.close();
|
||||
|
||||
auto& lm = env.app().getLedgerMaster();
|
||||
LedgerIndex minSeq = 2;
|
||||
LedgerIndex maxSeq = env.closed()->header().seq;
|
||||
auto& store = env.app().getSHAMapStore();
|
||||
store.rendezvous();
|
||||
LedgerIndex lastRotated = store.getLastRotated();
|
||||
BEAST_EXPECTS(maxSeq == 3, to_string(maxSeq));
|
||||
BEAST_EXPECTS(lm.getCompleteLedgers() == "2-3", lm.getCompleteLedgers());
|
||||
BEAST_EXPECTS(lastRotated == 3, to_string(lastRotated));
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
|
||||
|
||||
// Close enough ledgers to rotate a few times
|
||||
while (maxSeq < 20)
|
||||
{
|
||||
for (int t = 0; t < 3; ++t)
|
||||
{
|
||||
env(noop(alice));
|
||||
}
|
||||
env.close();
|
||||
store.rendezvous();
|
||||
|
||||
++maxSeq;
|
||||
|
||||
if (maxSeq + 1 == lastRotated + deleteInterval)
|
||||
{
|
||||
using namespace std::chrono_literals;
|
||||
|
||||
// The next ledger will trigger a rotation. Delete the
|
||||
// current ledger from LedgerMaster.
|
||||
std::this_thread::sleep_for(100ms);
|
||||
LedgerIndex const deleteSeq = maxSeq;
|
||||
while (!lm.haveLedger(deleteSeq))
|
||||
{
|
||||
std::this_thread::sleep_for(100ms);
|
||||
}
|
||||
lm.clearLedger(deleteSeq);
|
||||
|
||||
auto expectedRange = [](auto minSeq, auto deleteSeq, auto maxSeq) {
|
||||
std::stringstream expectedRange;
|
||||
expectedRange << minSeq << "-" << (deleteSeq - 1);
|
||||
if (deleteSeq + 1 == maxSeq)
|
||||
expectedRange << "," << maxSeq;
|
||||
else if (deleteSeq < maxSeq)
|
||||
expectedRange << "," << (deleteSeq + 1) << "-" << maxSeq;
|
||||
return expectedRange.str();
|
||||
};
|
||||
BEAST_EXPECTS(
|
||||
lm.getCompleteLedgers() == expectedRange(minSeq, deleteSeq, maxSeq),
|
||||
failureMessage(
|
||||
"Complete ledgers", expectedRange(minSeq, deleteSeq, maxSeq), lm.getCompleteLedgers()));
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 1);
|
||||
|
||||
// Close another ledger, which will trigger a rotation, but the
|
||||
// rotation will be stuck until the missing ledger is filled in.
|
||||
env.close();
|
||||
// DO NOT CALL rendezvous()! You'll end up with a deadlock.
|
||||
++maxSeq;
|
||||
|
||||
// Nothing has changed
|
||||
BEAST_EXPECTS(
|
||||
store.getLastRotated() == lastRotated,
|
||||
failureMessage("lastRotated", lastRotated, store.getLastRotated()));
|
||||
BEAST_EXPECTS(
|
||||
lm.getCompleteLedgers() == expectedRange(minSeq, deleteSeq, maxSeq),
|
||||
failureMessage(
|
||||
"Complete ledgers", expectedRange(minSeq, deleteSeq, maxSeq), lm.getCompleteLedgers()));
|
||||
|
||||
// Close 5 more ledgers, waiting one second in between to
|
||||
// simulate the ledger making progress while online delete waits
|
||||
// for the missing ledger to be filled in.
|
||||
// This ensures the healthWait check has time to run and
|
||||
// detect the gap.
|
||||
for (int l = 0; l < 5; ++l)
|
||||
{
|
||||
env.close();
|
||||
// DO NOT CALL rendezvous()! You'll end up with a deadlock.
|
||||
++maxSeq;
|
||||
// Nothing has changed
|
||||
BEAST_EXPECTS(
|
||||
store.getLastRotated() == lastRotated,
|
||||
failureMessage("lastRotated", lastRotated, store.getLastRotated()));
|
||||
BEAST_EXPECTS(
|
||||
lm.getCompleteLedgers() == expectedRange(minSeq, deleteSeq, maxSeq),
|
||||
failureMessage(
|
||||
"Complete Ledgers", expectedRange(minSeq, deleteSeq, maxSeq), lm.getCompleteLedgers()));
|
||||
std::this_thread::sleep_for(1s);
|
||||
}
|
||||
|
||||
// Put the missing ledger back in LedgerMaster
|
||||
lm.setLedgerRangePresent(deleteSeq, deleteSeq);
|
||||
|
||||
// Wait for the rotation to finish
|
||||
store.rendezvous();
|
||||
|
||||
minSeq = lastRotated;
|
||||
lastRotated = deleteSeq + 1;
|
||||
}
|
||||
BEAST_EXPECT(maxSeq != lastRotated + deleteInterval);
|
||||
BEAST_EXPECTS(
|
||||
env.closed()->header().seq == maxSeq, failureMessage("maxSeq", maxSeq, env.closed()->header().seq));
|
||||
BEAST_EXPECTS(
|
||||
store.getLastRotated() == lastRotated,
|
||||
failureMessage("lastRotated", lastRotated, store.getLastRotated()));
|
||||
std::stringstream expectedRange;
|
||||
expectedRange << minSeq << "-" << maxSeq;
|
||||
BEAST_EXPECTS(
|
||||
lm.getCompleteLedgers() == expectedRange.str(),
|
||||
failureMessage("CompleteLedgers", expectedRange.str(), lm.getCompleteLedgers()));
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
|
||||
BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
run() override
|
||||
{
|
||||
@@ -562,6 +720,7 @@ public:
|
||||
testAutomatic();
|
||||
testCanDelete();
|
||||
testRotate();
|
||||
testLedgerGaps();
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -58,6 +58,17 @@ envconfig(F&& modfunc, Args&&... args)
|
||||
return modfunc(envconfig(), std::forward<Args>(args)...);
|
||||
}
|
||||
|
||||
/// @brief adjust config to enable online_delete
|
||||
///
|
||||
/// @param cfg config instance to be modified
|
||||
///
|
||||
/// @param deleteInterval how many new ledgers should be available before
|
||||
/// rotating. Defaults to 8, because the standalone minimum is 8.
|
||||
///
|
||||
/// @return unique_ptr to Config instance
|
||||
std::unique_ptr<Config>
|
||||
online_delete(std::unique_ptr<Config> cfg, std::uint32_t deleteInterval = 8);
|
||||
|
||||
/// @brief adjust config so no admin ports are enabled
|
||||
///
|
||||
/// this is intended for use with envconfig, as in
|
||||
|
||||
@@ -53,6 +53,15 @@ setupConfigForUnitTests(Config& cfg)
|
||||
|
||||
namespace jtx {
|
||||
|
||||
std::unique_ptr<Config>
|
||||
online_delete(std::unique_ptr<Config> cfg, std::uint32_t deleteInterval)
|
||||
{
|
||||
cfg->LEDGER_HISTORY = deleteInterval;
|
||||
auto& section = cfg->section(ConfigSection::nodeDatabase());
|
||||
section.set("online_delete", std::to_string(deleteInterval));
|
||||
return cfg;
|
||||
}
|
||||
|
||||
std::unique_ptr<Config>
|
||||
no_admin(std::unique_ptr<Config> cfg)
|
||||
{
|
||||
|
||||
@@ -138,7 +138,7 @@ public:
|
||||
{
|
||||
std::shared_ptr<NodeObject> object;
|
||||
|
||||
Status const status = backend.fetch(batch[i]->getHash(), &object);
|
||||
Status const status = backend.fetch(batch[i]->getHash().cbegin(), &object);
|
||||
|
||||
BEAST_EXPECT(status == ok);
|
||||
|
||||
@@ -158,7 +158,7 @@ public:
|
||||
{
|
||||
std::shared_ptr<NodeObject> object;
|
||||
|
||||
Status const status = backend.fetch(batch[i]->getHash(), &object);
|
||||
Status const status = backend.fetch(batch[i]->getHash().cbegin(), &object);
|
||||
|
||||
BEAST_EXPECT(status == notFound);
|
||||
}
|
||||
|
||||
@@ -313,7 +313,7 @@ public:
|
||||
std::shared_ptr<NodeObject> obj;
|
||||
std::shared_ptr<NodeObject> result;
|
||||
obj = seq1_.obj(dist_(gen_));
|
||||
backend_.fetch(obj->getHash(), &result);
|
||||
backend_.fetch(obj->getHash().data(), &result);
|
||||
suite_.expect(result && isSame(result, obj));
|
||||
}
|
||||
catch (std::exception const& e)
|
||||
@@ -371,9 +371,9 @@ public:
|
||||
{
|
||||
try
|
||||
{
|
||||
auto const hash = seq2_.key(i);
|
||||
auto const key = seq2_.key(i);
|
||||
std::shared_ptr<NodeObject> result;
|
||||
backend_.fetch(hash, &result);
|
||||
backend_.fetch(key.data(), &result);
|
||||
suite_.expect(!result);
|
||||
}
|
||||
catch (std::exception const& e)
|
||||
@@ -438,9 +438,9 @@ public:
|
||||
{
|
||||
if (rand_(gen_) < missingNodePercent)
|
||||
{
|
||||
auto const hash = seq2_.key(dist_(gen_));
|
||||
auto const key = seq2_.key(dist_(gen_));
|
||||
std::shared_ptr<NodeObject> result;
|
||||
backend_.fetch(hash, &result);
|
||||
backend_.fetch(key.data(), &result);
|
||||
suite_.expect(!result);
|
||||
}
|
||||
else
|
||||
@@ -448,7 +448,7 @@ public:
|
||||
std::shared_ptr<NodeObject> obj;
|
||||
std::shared_ptr<NodeObject> result;
|
||||
obj = seq1_.obj(dist_(gen_));
|
||||
backend_.fetch(obj->getHash(), &result);
|
||||
backend_.fetch(obj->getHash().data(), &result);
|
||||
suite_.expect(result && isSame(result, obj));
|
||||
}
|
||||
}
|
||||
@@ -525,7 +525,7 @@ public:
|
||||
auto const j = older_(gen_);
|
||||
obj = seq1_.obj(j);
|
||||
std::shared_ptr<NodeObject> result1;
|
||||
backend_.fetch(obj->getHash(), &result);
|
||||
backend_.fetch(obj->getHash().data(), &result);
|
||||
suite_.expect(result != nullptr);
|
||||
suite_.expect(isSame(result, obj));
|
||||
}
|
||||
@@ -543,7 +543,7 @@ public:
|
||||
std::shared_ptr<NodeObject> result;
|
||||
auto const j = recent_(gen_);
|
||||
obj = seq1_.obj(j);
|
||||
backend_.fetch(obj->getHash(), &result);
|
||||
backend_.fetch(obj->getHash().data(), &result);
|
||||
suite_.expect(!result || isSame(result, obj));
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -105,7 +105,10 @@ public:
|
||||
failedSave(std::uint32_t seq, uint256 const& hash);
|
||||
|
||||
std::string
|
||||
getCompleteLedgers();
|
||||
getCompleteLedgers() const;
|
||||
|
||||
std::size_t
|
||||
missingFromCompleteLedgerRange(LedgerIndex first, LedgerIndex last) const;
|
||||
|
||||
/** Apply held transactions to the open ledger
|
||||
This is normally called as we close the ledger.
|
||||
@@ -319,7 +322,7 @@ private:
|
||||
// A set of transactions to replay during the next close
|
||||
std::unique_ptr<LedgerReplay> replayData;
|
||||
|
||||
std::recursive_mutex mCompleteLock;
|
||||
std::recursive_mutex mutable mCompleteLock;
|
||||
RangeSet<std::uint32_t> mCompleteLedgers;
|
||||
|
||||
// Publish thread is running.
|
||||
|
||||
@@ -1475,12 +1475,34 @@ LedgerMaster::getPublishedLedger()
|
||||
}
|
||||
|
||||
std::string
|
||||
LedgerMaster::getCompleteLedgers()
|
||||
LedgerMaster::getCompleteLedgers() const
|
||||
{
|
||||
std::lock_guard sl(mCompleteLock);
|
||||
return to_string(mCompleteLedgers);
|
||||
}
|
||||
|
||||
std::size_t
|
||||
LedgerMaster::missingFromCompleteLedgerRange(LedgerIndex first, LedgerIndex last) const
|
||||
{
|
||||
// Make a copy of the range to avoid holding the lock
|
||||
auto const range = [&] {
|
||||
std::lock_guard sl(mCompleteLock);
|
||||
return mCompleteLedgers;
|
||||
}();
|
||||
|
||||
std::size_t missing = 0;
|
||||
|
||||
for (LedgerIndex idx = first; idx <= last; ++idx)
|
||||
{
|
||||
if (!boost::icl::contains(range, idx))
|
||||
{
|
||||
++missing;
|
||||
}
|
||||
}
|
||||
|
||||
return missing;
|
||||
}
|
||||
|
||||
std::optional<NetClock::time_point>
|
||||
LedgerMaster::getCloseTimeBySeq(LedgerIndex ledgerIndex)
|
||||
{
|
||||
|
||||
@@ -130,6 +130,14 @@ std::unique_ptr<NodeStore::Database>
|
||||
SHAMapStoreImp::makeNodeStore(int readThreads)
|
||||
{
|
||||
auto nscfg = app_.config().section(ConfigSection::nodeDatabase());
|
||||
|
||||
// Provide default values:
|
||||
if (!nscfg.exists("cache_size"))
|
||||
nscfg.set("cache_size", std::to_string(app_.config().getValueFor(SizedItem::treeCacheSize, std::nullopt)));
|
||||
|
||||
if (!nscfg.exists("cache_age"))
|
||||
nscfg.set("cache_age", std::to_string(app_.config().getValueFor(SizedItem::treeCacheAge, std::nullopt)));
|
||||
|
||||
std::unique_ptr<NodeStore::Database> db;
|
||||
|
||||
if (deleteInterval_)
|
||||
@@ -218,6 +226,8 @@ SHAMapStoreImp::run()
|
||||
LedgerIndex lastRotated = state_db_.getState().lastRotated;
|
||||
netOPs_ = &app_.getOPs();
|
||||
ledgerMaster_ = &app_.getLedgerMaster();
|
||||
fullBelowCache_ = &(*app_.getNodeFamily().getFullBelowCache());
|
||||
treeNodeCache_ = &(*app_.getNodeFamily().getTreeNodeCache());
|
||||
|
||||
if (advisoryDelete_)
|
||||
canDelete_ = state_db_.getCanDelete();
|
||||
@@ -254,13 +264,25 @@ SHAMapStoreImp::run()
|
||||
bool const readyToRotate =
|
||||
validatedSeq >= lastRotated + deleteInterval_ && canDelete_ >= lastRotated - 1 && healthWait() == keepGoing;
|
||||
|
||||
JLOG(journal_.debug()) << "run: Setting lastGoodValidatedLedger_ to " << validatedSeq;
|
||||
|
||||
{
|
||||
// Note that this is set after the healthWait() check, so that we
|
||||
// don't start the rotation until the validated ledger is fully
|
||||
// processed. It is not guaranteed to be done at this point. It also
|
||||
// allows the testLedgerGaps unit test to work.
|
||||
std::unique_lock<std::mutex> lock(mutex_);
|
||||
lastGoodValidatedLedger_ = validatedSeq;
|
||||
}
|
||||
|
||||
// will delete up to (not including) lastRotated
|
||||
if (readyToRotate)
|
||||
{
|
||||
JLOG(journal_.warn()) << "rotating validatedSeq " << validatedSeq << " lastRotated " << lastRotated
|
||||
<< " deleteInterval " << deleteInterval_ << " canDelete_ " << canDelete_ << " state "
|
||||
<< app_.getOPs().strOperatingMode(false) << " age "
|
||||
<< ledgerMaster_->getValidatedLedgerAge().count() << 's';
|
||||
<< ledgerMaster_->getValidatedLedgerAge().count()
|
||||
<< "s. Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
|
||||
|
||||
clearPrior(lastRotated);
|
||||
if (healthWait() == stopping)
|
||||
@@ -313,7 +335,9 @@ SHAMapStoreImp::run()
|
||||
clearCaches(validatedSeq);
|
||||
});
|
||||
|
||||
JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
|
||||
JLOG(journal_.warn()) << "finished rotation. validatedSeq: " << validatedSeq
|
||||
<< ", lastRotated: " << lastRotated
|
||||
<< ". Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -480,12 +504,16 @@ void
|
||||
SHAMapStoreImp::clearCaches(LedgerIndex validatedSeq)
|
||||
{
|
||||
ledgerMaster_->clearLedgerCachePrior(validatedSeq);
|
||||
fullBelowCache_->clear();
|
||||
}
|
||||
|
||||
void
|
||||
SHAMapStoreImp::freshenCaches()
|
||||
{
|
||||
freshenCache(*app_.getNodeFamily().getTreeNodeCache()) && freshenCache(app_.getMasterTransaction().getCache());
|
||||
if (freshenCache(*treeNodeCache_))
|
||||
return;
|
||||
if (freshenCache(app_.getMasterTransaction().getCache()))
|
||||
return;
|
||||
}
|
||||
|
||||
void
|
||||
@@ -536,21 +564,37 @@ SHAMapStoreImp::clearPrior(LedgerIndex lastRotated)
|
||||
SHAMapStoreImp::HealthResult
|
||||
SHAMapStoreImp::healthWait()
|
||||
{
|
||||
auto index = ledgerMaster_->getValidLedgerIndex();
|
||||
auto age = ledgerMaster_->getValidatedLedgerAge();
|
||||
OperatingMode mode = netOPs_->getOperatingMode();
|
||||
std::unique_lock lock(mutex_);
|
||||
while (!stop_ && (mode != OperatingMode::FULL || age > ageThreshold_))
|
||||
|
||||
auto numMissing = ledgerMaster_->missingFromCompleteLedgerRange(lastGoodValidatedLedger_, index);
|
||||
while (!stop_ && (mode != OperatingMode::FULL || age > ageThreshold_ || numMissing > 0))
|
||||
{
|
||||
// this value shouldn't change, so grab it while we have the
|
||||
// lock
|
||||
auto const lastGood = lastGoodValidatedLedger_;
|
||||
|
||||
lock.unlock();
|
||||
JLOG(journal_.warn()) << "Waiting " << recoveryWaitTime_.count()
|
||||
<< "s for node to stabilize. state: " << app_.getOPs().strOperatingMode(mode, false)
|
||||
<< ". age " << age.count() << 's';
|
||||
|
||||
auto const stream = mode != OperatingMode::FULL || age > ageThreshold_ ? journal_.warn() : journal_.info();
|
||||
JLOG(stream) << "Waiting " << recoveryWaitTime_.count()
|
||||
<< "s for node to stabilize. state: " << app_.getOPs().strOperatingMode(mode, false) << ". age "
|
||||
<< age.count() << "s. Missing ledgers: " << numMissing << ". Expect: " << lastGood << "-" << index
|
||||
<< ". Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
|
||||
std::this_thread::sleep_for(recoveryWaitTime_);
|
||||
index = ledgerMaster_->getValidLedgerIndex();
|
||||
age = ledgerMaster_->getValidatedLedgerAge();
|
||||
mode = netOPs_->getOperatingMode();
|
||||
numMissing = ledgerMaster_->missingFromCompleteLedgerRange(lastGood, index);
|
||||
|
||||
lock.lock();
|
||||
}
|
||||
|
||||
JLOG(journal_.debug()) << "healthWait: Setting lastGoodValidatedLedger_ to " << index;
|
||||
lastGoodValidatedLedger_ = index;
|
||||
|
||||
return stop_ ? stopping : keepGoing;
|
||||
}
|
||||
|
||||
|
||||
@@ -71,6 +71,11 @@ private:
|
||||
std::thread thread_;
|
||||
bool stop_ = false;
|
||||
bool healthy_ = true;
|
||||
// Used to prevent ledger gaps from forming during online deletion. Keeps
|
||||
// track of the last validated ledger that was processed without gaps. There
|
||||
// are no guarantees about gaps while online delete is not running. For
|
||||
// that, use advisory_delete and check for gaps externally.
|
||||
LedgerIndex lastGoodValidatedLedger_ = 0;
|
||||
mutable std::condition_variable cond_;
|
||||
mutable std::condition_variable rendezvous_;
|
||||
mutable std::mutex mutex_;
|
||||
@@ -84,16 +89,18 @@ private:
|
||||
std::uint32_t deleteBatch_ = 100;
|
||||
std::chrono::milliseconds backOff_{100};
|
||||
std::chrono::seconds ageThreshold_{60};
|
||||
/// If the node is out of sync during an online_delete healthWait()
|
||||
/// call, sleep the thread for this time, and continue checking until
|
||||
/// recovery.
|
||||
/// If the node is out of sync, or any recent ledgers are not
|
||||
/// available during an online_delete healthWait() call, sleep
|
||||
/// the thread for this time, and continue checking until recovery.
|
||||
/// See also: "recovery_wait_seconds" in xrpld-example.cfg
|
||||
std::chrono::seconds recoveryWaitTime_{5};
|
||||
std::chrono::seconds recoveryWaitTime_{1};
|
||||
|
||||
// these do not exist upon SHAMapStore creation, but do exist
|
||||
// as of run() or before
|
||||
NetworkOPs* netOPs_ = nullptr;
|
||||
LedgerMaster* ledgerMaster_ = nullptr;
|
||||
FullBelowCache* fullBelowCache_ = nullptr;
|
||||
TreeNodeCache* treeNodeCache_ = nullptr;
|
||||
|
||||
static constexpr auto nodeStoreName_ = "NodeStore";
|
||||
|
||||
@@ -205,6 +212,8 @@ private:
|
||||
enum HealthResult { stopping, keepGoing };
|
||||
[[nodiscard]] HealthResult
|
||||
healthWait();
|
||||
bool
|
||||
hasCompleteRange(LedgerIndex first, LedgerIndex last);
|
||||
|
||||
public:
|
||||
void
|
||||
|
||||
Reference in New Issue
Block a user