From b7920f40b7896c56291b6bc5d63daf10844c722d Mon Sep 17 00:00:00 2001 From: JoelKatz Date: Wed, 5 Jun 2013 10:25:03 -0700 Subject: [PATCH] Implement the ephemeral cache. --- src/cpp/ripple/Application.cpp | 18 ++++- src/cpp/ripple/Application.h | 2 + src/cpp/ripple/Config.cpp | 2 + src/cpp/ripple/Config.h | 1 + src/cpp/ripple/HashedObject.cpp | 137 +++++++++++++++++++++++--------- src/cpp/ripple/HashedObject.h | 2 +- 6 files changed, 123 insertions(+), 39 deletions(-) diff --git a/src/cpp/ripple/Application.cpp b/src/cpp/ripple/Application.cpp index ec0bd7079a..8c7384f0fb 100644 --- a/src/cpp/ripple/Application.cpp +++ b/src/cpp/ripple/Application.cpp @@ -54,7 +54,7 @@ Application::Application() : mFeatureTable(2 * 7 * 24 * 60 * 60, 200), // two weeks, 200/256 mRpcDB(NULL), mTxnDB(NULL), mLedgerDB(NULL), mWalletDB(NULL), - mNetNodeDB(NULL), mPathFindDB(NULL), mHashNodeDB(NULL), mHashNodeLDB(NULL), + mNetNodeDB(NULL), mPathFindDB(NULL), mHashNodeDB(NULL), mHashNodeLDB(NULL), mEphemeralLDB(NULL), mConnectionPool(mIOService), mPeerDoor(NULL), mRPCDoor(NULL), mWSPublicDoor(NULL), mWSPrivateDoor(NULL), mSweepTimer(mAuxService), mShutdown(false) { @@ -82,6 +82,9 @@ void Application::stop() delete mHashNodeLDB; mHashNodeLDB = NULL; + delete mEphemeralLDB; + mEphemeralLDB = NULL; + WriteLog (lsINFO, Application) << "Stopped: " << mIOService.stopped(); Instance::shutdown(); } @@ -176,6 +179,18 @@ void Application::setup() StopSustain(); exit(3); } + + if (!theConfig.LDB_EPHEMERAL.empty()) + { + leveldb::Status status = leveldb::DB::Open(options, theConfig.LDB_EPHEMERAL, &mEphemeralLDB); + if (!status.ok() || !mEphemeralLDB) + { + WriteLog(lsFATAL, Application) << "Unable to open/create epehemeral db: " + << theConfig.LDB_EPHEMERAL << " " << status.ToString(); + StopSustain(); + exit(3); + } + } } else { @@ -404,6 +419,7 @@ Application::~Application() delete mNetNodeDB; delete mPathFindDB; delete mHashNodeLDB; + delete mEphemeralLDB; } void Application::startNewLedger() diff --git a/src/cpp/ripple/Application.h b/src/cpp/ripple/Application.h index 4d992de59b..efab7a2a76 100644 --- a/src/cpp/ripple/Application.h +++ b/src/cpp/ripple/Application.h @@ -82,6 +82,7 @@ class Application DatabaseCon *mRpcDB, *mTxnDB, *mLedgerDB, *mWalletDB, *mNetNodeDB, *mPathFindDB, *mHashNodeDB; leveldb::DB *mHashNodeLDB; + leveldb::DB *mEphemeralLDB; ConnectionPool mConnectionPool; PeerDoor* mPeerDoor; @@ -153,6 +154,7 @@ public: DatabaseCon* getHashNodeDB() { return mHashNodeDB; } leveldb::DB* getHashNodeLDB() { return mHashNodeLDB; } + leveldb::DB* getEphemeralLDB() { return mEphemeralLDB; } uint256 getNonce256() { return mNonce256; } std::size_t getNonceST() { return mNonceST; } diff --git a/src/cpp/ripple/Config.cpp b/src/cpp/ripple/Config.cpp index b695650afe..c73de571db 100644 --- a/src/cpp/ripple/Config.cpp +++ b/src/cpp/ripple/Config.cpp @@ -24,6 +24,7 @@ #define SECTION_FEE_ACCOUNT_RESERVE "fee_account_reserve" #define SECTION_FEE_OWNER_RESERVE "fee_owner_reserve" #define SECTION_NODE_DB "node_db" +#define SECTION_LDB_EPHEMERAL "ephemeral_db" #define SECTION_LEDGER_HISTORY "ledger_history" #define SECTION_IPS "ips" #define SECTION_NETWORK_QUORUM "network_quorum" @@ -357,6 +358,7 @@ void Config::load() (void) sectionSingleB(secConfig, SECTION_RPC_PASSWORD, RPC_PASSWORD); (void) sectionSingleB(secConfig, SECTION_RPC_USER, RPC_USER); (void) sectionSingleB(secConfig, SECTION_NODE_DB, NODE_DB); + (void) sectionSingleB(secConfig, SECTION_LDB_EPHEMERAL, LDB_EPHEMERAL); if (sectionSingleB(secConfig, SECTION_RPC_PORT, strTemp)) RPC_PORT = boost::lexical_cast(strTemp); diff --git a/src/cpp/ripple/Config.h b/src/cpp/ripple/Config.h index 581c7126e6..fa9dca3da8 100644 --- a/src/cpp/ripple/Config.h +++ b/src/cpp/ripple/Config.h @@ -85,6 +85,7 @@ public: boost::filesystem::path DEBUG_LOGFILE; boost::filesystem::path VALIDATORS_FILE; // As specifed in rippled.cfg. std::string NODE_DB; // Database to use for nodes + std::string LDB_EPHEMERAL; // Database for temporary storage bool LDB_IMPORT; // Import into LevelDB bool ELB_SUPPORT; // Support Amazon ELB diff --git a/src/cpp/ripple/HashedObject.cpp b/src/cpp/ripple/HashedObject.cpp index a158e5590a..096b9546c3 100644 --- a/src/cpp/ripple/HashedObject.cpp +++ b/src/cpp/ripple/HashedObject.cpp @@ -16,7 +16,7 @@ DECLARE_INSTANCE(HashedObject); HashedObjectStore::HashedObjectStore(int cacheSize, int cacheAge) : mCache("HashedObjectStore", cacheSize, cacheAge), mNegativeCache("HashedObjectNegativeCache", 0, 120), - mWriteGeneration(0), mWriteLoad(0), mWritePending(false), mLevelDB(false) + mWriteGeneration(0), mWriteLoad(0), mWritePending(false), mLevelDB(false), mEphemeralDB(false) { mWriteSet.reserve(128); @@ -29,6 +29,8 @@ HashedObjectStore::HashedObjectStore(int cacheSize, int cacheAge) : WriteLog (lsFATAL, HashedObject) << "Incorrect database selection"; assert(false); } + if (!theConfig.LDB_EPHEMERAL.empty()) + mEphemeralDB = true; } void HashedObjectStore::tune(int size, int age) @@ -51,6 +53,74 @@ int HashedObjectStore::getWriteLoad() return std::max(mWriteLoad, static_cast(mWriteSet.size())); } +static HashedObject::pointer LLRetrieve(const uint256& hash, leveldb::DB* db) +{ // low-level retrieve + std::string sData; + + leveldb::Status st = db->Get(leveldb::ReadOptions(), + leveldb::Slice(reinterpret_cast(hash.begin()), hash.size()), &sData); + if (!st.ok()) + { + assert(st.IsNotFound()); + return HashedObject::pointer(); + } + + const unsigned char* bufPtr = reinterpret_cast(&sData[0]); + uint32 index = htonl(*reinterpret_cast(bufPtr)); + int htype = bufPtr[8]; + + return boost::make_shared(static_cast(htype), index, + bufPtr + 9, sData.size() - 9, hash); +} + +static void LLWrite(boost::shared_ptr ptr, leveldb::DB* db) +{ // low-level write single + HashedObject& obj = *ptr; + std::vector rawData(9 + obj.mData.size()); + unsigned char* bufPtr = &rawData.front(); + + *reinterpret_cast(bufPtr + 0) = ntohl(obj.mLedgerIndex); + *reinterpret_cast(bufPtr + 4) = ntohl(obj.mLedgerIndex); + *(bufPtr + 8) = static_cast(obj.mType); + memcpy(bufPtr + 9, &obj.mData.front(), obj.mData.size()); + + leveldb::Status st = db->Put(leveldb::WriteOptions(), + leveldb::Slice(reinterpret_cast(obj.mHash.begin()), obj.mHash.size()), + leveldb::Slice(reinterpret_cast(bufPtr), rawData.size())); + if (!st.ok()) + { + WriteLog (lsFATAL, HashedObject) << "Failed to store hash node"; + assert(false); + } +} + +static void LLWrite(const std::vector< boost::shared_ptr >& set, leveldb::DB* db) +{ // low-level write set + leveldb::WriteBatch batch; + + BOOST_FOREACH(const boost::shared_ptr& it, set) + { + const HashedObject& obj = *it; + std::vector rawData(9 + obj.mData.size()); + unsigned char* bufPtr = &rawData.front(); + + *reinterpret_cast(bufPtr + 0) = ntohl(obj.mLedgerIndex); + *reinterpret_cast(bufPtr + 4) = ntohl(obj.mLedgerIndex); + *(bufPtr + 8) = static_cast(obj.mType); + memcpy(bufPtr + 9, &obj.mData.front(), obj.mData.size()); + + batch.Put(leveldb::Slice(reinterpret_cast(obj.mHash.begin()), obj.mHash.size()), + leveldb::Slice(reinterpret_cast(bufPtr), rawData.size())); + } + + leveldb::Status st = db->Write(leveldb::WriteOptions(), &batch); + if (!st.ok()) + { + WriteLog (lsFATAL, HashedObject) << "Failed to store hash node"; + assert(false); + } +} + bool HashedObjectStore::storeLevelDB(HashedObjectType type, uint32 index, const std::vector& data, const uint256& hash) { // return: false = already in cache, true = added to cache @@ -106,31 +176,9 @@ void HashedObjectStore::bulkWriteLevelDB(Job &) setSize = set.size(); } - { - leveldb::WriteBatch batch; - - BOOST_FOREACH(const boost::shared_ptr& it, set) - { - const HashedObject& obj = *it; - std::vector rawData(9 + obj.mData.size()); - unsigned char* bufPtr = &rawData.front(); - - *reinterpret_cast(bufPtr + 0) = ntohl(obj.mLedgerIndex); - *reinterpret_cast(bufPtr + 4) = ntohl(obj.mLedgerIndex); - *(bufPtr + 8) = static_cast(obj.mType); - memcpy(bufPtr + 9, &obj.mData.front(), obj.mData.size()); - - batch.Put(leveldb::Slice(reinterpret_cast(obj.mHash.begin()), obj.mHash.size()), - leveldb::Slice(reinterpret_cast(bufPtr), rawData.size())); - } - - leveldb::Status st = theApp->getHashNodeLDB()->Write(leveldb::WriteOptions(), &batch); - if (!st.ok()) - { - WriteLog (lsFATAL, HashedObject) << "Failed to store hash node"; - assert(false); - } - } + LLWrite(set, theApp->getHashNodeLDB()); + if (mEphemeralDB) + LLWrite(set, theApp->getEphemeralLDB()); } } @@ -140,27 +188,28 @@ HashedObject::pointer HashedObjectStore::retrieveLevelDB(const uint256& hash) if (obj || mNegativeCache.isPresent(hash) || !theApp || !theApp->getHashNodeLDB()) return obj; - std::string sData; + if (mEphemeralDB) + { + obj = LLRetrieve(hash, theApp->getEphemeralLDB()); + if (obj) + return obj; + } { LoadEvent::autoptr event(theApp->getJobQueue().getLoadEventAP(jtHO_READ, "HOS::retrieve")); - leveldb::Status st = theApp->getHashNodeLDB()->Get(leveldb::ReadOptions(), - leveldb::Slice(reinterpret_cast(hash.begin()), hash.size()), &sData); - if (!st.ok()) + obj = LLRetrieve(hash, theApp->getHashNodeLDB()); + if (!obj) { - assert(st.IsNotFound()); + mNegativeCache.add(hash); return obj; } } - const unsigned char* bufPtr = reinterpret_cast(&sData[0]); - uint32 index = htonl(*reinterpret_cast(bufPtr)); - int htype = bufPtr[8]; - - obj = boost::make_shared(static_cast(htype), index, - bufPtr + 9, sData.size() - 9, hash); mCache.canonicalize(hash, obj); + if (mEphemeralDB) + LLWrite(obj, theApp->getEphemeralLDB()); + WriteLog (lsTRACE, HashedObject) << "HOS: " << hash << " fetch: in db"; return obj; } @@ -196,6 +245,7 @@ bool HashedObjectStore::storeSQLite(HashedObjectType type, uint32 index, // else // WriteLog (lsTRACE, HashedObject) << "HOS: already had " << hash; mNegativeCache.del(hash); + return true; } @@ -223,6 +273,9 @@ void HashedObjectStore::bulkWriteSQLite(Job&) #ifndef NO_SQLITE3_PREPARE + if (mEphemeralDB) + LLWrite(set, theApp->getEphemeralLDB()); + { Database* db = theApp->getHashNodeDB()->getDB(); static SqliteStatement pStB(db->getSqliteDB(), "BEGIN TRANSACTION;", !theConfig.RUN_STANDALONE); @@ -308,6 +361,13 @@ HashedObject::pointer HashedObjectStore::retrieveSQLite(const uint256& hash) if (mNegativeCache.isPresent(hash)) return obj; + if (mEphemeralDB) + { + obj = LLRetrieve(hash, theApp->getEphemeralLDB()); + if (obj) + return obj; + } + if (!theApp || !theApp->getHashNodeDB()) return obj; @@ -387,6 +447,9 @@ HashedObject::pointer HashedObjectStore::retrieveSQLite(const uint256& hash) obj = boost::make_shared(htype, index, data, hash); mCache.canonicalize(hash, obj); + if (mEphemeralDB) + LLWrite(obj, theApp->getEphemeralLDB()); + WriteLog (lsTRACE, HashedObject) << "HOS: " << hash << " fetch: in db"; return obj; } diff --git a/src/cpp/ripple/HashedObject.h b/src/cpp/ripple/HashedObject.h index afc6782848..25c05aeba6 100644 --- a/src/cpp/ripple/HashedObject.h +++ b/src/cpp/ripple/HashedObject.h @@ -69,7 +69,7 @@ protected: std::vector< boost::shared_ptr > mWriteSet; bool mWritePending; - bool mLevelDB; + bool mLevelDB, mEphemeralDB; public: