ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2025-11-27 22:45:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

SHAMapStore Online Delete (RIPD-415):

Browse Source 
Makes rippled configurable to support deletion of all data in its key-value
store (nodestore) and ledger and transaction SQLite databases based on
validated ledger sequence numbers. All records from a specified ledger
and forward shall remain available in the key-value store and SQLite, and
all data prior to that specific ledger may be deleted.

Additionally, the administrator may require that an RPC command be
executed to enable deletion. This is to align data deletion with local
policy.
This commit is contained in:
Mark Travis
2014-10-22 11:52:53 -07:00
committed by Nik Bougalis
parent b44974677e
commit 02529a0fc2
50 changed files with 1851 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
Builds/VisualStudio2013/RippleD.vcxproj
View File

@@ -1852,6 +1852,13 @@
</ClCompile>
<ClInclude Include="..\..\src\ripple\app\misc\ProofOfWorkFactory.h">
</ClInclude>
<ClInclude Include="..\..\src\ripple\app\misc\SHAMapStore.h">
</ClInclude>
<ClCompile Include="..\..\src\ripple\app\misc\SHAMapStoreImp.cpp">
<ExcludedFromBuild>True</ExcludedFromBuild>
</ClCompile>
<ClInclude Include="..\..\src\ripple\app\misc\SHAMapStoreImp.h">
</ClInclude>
<ClCompile Include="..\..\src\ripple\app\misc\Validations.cpp">
<ExcludedFromBuild>True</ExcludedFromBuild>
</ClCompile>
@@ -2413,6 +2420,8 @@
</ClInclude>
<ClInclude Include="..\..\src\ripple\nodestore\Database.h">
</ClInclude>
<ClInclude Include="..\..\src\ripple\nodestore\DatabaseRotating.h">
</ClInclude>
<ClInclude Include="..\..\src\ripple\nodestore\DummyScheduler.h">
</ClInclude>
<ClInclude Include="..\..\src\ripple\nodestore\Factory.h">
@@ -2430,6 +2439,11 @@
</ClCompile>
<ClInclude Include="..\..\src\ripple\nodestore\impl\DatabaseImp.h">
</ClInclude>
<ClCompile Include="..\..\src\ripple\nodestore\impl\DatabaseRotatingImp.cpp">
<ExcludedFromBuild>True</ExcludedFromBuild>
</ClCompile>
<ClInclude Include="..\..\src\ripple\nodestore\impl\DatabaseRotatingImp.h">
</ClInclude>
<ClCompile Include="..\..\src\ripple\nodestore\impl\DecodedBlob.cpp">
<ExcludedFromBuild>True</ExcludedFromBuild>
</ClCompile>
@@ -2829,6 +2843,9 @@
<ClCompile Include="..\..\src\ripple\rpc\handlers\BookOffers.cpp">
<ExcludedFromBuild>True</ExcludedFromBuild>
</ClCompile>
<ClCompile Include="..\..\src\ripple\rpc\handlers\CanDelete.cpp">
<ExcludedFromBuild>True</ExcludedFromBuild>
</ClCompile>
<ClCompile Include="..\..\src\ripple\rpc\handlers\Connect.cpp">
<ExcludedFromBuild>True</ExcludedFromBuild>
</ClCompile>

21
Builds/VisualStudio2013/RippleD.vcxproj.filters
View File

@@ -2742,6 +2742,15 @@
<ClInclude Include="..\..\src\ripple\app\misc\ProofOfWorkFactory.h">
<Filter>ripple\app\misc</Filter>
</ClInclude>
<ClInclude Include="..\..\src\ripple\app\misc\SHAMapStore.h">
<Filter>ripple\app\misc</Filter>
</ClInclude>
<ClCompile Include="..\..\src\ripple\app\misc\SHAMapStoreImp.cpp">
<Filter>ripple\app\misc</Filter>
</ClCompile>
<ClInclude Include="..\..\src\ripple\app\misc\SHAMapStoreImp.h">
<Filter>ripple\app\misc</Filter>
</ClInclude>
<ClCompile Include="..\..\src\ripple\app\misc\Validations.cpp">
<Filter>ripple\app\misc</Filter>
</ClCompile>
@@ -3420,6 +3429,9 @@
<ClInclude Include="..\..\src\ripple\nodestore\Database.h">
<Filter>ripple\nodestore</Filter>
</ClInclude>
<ClInclude Include="..\..\src\ripple\nodestore\DatabaseRotating.h">
<Filter>ripple\nodestore</Filter>
</ClInclude>
<ClInclude Include="..\..\src\ripple\nodestore\DummyScheduler.h">
<Filter>ripple\nodestore</Filter>
</ClInclude>
@@ -3441,6 +3453,12 @@
<ClInclude Include="..\..\src\ripple\nodestore\impl\DatabaseImp.h">
<Filter>ripple\nodestore\impl</Filter>
</ClInclude>
<ClCompile Include="..\..\src\ripple\nodestore\impl\DatabaseRotatingImp.cpp">
<Filter>ripple\nodestore\impl</Filter>
</ClCompile>
<ClInclude Include="..\..\src\ripple\nodestore\impl\DatabaseRotatingImp.h">
<Filter>ripple\nodestore\impl</Filter>
</ClInclude>
<ClCompile Include="..\..\src\ripple\nodestore\impl\DecodedBlob.cpp">
<Filter>ripple\nodestore\impl</Filter>
</ClCompile>
@@ -3924,6 +3942,9 @@
<ClCompile Include="..\..\src\ripple\rpc\handlers\BookOffers.cpp">
<Filter>ripple\rpc\handlers</Filter>
</ClCompile>
<ClCompile Include="..\..\src\ripple\rpc\handlers\CanDelete.cpp">
<Filter>ripple\rpc\handlers</Filter>
</ClCompile>
<ClCompile Include="..\..\src\ripple\rpc\handlers\Connect.cpp">
<Filter>ripple\rpc\handlers</Filter>
</ClCompile>

11
doc/rippled-example.cfg
View File

@@ -675,6 +675,13 @@
#
# Optional keys:
# compression 0 for none, 1 for Snappy compression
# 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.
# advisory_delete 0 for disabled, 1 for enabled. If set, then
# require administrative RPC call "can_delete"
# to enable online deletion of ledger records.
#
# Notes:
# The 'node_db' entry configures the primary, persistent storage.
@@ -890,6 +897,8 @@ medium
# This is primary persistent datastore for rippled. This includes transaction
# metadata, account states, and ledger headers. Helpful information can be
# found here: https://ripple.com/wiki/NodeBackEnd
# delete old ledgers while maintaining at least 2000. Do not require an
# external administrative command to initiate deletion.
[node_db]
type=RocksDB
path=/var/lib/rippled/db/rocksdb
@@ -898,6 +907,8 @@ filter_bits=12
cache_mb=256
file_size_mb=8
file_size_mult=2
online_delete=2000
advisory_delete=0
[database_path]
/var/lib/rippled/db

35
src/ripple/app/data/DatabaseCon.cpp
View File

@@ -18,24 +18,21 @@
//==============================================================================
#include <ripple/app/data/DatabaseCon.h>
#include <ripple/core/Config.h>
namespace ripple {
DatabaseCon::DatabaseCon (std::string const& strName, const char* initStrings[], int initCount)
DatabaseCon::DatabaseCon (Setup const& setup,
std::string const& strName,
const char* initStrings[],
int initCount)
{
// VFALCO TODO remove this dependency on the config by making it the caller's
// responsibility to pass in the path. Add a member function to Application
// or Config to compute this path.
//
auto const startUp = getConfig ().START_UP;
auto const useTempFiles // Use temporary files or regular DB files?
= getConfig ().RUN_STANDALONE &&
startUp != Config::LOAD &&
startUp != Config::LOAD_FILE &&
startUp != Config::REPLAY;
= setup.standAlone &&
setup.startUp != Config::LOAD &&
setup.startUp != Config::LOAD_FILE &&
setup.startUp != Config::REPLAY;
boost::filesystem::path pPath = useTempFiles
? "" : (getConfig ().DATA_DIR / strName);
? "" : (setup.dataDir / strName);
mDatabase = new SqliteDatabase (pPath.string ().c_str ());
mDatabase->connect ();
@@ -50,4 +47,18 @@ DatabaseCon::~DatabaseCon ()
delete mDatabase;
}
DatabaseCon::Setup
setup_DatabaseCon (Config const& c)
{
DatabaseCon::Setup setup;
if (c.nodeDatabase["online_delete"].isNotEmpty())
setup.onlineDelete = c.nodeDatabase["online_delete"].getIntValue();
setup.startUp = c.START_UP;
setup.standAlone = c.RUN_STANDALONE;
setup.dataDir = c.DATA_DIR;
return setup;
}
} // ripple

27
src/ripple/app/data/DatabaseCon.h
View File

@@ -20,6 +20,8 @@
#ifndef RIPPLE_DATABASECON_H
#define RIPPLE_DATABASECON_H
#include <ripple/core/Config.h>
#include <mutex>
#include <string>
@@ -32,7 +34,18 @@ class Database;
class DatabaseCon
{
public:
DatabaseCon (std::string const& name, const char* initString[], int countInit);
struct Setup
{
bool onlineDelete = false;
Config::StartUpType startUp = Config::NORMAL;
bool standAlone = false;
boost::filesystem::path dataDir;
};
DatabaseCon (Setup const& setup,
std::string const& name,
const char* initString[],
int countInit);
~DatabaseCon ();
Database* getDB ()
@@ -42,16 +55,26 @@ public:
typedef std::recursive_mutex mutex;
std::unique_lock<mutex> lock()
std::unique_lock<mutex> lock ()
{
return std::unique_lock<mutex>(mLock);
}
mutex& peekMutex()
{
return mLock;
}
private:
Database* mDatabase;
mutex mLock;
};
//------------------------------------------------------------------------------
DatabaseCon::Setup
setup_DatabaseCon (Config const& c);
} // ripple
#endif

9
src/ripple/app/ledger/LedgerHistory.cpp
View File

@@ -294,4 +294,13 @@ void LedgerHistory::tune (int size, int age)
m_ledgers_by_hash.setTargetAge (age);
}
void LedgerHistory::clearLedgerCachePrior (LedgerIndex seq)
{
for (LedgerHash it: m_ledgers_by_hash.getKeys())
{
if (getLedgerByHash (it)->getLedgerSeq() < seq)
m_ledgers_by_hash.del (it, false);
}
}
} // ripple

2
src/ripple/app/ledger/LedgerHistory.h
View File

@@ -93,6 +93,8 @@ public:
*/
bool fixIndex(LedgerIndex ledgerIndex, LedgerHash const& ledgerHash);
void clearLedgerCachePrior (LedgerIndex seq);
private:
/** Log details in the case where we build one ledger but

19
src/ripple/app/ledger/LedgerMaster.cpp
View File

@@ -24,6 +24,7 @@
#include <ripple/app/ledger/OrderBookDB.h>
#include <ripple/app/paths/PathRequests.h>
#include <ripple/validators/Manager.h>
#include <ripple/app/misc/SHAMapStore.h>
#include <algorithm>
#include <cassert>
#include <beast/cxx14/memory.h> // <memory>
@@ -116,7 +117,7 @@ public:
, mValidLedgerSeq (0)
, mBuildingLedgerSeq (0)
, standalone_ (standalone)
, fetch_depth_ (fetch_depth)
, fetch_depth_ (getApp().getSHAMapStore().clampFetchDepth (fetch_depth))
, ledger_history_ (ledger_history)
, ledger_fetch_size_ (ledger_fetch_size)
{
@@ -198,6 +199,7 @@ public:
mValidLedgerClose = l->getCloseTimeNC();
mValidLedgerSeq = l->getLedgerSeq();
getApp().getOPs().updateLocalTx (l);
getApp().getSHAMapStore().onLedgerClosed (getValidatedLedger());
}
void setPubLedger(Ledger::ref l)
@@ -1483,6 +1485,21 @@ public:
{
return *mLedgerCleaner;
}
void clearPriorLedgers (LedgerIndex seq) override
{
ScopedLockType sl (mCompleteLock);
for (LedgerIndex i = mCompleteLedgers.getFirst(); i < seq; ++i)
{
if (haveLedger (i))
clearLedger (i);
}
}
void clearLedgerCachePrior (LedgerIndex seq) override
{
mLedgerHistory.clearLedgerCachePrior (seq);
}
};
//------------------------------------------------------------------------------

4
src/ripple/app/ledger/LedgerMaster.h
View File

@@ -149,6 +149,10 @@ public:
static bool shouldAcquire (std::uint32_t currentLedgerID,
std::uint32_t ledgerHistory, std::uint32_t targetLedger);
virtual void clearPriorLedgers (LedgerIndex seq) = 0;
virtual void clearLedgerCachePrior (LedgerIndex seq) = 0;
};
std::unique_ptr <LedgerMaster>

58
src/ripple/app/main/Application.cpp
View File

@@ -26,6 +26,7 @@
#include <ripple/basics/Sustain.h>
#include <ripple/basics/seconds_clock.h>
#include <ripple/basics/make_SSLContext.h>
#include <ripple/app/misc/SHAMapStore.h>
#include <ripple/core/LoadFeeTrack.h>
#include <ripple/net/SNTPClient.h>
#include <ripple/nodestore/Database.h>
@@ -153,9 +154,12 @@ public:
beast::Journal m_journal;
Application::LockType m_masterMutex;
// These are not Stoppable-derived
std::unique_ptr <NodeStore::Manager> m_nodeStoreManager;
NodeStoreScheduler m_nodeStoreScheduler;
std::unique_ptr <SHAMapStore> m_shaMapStore;
std::unique_ptr <NodeStore::Database> m_nodeStore;
// These are not Stoppable-derived
NodeCache m_tempNodeCache;
TreeNodeCache m_treeNodeCache;
SLECache m_sleCache;
@@ -167,7 +171,6 @@ public:
std::unique_ptr <FullBelowCache> m_fullBelowCache;
// These are Stoppable-related
NodeStoreScheduler m_nodeStoreScheduler;
std::unique_ptr <JobQueue> m_jobQueue;
std::unique_ptr <SiteFiles::Manager> m_siteFiles;
std::unique_ptr <RPC::Manager> m_rpcManager;
@@ -179,7 +182,6 @@ public:
std::unique_ptr <NetworkOPs> m_networkOPs;
std::unique_ptr <UniqueNodeList> m_deprecatedUNL;
std::unique_ptr <ServerHandler> serverHandler_;
std::unique_ptr <NodeStore::Database> m_nodeStore;
std::unique_ptr <SNTPClient> m_sntpClient;
std::unique_ptr <Validators::Manager> m_validators;
std::unique_ptr <AmendmentTable> m_amendmentTable;
@@ -243,6 +245,15 @@ public:
std::move (make_Factories (
getConfig ().getSize(siHashNodeDBCache) * 1024))))
, m_nodeStoreScheduler (*this)
, m_shaMapStore (make_SHAMapStore (setup_SHAMapStore (
getConfig()), *this, *m_nodeStoreManager.get(),
m_nodeStoreScheduler, m_logs.journal ("SHAMapStore"),
m_logs.journal ("NodeObject"), m_txMaster))
, m_nodeStore (m_shaMapStore->makeDatabase ("NodeStore.main", 4))
, m_tempNodeCache ("NodeCache", 16384, 90, get_seconds_clock (),
m_logs.journal("TaggedCache"))
@@ -262,8 +273,6 @@ public:
"full_below", get_seconds_clock (), m_collectorManager->collector (),
fullBelowTargetSize, fullBelowExpirationSeconds))
, m_nodeStoreScheduler (*this)
// The JobQueue has to come pretty early since
// almost everything is a Stoppable child of the JobQueue.
//
@@ -307,11 +316,6 @@ public:
get_io_service(), *m_jobQueue, *m_networkOPs,
*m_resourceManager))
, m_nodeStore (m_nodeStoreManager->make_Database ("NodeStore.main",
m_nodeStoreScheduler, m_logs.journal("NodeObject"),
4, // four read threads for now
getConfig ().nodeDatabase, getConfig ().ephemeralNodeDatabase))
, m_sntpClient (SNTPClient::New (*this))
, m_validators (add (Validators::Manager::New (
@@ -509,6 +513,11 @@ public:
return *mProofOfWorkFactory;
}
SHAMapStore& getSHAMapStore () override
{
return *m_shaMapStore;
}
Overlay& overlay ()
{
return *m_overlay;
@@ -559,10 +568,29 @@ public:
assert (mLedgerDB.get () == nullptr);
assert (mWalletDB.get () == nullptr);
mRpcDB = std::make_unique <DatabaseCon> ("rpc.db", RpcDBInit, RpcDBCount);
mTxnDB = std::make_unique <DatabaseCon> ("transaction.db", TxnDBInit, TxnDBCount);
mLedgerDB = std::make_unique <DatabaseCon> ("ledger.db", LedgerDBInit, LedgerDBCount);
mWalletDB = std::make_unique <DatabaseCon> ("wallet.db", WalletDBInit, WalletDBCount);
DatabaseCon::Setup setup = setup_DatabaseCon (getConfig());
mRpcDB = std::make_unique <DatabaseCon> (setup, "rpc.db", RpcDBInit,
RpcDBCount);
mTxnDB = std::make_unique <DatabaseCon> (setup, "transaction.db",
TxnDBInit, TxnDBCount);
mLedgerDB = std::make_unique <DatabaseCon> (setup, "ledger.db",
LedgerDBInit, LedgerDBCount);
mWalletDB = std::make_unique <DatabaseCon> (setup, "wallet.db",
WalletDBInit, WalletDBCount);
if (setup.onlineDelete && mTxnDB && mLedgerDB)
{
{
std::lock_guard <std::recursive_mutex> lock (
mTxnDB->peekMutex());
mTxnDB->getDB()->executeSQL ("VACUUM;");
}
{
std::lock_guard <std::recursive_mutex> lock (
mLedgerDB->peekMutex());
mLedgerDB->getDB()->executeSQL ("VACUUM;");
}
}
return
mRpcDB.get() != nullptr &&
@@ -916,7 +944,7 @@ public:
&TransactionMaster::sweep, &m_txMaster));
logTimedCall (m_journal.warning, "NodeStore::sweep", __FILE__, __LINE__, std::bind (
&NodeStore::Database::sweep, m_nodeStore.get ()));
&NodeStore::Database::sweep, m_nodeStore.get()));
logTimedCall (m_journal.warning, "LedgerMaster::sweep", __FILE__, __LINE__, std::bind (
&LedgerMaster::sweep, m_ledgerMaster.get()));

2
src/ripple/app/main/Application.h
View File

@@ -58,6 +58,7 @@ class TransactionMaster;
class Validations;
class DatabaseCon;
class SHAMapStore;
using NodeCache = TaggedCache <uint256, Blob>;
using SLECache = TaggedCache <uint256, STLedgerEntry>;
@@ -119,6 +120,7 @@ public:
virtual LocalCredentials& getLocalCredentials () = 0;
virtual Resource::Manager& getResourceManager () = 0;
virtual PathRequests& getPathRequests () = 0;
virtual SHAMapStore& getSHAMapStore () = 0;
virtual DatabaseCon& getRpcDB () = 0;
virtual DatabaseCon& getTxnDB () = 0;

14
src/ripple/app/main/BasicFullBelowCache.h
View File

@@ -54,6 +54,7 @@ public:
std::size_t expiration_seconds = defaultCacheExpirationSeconds)
: m_cache (name, clock, collector, target_size,
expiration_seconds)
, m_gen (1)
{
}
@@ -104,8 +105,21 @@ public:
m_cache.insert (key);
}
/** generation determines whether cached entry is valid */
std::uint32_t getGeneration (void) const
{
return m_gen;
}
void clear ()
{
m_cache.clear ();
++m_gen;
}
private:
KeyCache <Key> m_cache;
std::atomic <std::uint32_t> m_gen;
};
}

1
src/ripple/app/main/Main.cpp
View File

@@ -92,6 +92,7 @@ void printHelp (const po::options_description& desc)
" account_offers <account>|<account_public_key> [<ledger>]\n"
" account_tx accountID [ledger_min [ledger_max [limit [offset]]]] [binary] [count] [descending]\n"
" book_offers <taker_pays> <taker_gets> [<taker [<ledger> [<limit> [<proof> [<marker>]]]]]\n"
" can_delete [<ledgerid>|<ledgerhash>|now|always|never]\n"
" connect <ip> [<port>]\n"
" consensus_info\n"
" get_counts\n"

48
src/ripple/app/misc/README.md
View File

@@ -89,3 +89,51 @@ Nodes may veto Amendments they consider undesirable by never announcing their
support for those Amendments. Just a few nodes vetoing an Amendment will normally
keep it from being accepted. Nodes could also vote yes on an Amendments even
before it obtains a super-majority. This might make sense for a critical bug fix.
---
# SHAMapStore: Online Delete
Optional online deletion happens through the SHAMapStore. Records are deleted
from disk based on ledger sequence number. These records reside in the
key-value database as well as in the SQLite ledger and transaction databases.
Without online deletion storage usage grows without bounds. It can only
be pruned by stopping, manually deleting data, and restarting the server.
Online deletion requires less operator intervention to manage the server.
The main mechanism to delete data from the key-value database is to keep two
databases open at all times. One database has all writes directed to it. The
other database has recent archival data from just prior to that from the current
writable database.
Upon rotation, the archival database is deleted. The writable database becomes
archival, and a brand new database becomes writable. To ensure that no
necessary data for transaction processing is lost, a variety of steps occur,
including copying the contents of an entire ledger's account state map,
clearing caches, and copying the contents of (freshening) other caches.
Deleting from SQLite involves more straight-forward SQL DELETE queries from
the respective tables, with a rudimentary back-off algorithm to do portions
of the deletions at a time. This back-off is in place so that the database
lock is not held excessively. The SQLite database is not configured to
delete on-disk storage, so it will grow over time. However, with online delete
enabled, it grows at a very small rate compared with the key-value store.
The online delete routine aborts its current activities if it fails periodic
server health checks. This minimizes impact of I/O and locking of critical
objects. If interrupted, the routine will start again at the next validated
ledger close. Likewise, the routine will continue in a similar fashion if the
server restarts.
Configuration:
* In the [node_db] configuration section, an optional online_delete parameter is
set. If not set or if set to 0, online delete is disabled. Otherwise, the
setting defines number of ledgers between deletion cycles.
* Another optional parameter in [node_db] is that for advisory_delete. It is
disabled by default. If set to non-zero, requires an RPC call to activate the
deletion routine.
* online_delete must not be greater than the [ledger_history] parameter.
* [fetch_depth] will be silently set to equal the online_delete setting if
online_delete is greater than fetch_depth.
* In the [node_db] section, there is a performance tuning option, delete_batch,
which sets the maximum size in ledgers for each SQL DELETE query.

87
src/ripple/app/misc/SHAMapStore.h Normal file
View File

@@ -0,0 +1,87 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_APP_SHAMAPSTORE_H_INCLUDED
#define RIPPLE_APP_SHAMAPSTORE_H_INCLUDED
#include <ripple/core/Config.h>
#include <ripple/nodestore/Manager.h>
#include <ripple/nodestore/Scheduler.h>
#include <ripple/rpc/ErrorCodes.h>
namespace ripple {
/**
* class to create database, launch online delete thread, and
* related sqlite databse
*/
class SHAMapStore
: public beast::Stoppable
{
public:
struct Setup
{
std::uint32_t deleteInterval = 0;
bool advisoryDelete = false;
std::uint32_t ledgerHistory = 0;
beast::StringPairArray nodeDatabase;
beast::StringPairArray ephemeralNodeDatabase;
std::string databasePath;
std::uint32_t deleteBatch = 1000;
};
SHAMapStore (Stoppable& parent) : Stoppable ("SHAMapStore", parent) {}
/** Called by LedgerMaster every time a ledger validates. */
virtual void onLedgerClosed (Ledger::pointer validatedLedger) = 0;
virtual std::uint32_t clampFetchDepth (std::uint32_t fetch_depth) const = 0;
virtual std::unique_ptr <NodeStore::Database> makeDatabase (
std::string const& name, std::int32_t readThreads) = 0;
/** Highest ledger that may be deleted. */
virtual LedgerIndex setCanDelete (LedgerIndex canDelete) = 0;
/** Whether advisory delete is enabled. */
virtual bool advisoryDelete() const = 0;
/** Last ledger which was copied during rotation of backends. */
virtual LedgerIndex getLastRotated() = 0;
/** Highest ledger that may be deleted. */
virtual LedgerIndex getCanDelete() = 0;
};
//------------------------------------------------------------------------------
SHAMapStore::Setup
setup_SHAMapStore(Config const& c);
std::unique_ptr<SHAMapStore>
make_SHAMapStore(SHAMapStore::Setup const& s,
beast::Stoppable& parent,
NodeStore::Manager& manager,
NodeStore::Scheduler& scheduler,
beast::Journal journal,
beast::Journal nodeStoreJournal,
TransactionMaster& transactionMaster);
}
#endif

722
src/ripple/app/misc/SHAMapStoreImp.cpp Normal file
View File

@@ -0,0 +1,722 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include <ripple/app/misc/SHAMapStoreImp.h>
#include <beast/cxx14/memory.h> // <memory>
namespace ripple {
void
SHAMapStoreImp::SavedStateDB::init (std::string const& databasePath,
std::string const& dbName)
{
boost::filesystem::path pathName = databasePath;
pathName /= dbName;
std::lock_guard <std::mutex> lock (mutex_);
auto error = session_.open (pathName.string());
checkError (error);
session_.once (error) << "PRAGMA synchronous=FULL;";
checkError (error);
session_.once (error) <<
"CREATE TABLE IF NOT EXISTS DbState ("
" Key INTEGER PRIMARY KEY,"
" WritableDb TEXT,"
" ArchiveDb TEXT,"
" LastRotatedLedger INTEGER"
");"
;
checkError (error);
session_.once (error) <<
"CREATE TABLE IF NOT EXISTS CanDelete ("
" Key INTEGER PRIMARY KEY,"
" CanDeleteSeq INTEGER"
");"
;
std::int64_t count = 0;
beast::sqdb::statement st = (session_.prepare <<
"SELECT COUNT(Key) FROM DbState WHERE Key = 1;"
, beast::sqdb::into (count)
);
st.execute_and_fetch (error);
checkError (error);
if (!count)
{
session_.once (error) <<
"INSERT INTO DbState VALUES (1, '', '', 0);";
checkError (error);
}
st = (session_.prepare <<
"SELECT COUNT(Key) FROM CanDelete WHERE Key = 1;"
, beast::sqdb::into (count)
);
st.execute_and_fetch (error);
checkError (error);
if (!count)
{
session_.once (error) <<
"INSERT INTO CanDelete VALUES (1, 0);";
checkError (error);
}
}
LedgerIndex
SHAMapStoreImp::SavedStateDB::getCanDelete()
{
beast::Error error;
LedgerIndex seq;
{
std::lock_guard <std::mutex> lock (mutex_);
session_.once (error) <<
"SELECT CanDeleteSeq FROM CanDelete WHERE Key = 1;"
, beast::sqdb::into (seq);
;
}
checkError (error);
return seq;
}
LedgerIndex
SHAMapStoreImp::SavedStateDB::setCanDelete (LedgerIndex canDelete)
{
beast::Error error;
{
std::lock_guard <std::mutex> lock (mutex_);
session_.once (error) <<
"UPDATE CanDelete SET CanDeleteSeq = ? WHERE Key = 1;"
, beast::sqdb::use (canDelete)
;
}
checkError (error);
return canDelete;
}
SHAMapStoreImp::SavedState
SHAMapStoreImp::SavedStateDB::getState()
{
beast::Error error;
SavedState state;
{
std::lock_guard <std::mutex> lock (mutex_);
session_.once (error) <<
"SELECT WritableDb, ArchiveDb, LastRotatedLedger"
" FROM DbState WHERE Key = 1;"
, beast::sqdb::into (state.writableDb)
, beast::sqdb::into (state.archiveDb)
, beast::sqdb::into (state.lastRotated)
;
}
checkError (error);
return state;
}
void
SHAMapStoreImp::SavedStateDB::setState (SavedState const& state)
{
beast::Error error;
{
std::lock_guard <std::mutex> lock (mutex_);
session_.once (error) <<
"UPDATE DbState"
" SET WritableDb = ?,"
" ArchiveDb = ?,"
" LastRotatedLedger = ?"
" WHERE Key = 1;"
, beast::sqdb::use (state.writableDb)
, beast::sqdb::use (state.archiveDb)
, beast::sqdb::use (state.lastRotated)
;
}
checkError (error);
}
void
SHAMapStoreImp::SavedStateDB::setLastRotated (LedgerIndex seq)
{
beast::Error error;
{
std::lock_guard <std::mutex> lock (mutex_);
session_.once (error) <<
"UPDATE DbState SET LastRotatedLedger = ?"
" WHERE Key = 1;"
, beast::sqdb::use (seq)
;
}
checkError (error);
}
void
SHAMapStoreImp::SavedStateDB::checkError (beast::Error const& error)
{
if (error)
{
journal_.fatal << "state database error: " << error.code()
<< ": " << error.getReasonText();
throw std::runtime_error ("State database error.");
}
}
SHAMapStoreImp::SHAMapStoreImp (Setup const& setup,
Stoppable& parent,
NodeStore::Manager& manager,
NodeStore::Scheduler& scheduler,
beast::Journal journal,
beast::Journal nodeStoreJournal,
TransactionMaster& transactionMaster)
: SHAMapStore (parent)
, setup_ (setup)
, manager_ (manager)
, scheduler_ (scheduler)
, journal_ (journal)
, nodeStoreJournal_ (nodeStoreJournal)
, database_ (nullptr)
, transactionMaster_ (transactionMaster)
{
if (setup_.deleteInterval)
{
if (setup_.ledgerHistory > setup_.deleteInterval ||
setup_.ledgerHistory < minimumDeletionInterval_)
{
std::stringstream es;
es << "online_delete (" << setup_.deleteInterval
<< ") must be at least " << minimumDeletionInterval_
<< " and cannot be less than LEDGER_HISTORY ("
<< setup_.ledgerHistory << ")";
throw std::runtime_error (es.str());
}
state_db_.init (setup_.databasePath, dbName_);
dbPaths();
}
}
std::unique_ptr <NodeStore::Database>
SHAMapStoreImp::makeDatabase (std::string const& name,
std::int32_t readThreads)
{
std::unique_ptr <NodeStore::Database> db;
if (setup_.deleteInterval)
{
SavedState state = state_db_.getState();
std::shared_ptr <NodeStore::Backend> writableBackend (
makeBackendRotating (state.writableDb));
std::shared_ptr <NodeStore::Backend> archiveBackend (
makeBackendRotating (state.archiveDb));
std::unique_ptr <NodeStore::DatabaseRotating> dbr =
makeDatabaseRotating (name, readThreads, writableBackend,
archiveBackend);
if (!state.writableDb.size())
{
state.writableDb = writableBackend->getName();
state.archiveDb = archiveBackend->getName();
state_db_.setState (state);
}
database_ = dbr.get();
db.reset (dynamic_cast <NodeStore::Database*>(dbr.release()));
}
else
{
db = manager_.make_Database (name, scheduler_, nodeStoreJournal_,
readThreads, setup_.nodeDatabase,
setup_.ephemeralNodeDatabase);
}
return db;
}
void
SHAMapStoreImp::onLedgerClosed (Ledger::pointer validatedLedger)
{
{
std::lock_guard <std::mutex> lock (mutex_);
newLedger_ = validatedLedger;
}
cond_.notify_one();
}
bool
SHAMapStoreImp::copyNode (std::uint64_t& nodeCount,
SHAMapTreeNode const& node)
{
// Copy a single record from node to database_
database_->fetchNode (node.getNodeHash());
if (! (++nodeCount % checkHealthInterval_))
{
if (health())
return true;
}
return false;
}
void
SHAMapStoreImp::run()
{
LedgerIndex lastRotated = state_db_.getState().lastRotated;
netOPs_ = &getApp().getOPs();
ledgerMaster_ = &getApp().getLedgerMaster();
fullBelowCache_ = &getApp().getFullBelowCache();
treeNodeCache_ = &getApp().getTreeNodeCache();
transactionDb_ = &getApp().getTxnDB();
ledgerDb_ = &getApp().getLedgerDB();
while (1)
{
healthy_ = true;
validatedLedger_.reset();
std::unique_lock <std::mutex> lock (mutex_);
if (stop_)
{
stopped();
return;
}
cond_.wait (lock);
if (newLedger_)
validatedLedger_ = std::move (newLedger_);
else
continue;
lock.unlock();
LedgerIndex validatedSeq = validatedLedger_->getLedgerSeq();
if (!lastRotated)
{
lastRotated = validatedSeq;
state_db_.setLastRotated (lastRotated);
}
LedgerIndex canDelete = std::numeric_limits <LedgerIndex>::max();
if (setup_.advisoryDelete)
canDelete = state_db_.getCanDelete();
// will delete up to (not including) lastRotated)
if (validatedSeq >= lastRotated + setup_.deleteInterval
&& canDelete >= lastRotated - 1)
{
journal_.debug << "rotating validatedSeq " << validatedSeq
<< " lastRotated " << lastRotated << " deleteInterval "
<< setup_.deleteInterval << " canDelete " << canDelete;
switch (health())
{
case Health::stopping:
stopped();
return;
case Health::unhealthy:
continue;
case Health::ok:
default:
;
}
clearPrior (lastRotated);
switch (health())
{
case Health::stopping:
stopped();
return;
case Health::unhealthy:
continue;
case Health::ok:
default:
;
}
std::uint64_t nodeCount = 0;
validatedLedger_->peekAccountStateMap()->snapShot (
false)->visitNodes (
std::bind (&SHAMapStoreImp::copyNode, this,
std::ref(nodeCount), std::placeholders::_1));
journal_.debug << "copied ledger " << validatedSeq
<< " nodecount " << nodeCount;
switch (health())
{
case Health::stopping:
stopped();
return;
case Health::unhealthy:
continue;
case Health::ok:
default:
;
}
freshenCaches();
journal_.debug << validatedSeq << " freshened caches";
switch (health())
{
case Health::stopping:
stopped();
return;
case Health::unhealthy:
continue;
case Health::ok:
default:
;
}
std::shared_ptr <NodeStore::Backend> newBackend =
makeBackendRotating();
journal_.debug << validatedSeq << " new backend "
<< newBackend->getName();
std::shared_ptr <NodeStore::Backend> oldBackend;
clearCaches (validatedSeq);
switch (health())
{
case Health::stopping:
stopped();
return;
case Health::unhealthy:
continue;
case Health::ok:
default:
;
}
std::string nextArchiveDir =
database_->getWritableBackend()->getName();
lastRotated = validatedSeq;
{
std::lock_guard <std::mutex> lock (database_->peekMutex());
state_db_.setState (SavedState {newBackend->getName(),
nextArchiveDir, lastRotated});
clearCaches (validatedSeq);
oldBackend = database_->rotateBackends (newBackend);
}
journal_.debug << "finished rotation " << validatedSeq;
oldBackend->setDeletePath();
}
}
}
void
SHAMapStoreImp::dbPaths()
{
boost::filesystem::path dbPath =
setup_.nodeDatabase["path"].toStdString();
if (boost::filesystem::exists (dbPath))
{
if (! boost::filesystem::is_directory (dbPath))
{
std::cerr << "node db path must be a directory. "
<< dbPath.string();
throw std::runtime_error (
"node db path must be a directory.");
}
}
else
{
boost::filesystem::create_directories (dbPath);
}
SavedState state = state_db_.getState();
bool writableDbExists = false;
bool archiveDbExists = false;
for (boost::filesystem::directory_iterator it (dbPath);
it != boost::filesystem::directory_iterator(); ++it)
{
if (! state.writableDb.compare (it->path().string()))
writableDbExists = true;
else if (! state.archiveDb.compare (it->path().string()))
archiveDbExists = true;
else if (! dbPrefix_.compare (it->path().stem().string()))
boost::filesystem::remove_all (it->path());
}
if ((!writableDbExists && state.writableDb.size()) ||
(!archiveDbExists && state.archiveDb.size()) ||
(writableDbExists != archiveDbExists) ||
state.writableDb.empty() != state.archiveDb.empty())
{
boost::filesystem::path stateDbPathName = setup_.databasePath;
stateDbPathName /= dbName_;
stateDbPathName += "*";
std::cerr << "state db error: " << std::endl
<< " writableDbExists " << writableDbExists
<< " archiveDbExists " << archiveDbExists << std::endl
<< " writableDb '" << state.writableDb
<< "' archiveDb '" << state.archiveDb << "'"
<< std::endl << std::endl
<< "To resume operation, make backups of and "
<< "remove the files matching "
<< stateDbPathName.string()
<< " and contents of the directory "
<< setup_.nodeDatabase["path"].toStdString()
<< std::endl;
throw std::runtime_error ("state db error");
}
}
std::shared_ptr <NodeStore::Backend>
SHAMapStoreImp::makeBackendRotating (std::string path)
{
boost::filesystem::path newPath;
NodeStore::Parameters parameters = setup_.nodeDatabase;
if (path.size())
{
newPath = path;
}
else
{
boost::filesystem::path p = parameters["path"].toStdString();
p /= dbPrefix_;
p += ".%%%%";
newPath = boost::filesystem::unique_path (p);
}
parameters.set("path", newPath.string());
return manager_.make_Backend (parameters, scheduler_,
nodeStoreJournal_);
}
std::unique_ptr <NodeStore::DatabaseRotating>
SHAMapStoreImp::makeDatabaseRotating (std::string const& name,
std::int32_t readThreads,
std::shared_ptr <NodeStore::Backend> writableBackend,
std::shared_ptr <NodeStore::Backend> archiveBackend) const
{
std::unique_ptr <NodeStore::Backend> fastBackend (
(setup_.ephemeralNodeDatabase.size() > 0)
? manager_.make_Backend (setup_.ephemeralNodeDatabase,
scheduler_, journal_) : nullptr);
return manager_.make_DatabaseRotating ("NodeStore.main", scheduler_,
readThreads, writableBackend, archiveBackend,
std::move (fastBackend), nodeStoreJournal_);
}
void
SHAMapStoreImp::clearSql (DatabaseCon& database,
LedgerIndex lastRotated,
std::string const& minQuery,
std::string const& deleteQuery)
{
LedgerIndex min = std::numeric_limits <LedgerIndex>::max();
Database* db = database.getDB();
std::unique_lock <std::recursive_mutex> lock (database.peekMutex());
if (!db->executeSQL (minQuery) || !db->startIterRows())
return;
min = db->getBigInt (0);
db->endIterRows ();
lock.unlock();
if (health() != Health::ok)
return;
boost::format formattedDeleteQuery (deleteQuery);
journal_.debug << "start: " << deleteQuery << " from "
<< min << " to " << lastRotated;
while (min < lastRotated)
{
min = (min + setup_.deleteBatch >= lastRotated) ? lastRotated :
min + setup_.deleteBatch;
lock.lock();
db->executeSQL (boost::str (formattedDeleteQuery % min));
lock.unlock();
if (health())
return;
if (min < lastRotated)
std::this_thread::sleep_for (
std::chrono::microseconds (pause_));
}
journal_.debug << "finished: " << deleteQuery;
}
void
SHAMapStoreImp::clearCaches (LedgerIndex validatedSeq)
{
ledgerMaster_->clearLedgerCachePrior (validatedSeq);
fullBelowCache_->clear();
}
void
SHAMapStoreImp::freshenCaches()
{
if (freshenCache (database_->getPositiveCache()))
return;
if (freshenCache (*treeNodeCache_))
return;
if (freshenCache (transactionMaster_.getCache()))
return;
}
void
SHAMapStoreImp::clearPrior (LedgerIndex lastRotated)
{
ledgerMaster_->clearPriorLedgers (lastRotated);
if (health())
return;
// TODO This won't remove validations for ledgers that do not get
// validated. That will likely require inserting LedgerSeq into
// the validations table
clearSql (*ledgerDb_, lastRotated,
"SELECT MIN(LedgerSeq) FROM Ledgers;",
"DELETE FROM Validations WHERE Ledgers.LedgerSeq < %u"
" AND Validations.LedgerHash = Ledgers.LedgerHash;");
if (health())
return;
clearSql (*ledgerDb_, lastRotated,
"SELECT MIN(LedgerSeq) FROM Ledgers;",
"DELETE FROM Ledgers WHERE LedgerSeq < %u;");
if (health())
return;
clearSql (*transactionDb_, lastRotated,
"SELECT MIN(LedgerSeq) FROM Transactions;",
"DELETE FROM Transactions WHERE LedgerSeq < %u;");
if (health())
return;
clearSql (*transactionDb_, lastRotated,
"SELECT MIN(LedgerSeq) FROM AccountTransactions;",
"DELETE FROM AccountTransactions WHERE LedgerSeq < %u;");
if (health())
return;
}
SHAMapStoreImp::Health
SHAMapStoreImp::health()
{
{
std::lock_guard<std::mutex> lock (mutex_);
if (stop_)
return Health::stopping;
}
if (!netOPs_)
return Health::ok;
NetworkOPs::OperatingMode mode = netOPs_->getOperatingMode();
std::uint32_t age = netOPs_->getNetworkTimeNC() - (
validatedLedger_->getCloseTimeNC() -
validatedLedger_->getCloseResolution());
if (mode != NetworkOPs::omFULL || age >= ageTooHigh_)
{
journal_.warning << "server not healthy, not deleting. state: "
<< mode << " age " << age << " age threshold "
<< ageTooHigh_;
healthy_ = false;
}
if (healthy_)
return Health::ok;
else
return Health::unhealthy;
}
void
SHAMapStoreImp::onStop()
{
if (setup_.deleteInterval)
{
{
std::lock_guard <std::mutex> lock (mutex_);
stop_ = true;
}
cond_.notify_one();
}
else
{
stopped();
}
}
void
SHAMapStoreImp::onChildrenStopped()
{
if (setup_.deleteInterval)
{
{
std::lock_guard <std::mutex> lock (mutex_);
stop_ = true;
}
cond_.notify_one();
}
else
{
stopped();
}
}
//------------------------------------------------------------------------------
SHAMapStore::Setup
setup_SHAMapStore (Config const& c)
{
SHAMapStore::Setup setup;
if (c.nodeDatabase["online_delete"].isNotEmpty())
setup.deleteInterval = c.nodeDatabase["online_delete"].getIntValue();
if (c.nodeDatabase["advisory_delete"].isNotEmpty() && setup.deleteInterval)
setup.advisoryDelete = c.nodeDatabase["advisory_delete"].getIntValue();
setup.ledgerHistory = c.LEDGER_HISTORY;
setup.nodeDatabase = c.nodeDatabase;
setup.ephemeralNodeDatabase = c.ephemeralNodeDatabase;
setup.databasePath = c.DATABASE_PATH;
if (c.nodeDatabase["delete_batch"].isNotEmpty())
setup.deleteBatch = c.nodeDatabase["delete_batch"].getIntValue();
return setup;
}
std::unique_ptr<SHAMapStore>
make_SHAMapStore (SHAMapStore::Setup const& s,
beast::Stoppable& parent,
NodeStore::Manager& manager,
NodeStore::Scheduler& scheduler,
beast::Journal journal,
beast::Journal nodeStoreJournal,
TransactionMaster& transactionMaster)
{
return std::make_unique<SHAMapStoreImp> (s, parent, manager, scheduler,
journal, nodeStoreJournal, transactionMaster);
}
}

242
src/ripple/app/misc/SHAMapStoreImp.h Normal file
View File

@@ -0,0 +1,242 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_APP_SHAMAPSTOREIMP_H_INCLUDED
#define RIPPLE_APP_SHAMAPSTOREIMP_H_INCLUDED
#include <ripple/app/misc/SHAMapStore.h>
#include <ripple/nodestore/impl/Tuning.h>
#include <ripple/nodestore/DatabaseRotating.h>
#include <beast/module/sqdb/sqdb.h>
#include <iostream>
#include <condition_variable>
#include "NetworkOPs.h"
namespace ripple {
class SHAMapStoreImp : public SHAMapStore
{
private:
struct SavedState
{
std::string writableDb;
std::string archiveDb;
LedgerIndex lastRotated;
};
enum Health : std::uint8_t
{
ok = 0,
stopping,
unhealthy
};
class SavedStateDB
{
public:
beast::sqdb::session session_;
std::mutex mutex_;
beast::Journal journal_;
// Just instantiate without any logic in case online delete is not
// configured
SavedStateDB() = default;
// opens SQLite database and, if necessary, creates & initializes its tables.
void init (std::string const& databasePath, std::string const& dbName);
// get/set the ledger index that we can delete up to and including
LedgerIndex getCanDelete();
LedgerIndex setCanDelete (LedgerIndex canDelete);
SavedState getState();
void setState (SavedState const& state);
void setLastRotated (LedgerIndex seq);
void checkError (beast::Error const& error);
};
// name of sqlite state database
std::string const dbName_ = "state.db";
// prefix of on-disk nodestore backend instances
std::string const dbPrefix_ = "rippledb";
// check health/stop status as records are copied
std::uint64_t const checkHealthInterval_ = 1000;
// microseconds to back off between sqlite deletion batches
std::uint32_t pause_ = 1000;
// seconds to compare against ledger age
std::uint16_t ageTooHigh_ = 60;
// minimum # of ledgers to maintain for health of network
std::uint32_t minimumDeletionInterval_ = 256;
Setup setup_;
NodeStore::Manager& manager_;
NodeStore::Scheduler& scheduler_;
beast::Journal journal_;
beast::Journal nodeStoreJournal_;
NodeStore::DatabaseRotating* database_;
SavedStateDB state_db_;
std::thread thread_;
bool stop_ = false;
bool healthy_ = true;
mutable std::condition_variable cond_;
mutable std::mutex mutex_;
Ledger::pointer newLedger_;
Ledger::pointer validatedLedger_;
TransactionMaster& transactionMaster_;
// these do not exist upon SHAMapStore creation, but do exist
// as of onPrepare() or before
NetworkOPs* netOPs_ = nullptr;
LedgerMaster* ledgerMaster_ = nullptr;
FullBelowCache* fullBelowCache_ = nullptr;
TreeNodeCache* treeNodeCache_ = nullptr;
DatabaseCon* transactionDb_ = nullptr;
DatabaseCon* ledgerDb_ = nullptr;
public:
SHAMapStoreImp (Setup const& setup,
Stoppable& parent,
NodeStore::Manager& manager,
NodeStore::Scheduler& scheduler,
beast::Journal journal,
beast::Journal nodeStoreJournal,
TransactionMaster& transactionMaster);
~SHAMapStoreImp()
{
if (thread_.joinable())
thread_.join();
}
std::uint32_t
clampFetchDepth (std::uint32_t fetch_depth) const override
{
return setup_.deleteInterval ? std::min (fetch_depth,
setup_.deleteInterval) : fetch_depth;
}
std::unique_ptr <NodeStore::Database> makeDatabase (
std::string const&name, std::int32_t readThreads) override;
LedgerIndex
setCanDelete (LedgerIndex seq) override
{
return state_db_.setCanDelete (seq);
}
bool
advisoryDelete() const override
{
return setup_.advisoryDelete;
}
LedgerIndex
getLastRotated() override
{
return state_db_.getState().lastRotated;
}
LedgerIndex
getCanDelete() override
{
return state_db_.getCanDelete();
}
void onLedgerClosed (Ledger::pointer validatedLedger) override;
private:
// callback for visitNodes
bool copyNode (std::uint64_t& nodeCount, SHAMapTreeNode const &node);
void run();
void dbPaths();
std::shared_ptr <NodeStore::Backend> makeBackendRotating (
std::string path = std::string());
/**
* Creates a NodeStore with two
* backends to allow online deletion of data.
*
* @param name A diagnostic label for the database.
* @param readThreads The number of async read threads to create
* @param writableBackend backend for writing
* @param archiveBackend backend for archiving
*
* @return The opened database.
*/
std::unique_ptr <NodeStore::DatabaseRotating>
makeDatabaseRotating (std::string const&name,
std::int32_t readThreads,
std::shared_ptr <NodeStore::Backend> writableBackend,
std::shared_ptr <NodeStore::Backend> archiveBackend) const;
template <class CacheInstance>
bool
freshenCache (CacheInstance& cache)
{
std::uint64_t check = 0;
for (uint256 it: cache.getKeys())
{
database_->fetchNode (it);
if (! (++check % checkHealthInterval_) && health())
return true;
}
return false;
}
/** delete from sqlite table in batches to not lock the db excessively
* pause briefly to extend access time to other users
* call with mutex object unlocked
*/
void clearSql (DatabaseCon& database, LedgerIndex lastRotated,
std::string const& minQuery, std::string const& deleteQuery);
void clearCaches (LedgerIndex validatedSeq);
void freshenCaches();
void clearPrior (LedgerIndex lastRotated);
// If rippled is not healthy, defer rotate-delete.
// If already unhealthy, do not change state on further check.
// Assume that, once unhealthy, a necessary step has been
// aborted, so the online-delete process needs to restart
// at next ledger.
Health health();
//
// Stoppable
//
void
onPrepare() override
{
}
void
onStart() override
{
if (setup_.deleteInterval)
thread_ = std::thread (&SHAMapStoreImp::run, this);
}
// Called when the application begins shutdown
void onStop() override;
// Called when all child Stoppable objects have stoped
void onChildrenStopped() override;
};
}
#endif

5
src/ripple/app/node/SqliteFactory.cpp
View File

@@ -53,7 +53,8 @@ class SqliteBackend : public NodeStore::Backend
public:
explicit SqliteBackend (std::string const& path, int hashnode_cache_size)
: m_name (path)
, m_db (new DatabaseCon(path, s_nodeStoreDBInit, s_nodeStoreDBCount))
, m_db (new DatabaseCon(setup_DatabaseCon (getConfig()),
path, s_nodeStoreDBInit, s_nodeStoreDBCount))
{
std::string s ("PRAGMA cache_size=-");
s += std::to_string (hashnode_cache_size);
@@ -178,6 +179,8 @@ public:
return 0;
}
void setDeletePath() override {}
//--------------------------------------------------------------------------
void doBind (SqliteStatement& statement, NodeObject::ref object)

2
src/ripple/app/shamap/SHAMap.h
View File

@@ -179,7 +179,7 @@ public:
SHAMapItem::pointer peekNextItem (uint256 const& , SHAMapTreeNode::TNType & type);
SHAMapItem::pointer peekPrevItem (uint256 const& );
void visitNodes (std::function<void (SHAMapTreeNode&)> const&);
void visitNodes (std::function<bool (SHAMapTreeNode&)> const&);
void visitLeaves(std::function<void (SHAMapItem::ref)> const&);
// comparison/sync functions

19
src/ripple/app/shamap/SHAMapSync.cpp
View File

@@ -27,13 +27,15 @@ namespace ripple {
static const uint256 uZero;
static void visitLeavesHelper (
static bool visitLeavesHelper (
std::function <void (SHAMapItem::ref)> const& function,
SHAMapTreeNode& node)
{
// Adapt visitNodes to visitLeaves
if (!node.isInner ())
function (node.peekItem ());
return false;
}
void SHAMap::visitLeaves (std::function<void (SHAMapItem::ref item)> const& leafFunction)
@@ -42,7 +44,7 @@ void SHAMap::visitLeaves (std::function<void (SHAMapItem::ref item)> const& leaf
std::cref (leafFunction), std::placeholders::_1));
}
void SHAMap::visitNodes(std::function<void (SHAMapTreeNode&)> const& function)
void SHAMap::visitNodes(std::function<bool (SHAMapTreeNode&)> const& function)
{
// Visit every node in a SHAMap
assert (root->isValid ());
@@ -69,7 +71,8 @@ void SHAMap::visitNodes(std::function<void (SHAMapTreeNode&)> const& function)
if (!node->isEmptyBranch (pos))
{
SHAMapTreeNode::pointer child = descendNoStore (node, pos);
function (*child);
if (function (*child))
return;
if (child->isLeaf ())
++pos;
@@ -114,7 +117,8 @@ void SHAMap::getMissingNodes (std::vector<SHAMapNodeID>& nodeIDs, std::vector<ui
assert (root->isValid ());
assert (root->getNodeHash().isNonZero ());
if (root->isFullBelow ())
std::uint32_t generation = m_fullBelowCache.getGeneration();
if (root->isFullBelow (generation))
{
clearSynching ();
return;
@@ -190,7 +194,7 @@ void SHAMap::getMissingNodes (std::vector<SHAMapNodeID>& nodeIDs, std::vector<ui
fullBelow = false; // This node is not known full below
}
else if (d->isInner () && !d->isFullBelow ())
else if (d->isInner () && !d->isFullBelow (generation))
{
stack.push (std::make_tuple (node, nodeID,
firstChild, currentChild, fullBelow));
@@ -210,7 +214,7 @@ void SHAMap::getMissingNodes (std::vector<SHAMapNodeID>& nodeIDs, std::vector<ui
if (fullBelow)
{ // No partial node encountered below this node
node->setFullBelow ();
node->setFullBelowGen (generation);
if (mBacked)
m_fullBelowCache.insert (node->getNodeHash ());
}
@@ -452,10 +456,11 @@ SHAMap::addKnownNode (const SHAMapNodeID& node, Blob const& rawNode,
return SHAMapAddNode::duplicate ();
}
std::uint32_t generation = m_fullBelowCache.getGeneration();
SHAMapNodeID iNodeID;
SHAMapTreeNode* iNode = root.get ();
while (iNode->isInner () && !iNode->isFullBelow () &&
while (iNode->isInner () && !iNode->isFullBelow (generation) &&
(iNodeID.getDepth () < node.getDepth ()))
{
int branch = iNodeID.selectBranch (node.getNodeID ());

8
src/ripple/app/shamap/SHAMapTreeNode.cpp
View File

@@ -27,7 +27,7 @@ SHAMapTreeNode::SHAMapTreeNode (std::uint32_t seq)
: mSeq (seq)
, mType (tnERROR)
, mIsBranch (0)
, mFullBelow (false)
, mFullBelowGen (0)
{
}
@@ -36,7 +36,7 @@ SHAMapTreeNode::SHAMapTreeNode (const SHAMapTreeNode& node, std::uint32_t seq)
, mSeq (seq)
, mType (node.mType)
, mIsBranch (node.mIsBranch)
, mFullBelow (false)
, mFullBelowGen (0)
{
if (node.mItem)
mItem = node.mItem;
@@ -57,7 +57,7 @@ SHAMapTreeNode::SHAMapTreeNode (SHAMapItem::ref item,
, mSeq (seq)
, mType (type)
, mIsBranch (0)
, mFullBelow (false)
, mFullBelowGen (0)
{
assert (item->peekData ().size () >= 12);
updateHash ();
@@ -69,7 +69,7 @@ SHAMapTreeNode::SHAMapTreeNode (Blob const& rawNode,
: mSeq (seq)
, mType (tnERROR)
, mIsBranch (0)
, mFullBelow (false)
, mFullBelowGen (0)
{
if (format == snfWIRE)
{

10
src/ripple/app/shamap/SHAMapTreeNode.h
View File

@@ -169,13 +169,13 @@ public:
}
// sync functions
bool isFullBelow (void) const
bool isFullBelow (std::uint32_t generation) const
{
return mFullBelow;
return mFullBelowGen == generation;
}
void setFullBelow (void)
void setFullBelowGen (std::uint32_t gen)
{
mFullBelow = true;
mFullBelowGen = gen;
}
virtual void dump (SHAMapNodeID const&);
@@ -197,7 +197,7 @@ private:
std::uint32_t mSeq;
TNType mType;
int mIsBranch;
bool mFullBelow;
std::uint32_t mFullBelowGen;
bool updateHash ();

5
src/ripple/app/tx/TransactionMaster.cpp
View File

@@ -117,4 +117,9 @@ void TransactionMaster::sweep (void)
mCache.sweep ();
}
TaggedCache <uint256, Transaction>& TransactionMaster::getCache()
{
return mCache;
}
} // ripple

1
src/ripple/app/tx/TransactionMaster.h
View File

@@ -37,6 +37,7 @@ public:
bool inLedger (uint256 const& hash, std::uint32_t ledger);
bool canonicalize (Transaction::pointer* pTransaction);
void sweep (void);
TaggedCache <uint256, Transaction>& getCache();
private:
TaggedCache <uint256, Transaction> mCache;

14
src/ripple/basics/TaggedCache.h
View File

@@ -473,6 +473,20 @@ public:
return m_mutex;
}
std::vector <key_type> getKeys ()
{
std::vector <key_type> v;
{
lock_guard lock (m_mutex);
v.reserve (m_cache.size());
for (auto const& _ : m_cache)
v.push_back (_.first);
}
return v;
}
private:
void collect_metrics ()
{

19
src/ripple/net/impl/RPCCall.cpp
View File

@@ -358,6 +358,24 @@ private:
return jvRequest;
}
// can_delete [<ledgerid>|<ledgerhash>|now|always|never]
Json::Value parseCanDelete (Json::Value const& jvParams)
{
Json::Value jvRequest (Json::objectValue);
if (!jvParams.size ())
return jvRequest;
std::string input = jvParams[0u].asString();
if (input.find_first_not_of("0123456789") ==
std::string::npos)
jvRequest["can_delete"] = jvParams[0u].asUInt();
else
jvRequest["can_delete"] = input;
return jvRequest;
}
// connect <ip> [port]
Json::Value parseConnect (Json::Value const& jvParams)
{
@@ -841,6 +859,7 @@ public:
{ "account_offers", &RPCParser::parseAccountItems, 1, 4 },
{ "account_tx", &RPCParser::parseAccountTransactions, 1, 8 },
{ "book_offers", &RPCParser::parseBookOffers, 2, 7 },
{ "can_delete", &RPCParser::parseCanDelete, 0, 1 },
{ "connect", &RPCParser::parseConnect, 1, 2 },
{ "consensus_info", &RPCParser::parseAsIs, 0, 0 },
{ "feature", &RPCParser::parseFeature, 0, 2 },

3
src/ripple/nodestore/Backend.h
View File

@@ -84,6 +84,9 @@ public:
/** Estimate the number of write operations pending. */
virtual int getWriteLoad () = 0;
/** Remove contents on disk upon destruction. */
virtual void setDeletePath() = 0;
};
}

4
src/ripple/nodestore/Database.h
View File

@@ -21,6 +21,8 @@
#define RIPPLE_NODESTORE_DATABASE_H_INCLUDED
#include <ripple/nodestore/NodeObject.h>
#include <ripple/nodestore/Backend.h>
#include <ripple/basics/TaggedCache.h>
namespace ripple {
namespace NodeStore {
@@ -118,7 +120,7 @@ public:
/** Retrieve the estimated number of pending write operations.
This is used for diagnostics.
*/
virtual int getWriteLoad () = 0;
virtual std::int32_t getWriteLoad() const = 0;
/** Get the positive cache hits to total attempts ratio. */
virtual float getCacheHitRate () = 0;

56
src/ripple/nodestore/DatabaseRotating.h Normal file
View File

@@ -0,0 +1,56 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_NODESTORE_DATABASEROTATING_H_INCLUDED
#define RIPPLE_NODESTORE_DATABASEROTATING_H_INCLUDED
#include <ripple/nodestore/Database.h>
namespace ripple {
namespace NodeStore {
/* This class has two key-value store Backend objects for persisting SHAMap
* records. This facilitates online deletion of data. New backends are
* rotated in. Old ones are rotated out and deleted.
*/
class DatabaseRotating
{
public:
virtual ~DatabaseRotating() = default;
virtual TaggedCache <uint256, NodeObject>& getPositiveCache() = 0;
virtual std::mutex& peekMutex() const = 0;
virtual std::shared_ptr <Backend> const& getWritableBackend() const = 0;
virtual std::shared_ptr <Backend> const& getArchiveBackend () const = 0;
virtual std::shared_ptr <Backend> rotateBackends (
std::shared_ptr <Backend> const& newBackend) = 0;
/** Ensure that node is in writableBackend */
virtual NodeObject::Ptr fetchNode (uint256 const& hash) = 0;
};
}
}
#endif

10
src/ripple/nodestore/Manager.h
View File

@@ -21,6 +21,7 @@
#define RIPPLE_NODESTORE_MANAGER_H_INCLUDED
#include <ripple/nodestore/Factory.h>
#include <ripple/nodestore/DatabaseRotating.h>
namespace ripple {
namespace NodeStore {
@@ -75,6 +76,15 @@ public:
Scheduler& scheduler, beast::Journal journal, int readThreads,
Parameters const& backendParameters,
Parameters fastBackendParameters = Parameters ()) = 0;
virtual std::unique_ptr <DatabaseRotating> make_DatabaseRotating (
std::string const& name,
Scheduler& scheduler,
std::int32_t readThreads,
std::shared_ptr <Backend> writableBackend,
std::shared_ptr <Backend> archiveBackend,
std::unique_ptr <Backend> fastBackend,
beast::Journal journal) = 0;
};
//------------------------------------------------------------------------------

18
src/ripple/nodestore/backend/HyperDBFactory.cpp
View File

@@ -29,6 +29,9 @@ class HyperDBBackend
, public BatchWriter::Callback
, public beast::LeakChecked <HyperDBBackend>
{
private:
std::atomic <bool> m_deletePath;
public:
beast::Journal m_journal;
size_t const m_keyBytes;
@@ -40,7 +43,8 @@ public:
HyperDBBackend (size_t keyBytes, Parameters const& keyValues,
Scheduler& scheduler, beast::Journal journal)
: m_journal (journal)
: m_deletePath (false)
, m_journal (journal)
, m_keyBytes (keyBytes)
, m_scheduler (scheduler)
, m_batch (*this, scheduler)
@@ -88,6 +92,12 @@ public:
~HyperDBBackend ()
{
if (m_deletePath)
{
m_db.reset();
boost::filesystem::path dir = m_name;
boost::filesystem::remove_all (dir);
}
}
std::string
@@ -219,6 +229,12 @@ public:
return m_batch.getWriteLoad ();
}
void
setDeletePath() override
{
m_deletePath = true;
}
//--------------------------------------------------------------------------
void

22
src/ripple/nodestore/backend/LevelDBFactory.cpp
View File

@@ -29,6 +29,9 @@ class LevelDBBackend
, public BatchWriter::Callback
, public beast::LeakChecked <LevelDBBackend>
{
private:
std::atomic <bool> m_deletePath;
public:
beast::Journal m_journal;
size_t const m_keyBytes;
@@ -40,7 +43,8 @@ public:
LevelDBBackend (int keyBytes, Parameters const& keyValues,
Scheduler& scheduler, beast::Journal journal)
: m_journal (journal)
: m_deletePath (false)
, m_journal (journal)
, m_keyBytes (keyBytes)
, m_scheduler (scheduler)
, m_batch (*this, scheduler)
@@ -93,6 +97,16 @@ public:
m_db.reset (db);
}
~LevelDBBackend()
{
if (m_deletePath)
{
m_db.reset();
boost::filesystem::path dir = m_name;
boost::filesystem::remove_all (dir);
}
}
std::string
getName()
{
@@ -222,6 +236,12 @@ public:
return m_batch.getWriteLoad ();
}
void
setDeletePath() override
{
m_deletePath = true;
}
//--------------------------------------------------------------------------
void

3
src/ripple/nodestore/backend/MemoryFactory.cpp
View File

@@ -98,6 +98,9 @@ public:
{
return 0;
}
void
setDeletePath() override {}
};
//------------------------------------------------------------------------------

3
src/ripple/nodestore/backend/NullFactory.cpp
View File

@@ -64,6 +64,9 @@ public:
return 0;
}
void
setDeletePath() override {}
private:
};

18
src/ripple/nodestore/backend/RocksDBFactory.cpp
View File

@@ -79,6 +79,9 @@ class RocksDBBackend
, public BatchWriter::Callback
, public beast::LeakChecked <RocksDBBackend>
{
private:
std::atomic <bool> m_deletePath;
public:
beast::Journal m_journal;
size_t const m_keyBytes;
@@ -89,7 +92,8 @@ public:
RocksDBBackend (int keyBytes, Parameters const& keyValues,
Scheduler& scheduler, beast::Journal journal, RocksDBEnv* env)
: m_journal (journal)
: m_deletePath (false)
, m_journal (journal)
, m_keyBytes (keyBytes)
, m_scheduler (scheduler)
, m_batch (*this, scheduler)
@@ -192,6 +196,12 @@ public:
~RocksDBBackend ()
{
if (m_deletePath)
{
m_db.reset();
boost::filesystem::path dir = m_name;
boost::filesystem::remove_all (dir);
}
}
std::string
@@ -326,6 +336,12 @@ public:
return m_batch.getWriteLoad ();
}
void
setDeletePath() override
{
m_deletePath = true;
}
//--------------------------------------------------------------------------
void

15
src/ripple/nodestore/backend/RocksDBQuickFactory.cpp
View File

@@ -78,6 +78,9 @@ class RocksDBQuickBackend
: public Backend
, public beast::LeakChecked <RocksDBQuickBackend>
{
private:
std::atomic <bool> m_deletePath;
public:
beast::Journal m_journal;
size_t const m_keyBytes;
@@ -162,6 +165,12 @@ public:
~RocksDBQuickBackend ()
{
if (m_deletePath)
{
m_db.reset();
boost::filesystem::path dir = m_name;
boost::filesystem::remove_all (dir);
}
}
std::string
@@ -299,6 +308,12 @@ public:
return 0;
}
void
setDeletePath() override
{
m_deletePath = true;
}
//--------------------------------------------------------------------------
void

59
src/ripple/nodestore/impl/DatabaseImp.h
View File

@@ -81,7 +81,8 @@ public:
, m_fetchSize (0)
{
for (int i = 0; i < readThreads; ++i)
m_readThreads.push_back (std::thread (&DatabaseImp::threadEntry, this));
m_readThreads.push_back (std::thread (&DatabaseImp::threadEntry,
this));
}
~DatabaseImp ()
@@ -98,7 +99,7 @@ public:
}
std::string
getName () const
getName () const override
{
return m_backend->getName ();
}
@@ -201,7 +202,8 @@ public:
{
// Yes so at last we will try the main database.
//
obj = fetchInternal (*m_backend, hash);
obj = fetchFrom (hash);
++m_fetchTotalCount;
}
if (obj == nullptr)
@@ -230,7 +232,7 @@ public:
{
m_fastBackend->store (obj);
++m_storeCount;
if (obj.get())
if (obj)
m_storeSize += obj->getData().size();
}
@@ -244,19 +246,23 @@ public:
return obj;
}
virtual NodeObject::Ptr fetchFrom (uint256 const& hash)
{
return fetchInternal (*m_backend, hash);
}
NodeObject::Ptr fetchInternal (Backend& backend,
uint256 const& hash)
{
NodeObject::Ptr object;
Status const status = backend.fetch (hash.begin (), &object);
++m_fetchTotalCount;
switch (status)
{
case ok:
++m_fetchHitCount;
if (object.get())
if (object)
m_fetchSize += object->getData().size();
case notFound:
break;
@@ -284,7 +290,17 @@ public:
Blob&& data,
uint256 const& hash)
{
NodeObject::Ptr object = NodeObject::createObject(type, index, std::move(data), hash);
storeInternal (type, index, std::move(data), hash, *m_backend.get());
}
void storeInternal (NodeObjectType type,
std::uint32_t index,
Blob&& data,
uint256 const& hash,
Backend& backend)
{
NodeObject::Ptr object = NodeObject::createObject(type, index,
std::move(data), hash);
#if RIPPLE_VERIFY_NODEOBJECT_KEYS
assert (hash == Serializer::getSHA512Half (data));
@@ -292,9 +308,9 @@ public:
m_cache.canonicalize (hash, object, true);
m_backend->store (object);
backend.store (object);
++m_storeCount;
if (object.get())
if (object)
m_storeSize += object->getData().size();
m_negCache.erase (hash);
@@ -303,7 +319,7 @@ public:
{
m_fastBackend->store (object);
++m_storeCount;
if (object.get())
if (object)
m_storeSize += object->getData().size();
}
}
@@ -329,9 +345,9 @@ public:
m_negCache.sweep ();
}
int getWriteLoad ()
std::int32_t getWriteLoad() const override
{
return m_backend->getWriteLoad ();
return m_backend->getWriteLoad();
}
//------------------------------------------------------------------------------
@@ -380,33 +396,38 @@ public:
//------------------------------------------------------------------------------
void for_each (std::function <void(NodeObject::Ptr)> f)
void for_each (std::function <void(NodeObject::Ptr)> f) override
{
m_backend->for_each (f);
}
void import (Database& source)
{
importInternal (source, *m_backend.get());
}
void importInternal (Database& source, Backend& dest)
{
Batch b;
b.reserve (batchWritePreallocationSize);
source.for_each ([&](NodeObject::Ptr object)
{
if (b.size () >= batchWritePreallocationSize)
if (b.size() >= batchWritePreallocationSize)
{
this->m_backend->storeBatch (b);
b.clear ();
dest.storeBatch (b);
b.clear();
b.reserve (batchWritePreallocationSize);
}
b.push_back (object);
++m_storeCount;
if (object.get())
if (object)
m_storeSize += object->getData().size();
});
if (! b.empty ())
m_backend->storeBatch (b);
if (! b.empty())
dest.storeBatch (b);
}
std::uint32_t getStoreCount () const override

54
src/ripple/nodestore/impl/DatabaseRotatingImp.cpp Normal file
View File

@@ -0,0 +1,54 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include <ripple/nodestore/impl/DatabaseRotatingImp.h>
namespace ripple {
namespace NodeStore {
// Make sure to call it already locked!
std::shared_ptr <Backend> DatabaseRotatingImp::rotateBackends (
std::shared_ptr <Backend> const& newBackend)
{
std::shared_ptr <Backend> oldBackend = archiveBackend_;
archiveBackend_ = writableBackend_;
writableBackend_ = newBackend;
return oldBackend;
}
NodeObject::Ptr DatabaseRotatingImp::fetchFrom (uint256 const& hash)
{
Backends b = getBackends();
NodeObject::Ptr object = fetchInternal (*b.writableBackend, hash);
if (!object)
{
object = fetchInternal (*b.archiveBackend, hash);
if (object)
{
getWritableBackend()->store (object);
m_negCache.erase (hash);
}
}
return object;
}
}
}

129
src/ripple/nodestore/impl/DatabaseRotatingImp.h Normal file
View File

@@ -0,0 +1,129 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_NODESTORE_DATABASEROTATINGIMP_H_INCLUDED
#define RIPPLE_NODESTORE_DATABASEROTATINGIMP_H_INCLUDED
#include <ripple/nodestore/impl/DatabaseImp.h>
#include <ripple/nodestore/DatabaseRotating.h>
namespace ripple {
namespace NodeStore {
class DatabaseRotatingImp
: public DatabaseImp
, public DatabaseRotating
{
private:
std::shared_ptr <Backend> writableBackend_;
std::shared_ptr <Backend> archiveBackend_;
mutable std::mutex rotateMutex_;
struct Backends {
std::shared_ptr <Backend> const& writableBackend;
std::shared_ptr <Backend> const& archiveBackend;
};
Backends getBackends() const
{
std::lock_guard <std::mutex> lock (rotateMutex_);
return Backends {writableBackend_, archiveBackend_};
}
public:
DatabaseRotatingImp (std::string const& name,
Scheduler& scheduler,
int readThreads,
std::shared_ptr <Backend> writableBackend,
std::shared_ptr <Backend> archiveBackend,
std::unique_ptr <Backend> fastBackend,
beast::Journal journal)
: DatabaseImp (name, scheduler, readThreads,
std::unique_ptr <Backend>(), std::move (fastBackend),
journal)
, writableBackend_ (writableBackend)
, archiveBackend_ (archiveBackend)
{}
std::shared_ptr <Backend> const& getWritableBackend() const override
{
std::lock_guard <std::mutex> lock (rotateMutex_);
return writableBackend_;
}
std::shared_ptr <Backend> const& getArchiveBackend() const override
{
std::lock_guard <std::mutex> lock (rotateMutex_);
return archiveBackend_;
}
std::shared_ptr <Backend> rotateBackends (
std::shared_ptr <Backend> const& newBackend) override;
std::mutex& peekMutex() const override
{
return rotateMutex_;
}
std::string getName() const override
{
return getWritableBackend()->getName();
}
std::int32_t getWriteLoad() const override
{
return getWritableBackend()->getWriteLoad();
}
void for_each (std::function <void(NodeObject::Ptr)> f) override
{
Backends b = getBackends();
b.archiveBackend->for_each (f);
b.writableBackend->for_each (f);
}
void import (Database& source) override
{
importInternal (source, *getWritableBackend());
}
void store (NodeObjectType type,
std::uint32_t index,
Blob&& data,
uint256 const& hash) override
{
storeInternal (type, index, std::move(data), hash,
*getWritableBackend());
}
NodeObject::Ptr fetchNode (uint256 const& hash) override
{
return fetchFrom (hash);
}
NodeObject::Ptr fetchFrom (uint256 const& hash) override;
TaggedCache <uint256, NodeObject>& getPositiveCache() override
{
return m_cache;
}
};
}
}
#endif

15
src/ripple/nodestore/impl/Manager.cpp
View File

@@ -140,6 +140,21 @@ public:
return std::make_unique <DatabaseImp> (name, scheduler, readThreads,
std::move (backend), std::move (fastBackend), journal);
}
std::unique_ptr <DatabaseRotating>
make_DatabaseRotating (
std::string const& name,
Scheduler& scheduler,
std::int32_t readThreads,
std::shared_ptr <Backend> writableBackend,
std::shared_ptr <Backend> archiveBackend,
std::unique_ptr <Backend> fastBackend,
beast::Journal journal)
{
return std::make_unique <DatabaseRotatingImp> (name, scheduler,
readThreads, writableBackend, archiveBackend,
std::move (fastBackend), journal);
}
};
//------------------------------------------------------------------------------

1
src/ripple/protocol/JsonFields.h
View File

@@ -49,6 +49,7 @@ JSS ( base_fee_xrp );
JSS ( bids );
JSS ( binary );
JSS ( build_version );
JSS ( can_delete );
JSS ( closed );
JSS ( closed_ledger );
JSS ( close_time );

3
src/ripple/rpc/ErrorCodes.h
View File

@@ -40,6 +40,7 @@ enum error_code_i
// Programs should use error tokens.
// Misc failure
rpcGENERAL,
rpcLOAD_FAILED,
rpcNO_PERMISSION,
rpcNO_EVENTS,
@@ -47,6 +48,8 @@ enum error_code_i
rpcTOO_BUSY,
rpcSLOW_DOWN,
rpcHIGH_FEE,
rpcNOT_ENABLED,
rpcNOT_READY,
// Networking
rpcNO_CLOSED,

96
src/ripple/rpc/handlers/CanDelete.cpp Normal file
View File

@@ -0,0 +1,96 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012-2014 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include <ripple/app/misc/SHAMapStore.h>
#include <boost/algorithm/string/case_conv.hpp>
#include <boost/format.hpp>
namespace ripple {
// can_delete [<ledgerid>|<ledgerhash>|now|always|never]
Json::Value doCanDelete (RPC::Context& context)
{
if (! getApp().getSHAMapStore().advisoryDelete())
return RPC::make_error(rpcNOT_ENABLED);
Json::Value ret (Json::objectValue);
if (context.params.isMember("can_delete"))
{
Json::Value canDelete = context.params.get(jss::can_delete, 0);
std::uint32_t canDeleteSeq = 0;
if (canDelete.isUInt())
{
canDeleteSeq = canDelete.asUInt();
}
else
{
std::string canDeleteStr = canDelete.asString();
boost::to_lower (canDeleteStr);
if (canDeleteStr.find_first_not_of ("0123456789") ==
std::string::npos)
{
canDeleteSeq =
beast::lexicalCast <std::uint32_t>(canDeleteStr);
}
else if (canDeleteStr == "never")
{
canDeleteSeq = 0;
}
else if (canDeleteStr == "always")
{
canDeleteSeq = std::numeric_limits <std::uint32_t>::max();
}
else if (canDeleteStr == "now")
{
canDeleteSeq = getApp().getSHAMapStore().getLastRotated();
if (!canDeleteSeq)
return RPC::make_error (rpcNOT_READY); }
else if (canDeleteStr.size() == 64 &&
canDeleteStr.find_first_not_of("0123456789abcdef") ==
std::string::npos)
{
uint256 ledgerHash (canDeleteStr);
Ledger::pointer ledger =
context.netOps.getLedgerByHash (ledgerHash);
if (!ledger)
return RPC::make_error(rpcLGR_NOT_FOUND, "ledgerNotFound");
canDeleteSeq = ledger->getLedgerSeq();
}
else
{
return RPC::make_error (rpcINVALID_PARAMS);
}
}
ret["can_delete"] =
getApp().getSHAMapStore().setCanDelete (canDeleteSeq);
}
else
{
ret["can_delete"] = getApp().getSHAMapStore().getCanDelete();
}
return ret;
}
} // ripple

1
src/ripple/rpc/handlers/Handlers.h
View File

@@ -31,6 +31,7 @@ Json::Value doAccountTxSwitch (RPC::Context&);
Json::Value doAccountTxOld (RPC::Context&);
Json::Value doBookOffers (RPC::Context&);
Json::Value doBlackList (RPC::Context&);
Json::Value doCanDelete (RPC::Context&);
Json::Value doConnect (RPC::Context&);
Json::Value doConsensusInfo (RPC::Context&);
Json::Value doFeature (RPC::Context&);

3
src/ripple/rpc/impl/ErrorCodes.cpp
View File

@@ -65,6 +65,7 @@ public:
add (rpcDST_ISR_MALFORMED, "dstIsrMalformed", "Destination issuer is malformed.");
add (rpcFAIL_GEN_DECRYPT, "failGenDecrypt", "Failed to decrypt generator.");
add (rpcFORBIDDEN, "forbidden", "Bad credentials.");
add (rpcGENERAL, "general", "Generic error reason.");
add (rpcGETS_ACT_MALFORMED, "getsActMalformed", "Gets account malformed.");
add (rpcGETS_AMT_MALFORMED, "getsAmtMalformed", "Gets amount malformed.");
add (rpcHIGH_FEE, "highFee", "Current transaction fee exceeds your limit.");
@@ -78,7 +79,9 @@ public:
add (rpcLGR_NOT_FOUND, "lgrNotFound", "Ledger not found.");
add (rpcLOAD_FAILED, "loadFailed", "Load failed");
add (rpcMASTER_DISABLED, "masterDisabled", "Master key is disabled.");
add (rpcNOT_ENABLED, "notEnabled", "Not enabled in configuration.");
add (rpcNOT_IMPL, "notImpl", "Not implemented.");
add (rpcNOT_READY, "notReady", "Not ready to handle this request.");
add (rpcNOT_STANDALONE, "notStandAlone", "Operation valid in debug mode only.");
add (rpcNOT_SUPPORTED, "notSupported", "Operation not supported.");
add (rpcNO_ACCOUNT, "noAccount", "No such account.");

2
src/ripple/rpc/impl/Handler.cpp
View File

@@ -49,6 +49,7 @@ HandlerTable HANDLERS({
{ "account_tx", &doAccountTxSwitch, Role::USER, NEEDS_NETWORK_CONNECTION },
{ "blacklist", &doBlackList, Role::ADMIN, NO_CONDITION },
{ "book_offers", &doBookOffers, Role::USER, NEEDS_CURRENT_LEDGER },
{ "can_delete", &doCanDelete, Role::ADMIN, NO_CONDITION },
{ "connect", &doConnect, Role::ADMIN, NO_CONDITION },
{ "consensus_info", &doConsensusInfo, Role::ADMIN, NO_CONDITION },
{ "get_counts", &doGetCounts, Role::ADMIN, NO_CONDITION },
@@ -98,7 +99,6 @@ HandlerTable HANDLERS({
{ "wallet_accounts", &doWalletAccounts, Role::USER, NEEDS_CURRENT_LEDGER },
{ "wallet_propose", &doWalletPropose, Role::ADMIN, NO_CONDITION },
{ "wallet_seed", &doWalletSeed, Role::ADMIN, NO_CONDITION },
// Evented methods
{ "subscribe", &doSubscribe, Role::USER, NO_CONDITION },
{ "unsubscribe", &doUnsubscribe, Role::USER, NO_CONDITION },

2
src/ripple/rpc/impl/LookupLedger.cpp
View File

@@ -66,7 +66,7 @@ Json::Value lookupLedger (
}
}
uint256 ledgerHash (0);
uint256 ledgerHash;
if (!jsonHash.isString() || !ledgerHash.SetHex (jsonHash.asString ()))
return make_error(rpcINVALID_PARAMS, "ledgerHashMalformed");

1
src/ripple/unity/app2.cpp
View File

@@ -30,6 +30,7 @@
#include <ripple/app/ledger/OrderBookDB.cpp>
#include <ripple/app/main/LoadManager.cpp>
#include <ripple/app/misc/CanonicalTXSet.cpp>
#include <ripple/app/misc/SHAMapStoreImp.cpp>
#include <ripple/app/shamap/SHAMap.cpp>
#include <ripple/app/shamap/SHAMapItem.cpp>
#include <ripple/app/shamap/SHAMapSync.cpp>

2
src/ripple/unity/nodestore.cpp
View File

@@ -51,7 +51,9 @@
#include <ripple/nodestore/impl/Backend.cpp>
#include <ripple/nodestore/impl/BatchWriter.cpp>
#include <ripple/nodestore/impl/DatabaseImp.h>
#include <ripple/nodestore/impl/DatabaseRotatingImp.h>
#include <ripple/nodestore/impl/Database.cpp>
#include <ripple/nodestore/impl/DatabaseRotatingImp.cpp>
#include <ripple/nodestore/impl/DummyScheduler.cpp>
#include <ripple/nodestore/impl/DecodedBlob.cpp>
#include <ripple/nodestore/impl/EncodedBlob.cpp>

1
src/ripple/unity/rpcx.cpp
View File

@@ -49,6 +49,7 @@
#include <ripple/rpc/handlers/AccountTxSwitch.cpp>
#include <ripple/rpc/handlers/BlackList.cpp>
#include <ripple/rpc/handlers/BookOffers.cpp>
#include <ripple/rpc/handlers/CanDelete.cpp>
#include <ripple/rpc/handlers/Connect.cpp>
#include <ripple/rpc/handlers/ConsensusInfo.cpp>
#include <ripple/rpc/handlers/Feature.cpp>