ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2026-04-29 15:37:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make database reads async

Browse Source 
* yield on db read using asio
* PostgresBackend fetches multiple transactions or objects in parallel
This commit is contained in:
natenichols
2022-02-07 07:00:15 -06:00
committed by CJ Cobb
parent 7c2bef70bc
commit d016253264
50 changed files with 3612 additions and 2593 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/backend/BackendFactory.h
View File

@@ -8,7 +8,7 @@
namespace Backend {
std::shared_ptr<BackendInterface>
make_Backend(boost::json::object const& config)
make_Backend(boost::asio::io_context& ioc, boost::json::object const& config)
{
BOOST_LOG_TRIVIAL(info) << __func__ << ": Constructing BackendInterface";
@@ -27,13 +27,13 @@ make_Backend(boost::json::object const& config)
if (config.contains("online_delete"))
dbConfig.at(type).as_object()["ttl"] =
config.at("online_delete").as_int64() * 4;
backend =
std::make_shared<CassandraBackend>(dbConfig.at(type).as_object());
backend = std::make_shared<CassandraBackend>(
ioc, dbConfig.at(type).as_object());
}
else if (boost::iequals(type, "postgres"))
{
backend =
std::make_shared<PostgresBackend>(dbConfig.at(type).as_object());
backend = std::make_shared<PostgresBackend>(
ioc, dbConfig.at(type).as_object());
}
if (!backend)

85
src/backend/BackendInterface.cpp
View File

@@ -3,7 +3,7 @@
#include <backend/BackendInterface.h>
namespace Backend {
bool
BackendInterface::finishWrites(uint32_t ledgerSequence)
BackendInterface::finishWrites(std::uint32_t const ledgerSequence)
{
auto commitRes = doFinishWrites();
if (commitRes)
@@ -15,7 +15,7 @@ BackendInterface::finishWrites(uint32_t ledgerSequence)
void
BackendInterface::writeLedgerObject(
std::string&& key,
uint32_t seq,
std::uint32_t const seq,
std::string&& blob)
{
assert(key.size() == sizeof(ripple::uint256));
@@ -23,17 +23,37 @@ BackendInterface::writeLedgerObject(
doWriteLedgerObject(std::move(key), seq, std::move(blob));
}
std::optional<LedgerRange>
BackendInterface::hardFetchLedgerRangeNoThrow(
boost::asio::yield_context& yield) const
{
BOOST_LOG_TRIVIAL(debug) << __func__;
while (true)
{
try
{
return hardFetchLedgerRange(yield);
}
catch (DatabaseTimeout& t)
{
;
}
}
}
std::optional<LedgerRange>
BackendInterface::hardFetchLedgerRangeNoThrow() const
{
BOOST_LOG_TRIVIAL(debug) << __func__;
return retryOnTimeout([&]() { return hardFetchLedgerRange(); });
}
// *** state data methods
std::optional<Blob>
BackendInterface::fetchLedgerObject(
ripple::uint256 const& key,
uint32_t sequence) const
std::uint32_t const sequence,
boost::asio::yield_context& yield) const
{
auto obj = cache_.get(key, sequence);
if (obj)
@@ -46,7 +66,7 @@ BackendInterface::fetchLedgerObject(
{
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " - cache miss - " << ripple::strHex(key);
auto dbObj = doFetchLedgerObject(key, sequence);
auto dbObj = doFetchLedgerObject(key, sequence, yield);
if (!dbObj)
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " - missed cache and missed in db";
@@ -60,7 +80,8 @@ BackendInterface::fetchLedgerObject(
std::vector<Blob>
BackendInterface::fetchLedgerObjects(
std::vector<ripple::uint256> const& keys,
uint32_t sequence) const
std::uint32_t const sequence,
boost::asio::yield_context& yield) const
{
std::vector<Blob> results;
results.resize(keys.size());
@@ -79,7 +100,7 @@ BackendInterface::fetchLedgerObjects(
if (misses.size())
{
auto objs = doFetchLedgerObjects(misses, sequence);
auto objs = doFetchLedgerObjects(misses, sequence, yield);
for (size_t i = 0, j = 0; i < results.size(); ++i)
{
if (results[i].size() == 0)
@@ -89,13 +110,15 @@ BackendInterface::fetchLedgerObjects(
}
}
}
return results;
}
// Fetches the successor to key/index
std::optional<ripple::uint256>
BackendInterface::fetchSuccessorKey(
ripple::uint256 key,
uint32_t ledgerSequence) const
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
auto succ = cache_.getSuccessor(key, ledgerSequence);
if (succ)
@@ -104,28 +127,32 @@ BackendInterface::fetchSuccessorKey(
else
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " - cache miss - " << ripple::strHex(key);
return succ ? succ->key : doFetchSuccessorKey(key, ledgerSequence);
return succ ? succ->key : doFetchSuccessorKey(key, ledgerSequence, yield);
}
std::optional<LedgerObject>
BackendInterface::fetchSuccessorObject(
ripple::uint256 key,
uint32_t ledgerSequence) const
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
auto succ = fetchSuccessorKey(key, ledgerSequence);
auto succ = fetchSuccessorKey(key, ledgerSequence, yield);
if (succ)
{
auto obj = fetchLedgerObject(*succ, ledgerSequence);
auto obj = fetchLedgerObject(*succ, ledgerSequence, yield);
assert(obj);
return {{*succ, *obj}};
}
return {};
}
BookOffersPage
BackendInterface::fetchBookOffers(
ripple::uint256 const& book,
uint32_t ledgerSequence,
std::uint32_t limit,
std::optional<ripple::uint256> const& cursor) const
std::uint32_t const ledgerSequence,
std::uint32_t const limit,
std::optional<ripple::uint256> const& cursor,
boost::asio::yield_context& yield) const
{
// TODO try to speed this up. This can take a few seconds. The goal is
// to get it down to a few hundred milliseconds.
@@ -139,14 +166,14 @@ BackendInterface::fetchBookOffers(
.count();
};
auto begin = std::chrono::system_clock::now();
uint32_t numSucc = 0;
uint32_t numPages = 0;
std::uint32_t numSucc = 0;
std::uint32_t numPages = 0;
long succMillis = 0;
long pageMillis = 0;
while (keys.size() < limit)
{
auto mid1 = std::chrono::system_clock::now();
auto offerDir = fetchSuccessorObject(uTipIndex, ledgerSequence);
auto offerDir = fetchSuccessorObject(uTipIndex, ledgerSequence, yield);
auto mid2 = std::chrono::system_clock::now();
numSucc++;
succMillis += getMillis(mid2 - mid1);
@@ -174,7 +201,8 @@ BackendInterface::fetchBookOffers(
break;
}
auto nextKey = ripple::keylet::page(uTipIndex, next);
auto nextDir = fetchLedgerObject(nextKey.key, ledgerSequence);
auto nextDir =
fetchLedgerObject(nextKey.key, ledgerSequence, yield);
assert(nextDir);
offerDir->blob = *nextDir;
offerDir->key = nextKey.key;
@@ -183,7 +211,7 @@ BackendInterface::fetchBookOffers(
pageMillis += getMillis(mid3 - mid2);
}
auto mid = std::chrono::system_clock::now();
auto objs = fetchLedgerObjects(keys, ledgerSequence);
auto objs = fetchLedgerObjects(keys, ledgerSequence, yield);
for (size_t i = 0; i < keys.size() && i < limit; ++i)
{
BOOST_LOG_TRIVIAL(debug)
@@ -215,9 +243,10 @@ BackendInterface::fetchBookOffers(
LedgerPage
BackendInterface::fetchLedgerPage(
std::optional<ripple::uint256> const& cursor,
std::uint32_t ledgerSequence,
std::uint32_t limit,
std::uint32_t limitHint) const
std::uint32_t const ledgerSequence,
std::uint32_t const limit,
std::uint32_t const limitHint,
boost::asio::yield_context& yield) const
{
LedgerPage page;
@@ -226,13 +255,14 @@ BackendInterface::fetchLedgerPage(
{
ripple::uint256 const& curCursor =
keys.size() ? keys.back() : cursor ? *cursor : firstKey;
auto succ = fetchSuccessorKey(curCursor, ledgerSequence);
auto succ = fetchSuccessorKey(curCursor, ledgerSequence, yield);
if (!succ)
break;
keys.push_back(std::move(*succ));
}
auto objects = fetchLedgerObjects(keys, ledgerSequence);
auto objects = fetchLedgerObjects(keys, ledgerSequence, yield);
for (size_t i = 0; i < objects.size(); ++i)
{
assert(objects[i].size());
@@ -240,16 +270,19 @@ BackendInterface::fetchLedgerPage(
}
if (page.objects.size() >= limit)
page.cursor = page.objects.back().key;
return page;
}
std::optional<ripple::Fees>
BackendInterface::fetchFees(std::uint32_t seq) const
BackendInterface::fetchFees(
std::uint32_t const seq,
boost::asio::yield_context& yield) const
{
ripple::Fees fees;
auto key = ripple::keylet::fees().key;
auto bytes = fetchLedgerObject(key, seq);
auto bytes = fetchLedgerObject(key, seq, yield);
if (!bytes)
{

161
src/backend/BackendInterface.h
View File

@@ -37,12 +37,35 @@ retryOnTimeout(F func, size_t waitMs = 500)
}
}
// Please note, this function only works w/ non-void return type. Writes are
// synchronous anyways, so
template <class F>
void
synchronous(F&& f)
{
boost::asio::io_context ctx;
std::optional<boost::asio::io_context::work> work;
work.emplace(ctx);
boost::asio::spawn(ctx, [&f, &work](boost::asio::yield_context yield) {
f(yield);
work.reset();
});
ctx.run();
}
class BackendInterface
{
protected:
std::optional<LedgerRange> range;
SimpleCache cache_;
// mutex used for open() and close()
mutable std::mutex mutex_;
public:
BackendInterface(boost::json::object const& config)
{
@@ -72,98 +95,150 @@ public:
}
virtual std::optional<ripple::LedgerInfo>
fetchLedgerBySequence(uint32_t sequence) const = 0;
fetchLedgerBySequence(
std::uint32_t const sequence,
boost::asio::yield_context& yield) const = 0;
virtual std::optional<ripple::LedgerInfo>
fetchLedgerByHash(ripple::uint256 const& hash) const = 0;
fetchLedgerByHash(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const = 0;
virtual std::optional<uint32_t>
fetchLatestLedgerSequence() const = 0;
virtual std::optional<std::uint32_t>
fetchLatestLedgerSequence(boost::asio::yield_context& yield) const = 0;
std::optional<LedgerRange>
fetchLedgerRange() const
{
std::lock_guard lk(mutex_);
return range;
}
std::optional<ripple::Fees>
fetchFees(std::uint32_t seq) const;
fetchFees(std::uint32_t const seq, boost::asio::yield_context& yield) const;
// *** transaction methods
virtual std::optional<TransactionAndMetadata>
fetchTransaction(ripple::uint256 const& hash) const = 0;
fetchTransaction(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const = 0;
virtual std::vector<TransactionAndMetadata>
fetchTransactions(std::vector<ripple::uint256> const& hashes) const = 0;
fetchTransactions(
std::vector<ripple::uint256> const& hashes,
boost::asio::yield_context& yield) const = 0;
virtual AccountTransactions
fetchAccountTransactions(
ripple::AccountID const& account,
std::uint32_t limit,
bool forward = false,
std::optional<AccountTransactionsCursor> const& cursor = {}) const = 0;
std::uint32_t const limit,
bool forward,
std::optional<AccountTransactionsCursor> const& cursor,
boost::asio::yield_context& yield) const = 0;
virtual std::vector<TransactionAndMetadata>
fetchAllTransactionsInLedger(uint32_t ledgerSequence) const = 0;
fetchAllTransactionsInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const = 0;
virtual std::vector<ripple::uint256>
fetchAllTransactionHashesInLedger(uint32_t ledgerSequence) const = 0;
fetchAllTransactionHashesInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const = 0;
// *** state data methods
std::optional<Blob>
fetchLedgerObject(ripple::uint256 const& key, uint32_t sequence) const;
fetchLedgerObject(
ripple::uint256 const& key,
std::uint32_t const sequence,
boost::asio::yield_context& yield) const;
std::vector<Blob>
fetchLedgerObjects(
std::vector<ripple::uint256> const& keys,
uint32_t sequence) const;
std::uint32_t const sequence,
boost::asio::yield_context& yield) const;
virtual std::optional<Blob>
doFetchLedgerObject(ripple::uint256 const& key, uint32_t sequence)
const = 0;
doFetchLedgerObject(
ripple::uint256 const& key,
std::uint32_t const sequence,
boost::asio::yield_context& yield) const = 0;
virtual std::vector<Blob>
doFetchLedgerObjects(
std::vector<ripple::uint256> const& keys,
uint32_t sequence) const = 0;
std::uint32_t const sequence,
boost::asio::yield_context& yield) const = 0;
virtual std::vector<LedgerObject>
fetchLedgerDiff(uint32_t ledgerSequence) const = 0;
fetchLedgerDiff(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const = 0;
// Fetches a page of ledger objects, ordered by key/index.
// Used by ledger_data
LedgerPage
fetchLedgerPage(
std::optional<ripple::uint256> const& cursor,
std::uint32_t ledgerSequence,
std::uint32_t limit,
std::uint32_t limitHint = 0) const;
std::uint32_t const ledgerSequence,
std::uint32_t const limit,
std::uint32_t const limitHint,
boost::asio::yield_context& yield) const;
// Fetches the successor to key/index
std::optional<LedgerObject>
fetchSuccessorObject(ripple::uint256 key, uint32_t ledgerSequence) const;
fetchSuccessorObject(
ripple::uint256 key,
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const;
std::optional<ripple::uint256>
fetchSuccessorKey(ripple::uint256 key, uint32_t ledgerSequence) const;
fetchSuccessorKey(
ripple::uint256 key,
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const;
// Fetches the successor to key/index
virtual std::optional<ripple::uint256>
doFetchSuccessorKey(ripple::uint256 key, uint32_t ledgerSequence) const = 0;
doFetchSuccessorKey(
ripple::uint256 key,
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const = 0;
BookOffersPage
fetchBookOffers(
ripple::uint256 const& book,
uint32_t ledgerSequence,
std::uint32_t limit,
std::optional<ripple::uint256> const& cursor = {}) const;
std::uint32_t const ledgerSequence,
std::uint32_t const limit,
std::optional<ripple::uint256> const& cursor,
boost::asio::yield_context& yield) const;
std::optional<LedgerRange>
hardFetchLedgerRange() const
{
std::optional<LedgerRange> range = {};
synchronous([&](boost::asio::yield_context yield) {
range = hardFetchLedgerRange(yield);
});
return range;
}
virtual std::optional<LedgerRange>
hardFetchLedgerRange() const = 0;
hardFetchLedgerRange(boost::asio::yield_context& yield) const = 0;
// Doesn't throw DatabaseTimeout. Should be used with care.
std::optional<LedgerRange>
hardFetchLedgerRangeNoThrow() const;
// Doesn't throw DatabaseTimeout. Should be used with care.
std::optional<LedgerRange>
hardFetchLedgerRangeNoThrow(boost::asio::yield_context& yield) const;
void
updateRange(uint32_t newMax)
updateRange(std::uint32_t const newMax)
{
std::lock_guard lk(mutex_);
if (!range)
range = {newMax, newMax};
else
@@ -175,14 +250,17 @@ public:
ripple::LedgerInfo const& ledgerInfo,
std::string&& ledgerHeader) = 0;
void
writeLedgerObject(std::string&& key, uint32_t seq, std::string&& blob);
virtual void
writeLedgerObject(
std::string&& key,
std::uint32_t const seq,
std::string&& blob);
virtual void
writeTransaction(
std::string&& hash,
uint32_t seq,
uint32_t date,
std::uint32_t const seq,
std::uint32_t const date,
std::string&& transaction,
std::string&& metadata) = 0;
@@ -192,20 +270,23 @@ public:
virtual void
writeSuccessor(
std::string&& key,
uint32_t seq,
std::uint32_t const seq,
std::string&& successor) = 0;
// Tell the database we are about to begin writing data for a particular
// ledger.
virtual void
startWrites() = 0;
startWrites() const = 0;
// Tell the database we have finished writing all data for a particular
// ledger
bool
finishWrites(uint32_t ledgerSequence);
finishWrites(std::uint32_t const ledgerSequence);
virtual bool
doOnlineDelete(uint32_t numLedgersToKeep) const = 0;
doOnlineDelete(
std::uint32_t numLedgersToKeep,
boost::asio::yield_context& yield) const = 0;
// Open the database. Set up all of the necessary objects and
// datastructures. After this call completes, the database is ready for
@@ -215,18 +296,18 @@ public:
// Close the database, releasing any resources
virtual void
close() = 0;
close(){};
// *** private helper methods
private:
virtual void
doWriteLedgerObject(
std::string&& key,
uint32_t seq,
std::uint32_t const seq,
std::string&& blob) = 0;
virtual bool
doFinishWrites() = 0;
doFinishWrites() const = 0;
};
} // namespace Backend

190
src/backend/CassandraBackend.cpp
View File

@@ -3,6 +3,13 @@
#include <functional>
#include <unordered_map>
namespace Backend {
// Type alias for async completion handlers
using completion_token = boost::asio::yield_context;
using function_type = void(boost::system::error_code);
using result_type = boost::asio::async_result<completion_token, function_type>;
using handler_type = typename result_type::completion_handler_type;
template <class T, class F>
void
processAsyncWriteResponse(T& requestParams, CassFuture* fut, F func)
@@ -50,7 +57,7 @@ struct WriteCallbackData
CassandraBackend const* backend;
T data;
std::function<void(WriteCallbackData<T, B>&, bool)> retry;
uint32_t currentRetries;
std::uint32_t currentRetries;
std::atomic<int> refs = 1;
std::string id;
@@ -95,6 +102,7 @@ struct WriteCallbackData
return id;
}
};
template <class T, class B>
struct BulkWriteCallbackData : public WriteCallbackData<T, B>
{
@@ -162,7 +170,7 @@ makeAndExecuteBulkAsyncWrite(
void
CassandraBackend::doWriteLedgerObject(
std::string&& key,
uint32_t seq,
std::uint32_t const seq,
std::string&& blob)
{
BOOST_LOG_TRIVIAL(trace) << "Writing ledger object to cassandra";
@@ -196,7 +204,7 @@ CassandraBackend::doWriteLedgerObject(
void
CassandraBackend::writeSuccessor(
std::string&& key,
uint32_t seq,
std::uint32_t const seq,
std::string&& successor)
{
BOOST_LOG_TRIVIAL(trace)
@@ -277,8 +285,8 @@ CassandraBackend::writeAccountTransactions(
void
CassandraBackend::writeTransaction(
std::string&& hash,
uint32_t seq,
uint32_t date,
std::uint32_t const seq,
std::uint32_t const date,
std::string&& transaction,
std::string&& metadata)
{
@@ -317,11 +325,12 @@ CassandraBackend::writeTransaction(
}
std::optional<LedgerRange>
CassandraBackend::hardFetchLedgerRange() const
CassandraBackend::hardFetchLedgerRange(boost::asio::yield_context& yield) const
{
BOOST_LOG_TRIVIAL(trace) << "Fetching from cassandra";
CassandraStatement statement{selectLedgerRange_};
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result)
{
BOOST_LOG_TRIVIAL(error) << __func__ << " - no rows";
@@ -339,26 +348,31 @@ CassandraBackend::hardFetchLedgerRange() const
}
return range;
}
std::vector<TransactionAndMetadata>
CassandraBackend::fetchAllTransactionsInLedger(uint32_t ledgerSequence) const
CassandraBackend::fetchAllTransactionsInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
auto hashes = fetchAllTransactionHashesInLedger(ledgerSequence);
return fetchTransactions(hashes);
auto hashes = fetchAllTransactionHashesInLedger(ledgerSequence, yield);
return fetchTransactions(hashes, yield);
}
template <class Result>
struct ReadCallbackData
{
std::function<void(CassandraResult&)> onSuccess;
using handler_type = typename Result::completion_handler_type;
std::atomic_int& numOutstanding;
std::mutex& mtx;
std::condition_variable& cv;
handler_type handler;
std::function<void(CassandraResult&)> onSuccess;
std::atomic_bool errored = false;
ReadCallbackData(
std::atomic_int& numOutstanding,
std::mutex& m,
std::condition_variable& cv,
handler_type& handler,
std::function<void(CassandraResult&)> onSuccess)
: numOutstanding(numOutstanding), mtx(m), cv(cv), onSuccess(onSuccess)
: numOutstanding(numOutstanding), handler(handler), onSuccess(onSuccess)
{
}
@@ -375,35 +389,55 @@ struct ReadCallbackData
CassandraResult result{cass_future_get_result(fut)};
onSuccess(result);
}
std::lock_guard lck(mtx);
if (--numOutstanding == 0)
cv.notify_one();
resume();
}
void
resume()
{
boost::asio::post(
boost::asio::get_associated_executor(handler),
[handler = std::move(handler)]() mutable {
handler(boost::system::error_code{});
});
}
};
void
processAsyncRead(CassFuture* fut, void* cbData)
{
ReadCallbackData& cb = *static_cast<ReadCallbackData*>(cbData);
ReadCallbackData<result_type>& cb =
*static_cast<ReadCallbackData<result_type>*>(cbData);
cb.finish(fut);
}
std::vector<TransactionAndMetadata>
CassandraBackend::fetchTransactions(
std::vector<ripple::uint256> const& hashes) const
std::vector<ripple::uint256> const& hashes,
boost::asio::yield_context& yield) const
{
if (hashes.size() == 0)
return {};
handler_type handler(std::forward<decltype(yield)>(yield));
result_type result(handler);
std::size_t const numHashes = hashes.size();
std::atomic_int numOutstanding = numHashes;
std::condition_variable cv;
std::mutex mtx;
std::vector<TransactionAndMetadata> results{numHashes};
std::vector<std::shared_ptr<ReadCallbackData>> cbs;
std::vector<std::shared_ptr<ReadCallbackData<result_type>>> cbs;
cbs.reserve(numHashes);
auto start = std::chrono::system_clock::now();
for (std::size_t i = 0; i < hashes.size(); ++i)
{
CassandraStatement statement{selectTransaction_};
statement.bindNextBytes(hashes[i]);
cbs.push_back(std::make_shared<ReadCallbackData>(
numOutstanding, mtx, cv, [i, &results](auto& result) {
cbs.push_back(std::make_shared<ReadCallbackData<result_type>>(
numOutstanding, handler, [i, &results](auto& result) {
if (result.hasResult())
results[i] = {
result.getBytes(),
@@ -411,12 +445,14 @@ CassandraBackend::fetchTransactions(
result.getUInt32(),
result.getUInt32()};
}));
executeAsyncRead(statement, processAsyncRead, *cbs[i]);
}
assert(results.size() == cbs.size());
std::unique_lock<std::mutex> lck(mtx);
cv.wait(lck, [&numOutstanding]() { return numOutstanding == 0; });
// suspend the coroutine until completion handler is called.
result.get();
auto end = std::chrono::system_clock::now();
for (auto const& cb : cbs)
{
@@ -431,14 +467,18 @@ CassandraBackend::fetchTransactions(
<< " milliseconds";
return results;
}
std::vector<ripple::uint256>
CassandraBackend::fetchAllTransactionHashesInLedger(
uint32_t ledgerSequence) const
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
CassandraStatement statement{selectAllTransactionHashesInLedger_};
statement.bindNextInt(ledgerSequence);
auto start = std::chrono::system_clock::now();
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
auto end = std::chrono::system_clock::now();
if (!result)
{
@@ -464,9 +504,10 @@ CassandraBackend::fetchAllTransactionHashesInLedger(
AccountTransactions
CassandraBackend::fetchAccountTransactions(
ripple::AccountID const& account,
std::uint32_t limit,
bool forward,
std::optional<AccountTransactionsCursor> const& cursorIn) const
std::uint32_t const limit,
bool const forward,
std::optional<AccountTransactionsCursor> const& cursorIn,
boost::asio::yield_context& yield) const
{
auto rng = fetchLedgerRange();
if (!rng)
@@ -494,7 +535,8 @@ CassandraBackend::fetchAccountTransactions(
else
{
int seq = forward ? rng->minSequence : rng->maxSequence;
int placeHolder = forward ? 0 : std::numeric_limits<uint32_t>::max();
int placeHolder =
forward ? 0 : std::numeric_limits<std::uint32_t>::max();
statement.bindNextIntTuple(placeHolder, placeHolder);
BOOST_LOG_TRIVIAL(debug)
@@ -503,7 +545,8 @@ CassandraBackend::fetchAccountTransactions(
}
statement.bindNextUInt(limit);
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result.hasResult())
{
BOOST_LOG_TRIVIAL(debug) << __func__ << " - no rows returned";
@@ -520,13 +563,16 @@ CassandraBackend::fetchAccountTransactions(
{
BOOST_LOG_TRIVIAL(debug) << __func__ << " setting cursor";
auto [lgrSeq, txnIdx] = result.getInt64Tuple();
cursor = {(uint32_t)lgrSeq, (uint32_t)txnIdx};
cursor = {
static_cast<std::uint32_t>(lgrSeq),
static_cast<std::uint32_t>(txnIdx)};
if (forward)
++cursor->transactionIndex;
}
} while (result.nextRow());
auto txns = fetchTransactions(hashes);
auto txns = fetchTransactions(hashes, yield);
BOOST_LOG_TRIVIAL(debug) << __func__ << "txns = " << txns.size();
if (txns.size() == limit)
@@ -540,13 +586,16 @@ CassandraBackend::fetchAccountTransactions(
std::optional<ripple::uint256>
CassandraBackend::doFetchSuccessorKey(
ripple::uint256 key,
uint32_t ledgerSequence) const
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
BOOST_LOG_TRIVIAL(trace) << "Fetching from cassandra";
CassandraStatement statement{selectSuccessor_};
statement.bindNextBytes(key);
statement.bindNextInt(ledgerSequence);
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result)
{
BOOST_LOG_TRIVIAL(debug) << __func__ << " - no rows";
@@ -557,16 +606,20 @@ CassandraBackend::doFetchSuccessorKey(
return {};
return next;
}
std::optional<Blob>
CassandraBackend::doFetchLedgerObject(
ripple::uint256 const& key,
uint32_t sequence) const
std::uint32_t const sequence,
boost::asio::yield_context& yield) const
{
BOOST_LOG_TRIVIAL(trace) << "Fetching from cassandra";
CassandraStatement statement{selectObject_};
statement.bindNextBytes(key);
statement.bindNextInt(sequence);
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result)
{
BOOST_LOG_TRIVIAL(debug) << __func__ << " - no rows";
@@ -581,21 +634,26 @@ CassandraBackend::doFetchLedgerObject(
std::vector<Blob>
CassandraBackend::doFetchLedgerObjects(
std::vector<ripple::uint256> const& keys,
uint32_t sequence) const
std::uint32_t const sequence,
boost::asio::yield_context& yield) const
{
if (keys.size() == 0)
return {};
handler_type handler(std::forward<decltype(yield)>(yield));
result_type result(handler);
std::size_t const numKeys = keys.size();
BOOST_LOG_TRIVIAL(trace)
<< "Fetching " << numKeys << " records from Cassandra";
std::atomic_int numOutstanding = numKeys;
std::condition_variable cv;
std::mutex mtx;
std::vector<Blob> results{numKeys};
std::vector<std::shared_ptr<ReadCallbackData>> cbs;
std::vector<std::shared_ptr<ReadCallbackData<result_type>>> cbs;
cbs.reserve(numKeys);
for (std::size_t i = 0; i < keys.size(); ++i)
{
cbs.push_back(std::make_shared<ReadCallbackData>(
numOutstanding, mtx, cv, [i, &results](auto& result) {
cbs.push_back(std::make_shared<ReadCallbackData<result_type>>(
numOutstanding, handler, [i, &results](auto& result) {
if (result.hasResult())
results[i] = result.getBytes();
}));
@@ -606,8 +664,9 @@ CassandraBackend::doFetchLedgerObjects(
}
assert(results.size() == cbs.size());
std::unique_lock<std::mutex> lck(mtx);
cv.wait(lck, [&numOutstanding]() { return numOutstanding == 0; });
// suspend the coroutine until completion handler is called.
result.get();
for (auto const& cb : cbs)
{
if (cb->errored)
@@ -618,14 +677,20 @@ CassandraBackend::doFetchLedgerObjects(
<< "Fetched " << numKeys << " records from Cassandra";
return results;
}
std::vector<LedgerObject>
CassandraBackend::fetchLedgerDiff(uint32_t ledgerSequence) const
CassandraBackend::fetchLedgerDiff(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
CassandraStatement statement{selectDiff_};
statement.bindNextInt(ledgerSequence);
auto start = std::chrono::system_clock::now();
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
auto end = std::chrono::system_clock::now();
if (!result)
{
BOOST_LOG_TRIVIAL(error)
@@ -643,7 +708,7 @@ CassandraBackend::fetchLedgerDiff(uint32_t ledgerSequence) const
<< std::chrono::duration_cast<std::chrono::milliseconds>(end - start)
.count()
<< " milliseconds";
auto objs = fetchLedgerObjects(keys, ledgerSequence);
auto objs = fetchLedgerObjects(keys, ledgerSequence, yield);
std::vector<LedgerObject> results;
std::transform(
keys.begin(),
@@ -657,7 +722,9 @@ CassandraBackend::fetchLedgerDiff(uint32_t ledgerSequence) const
}
bool
CassandraBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
CassandraBackend::doOnlineDelete(
std::uint32_t const numLedgersToKeep,
boost::asio::yield_context& yield) const
{
// calculate TTL
// ledgers close roughly every 4 seconds. We double the TTL so that way
@@ -666,7 +733,7 @@ CassandraBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
auto rng = fetchLedgerRange();
if (!rng)
return false;
uint32_t minLedger = rng->maxSequence - numLedgersToKeep;
std::uint32_t minLedger = rng->maxSequence - numLedgersToKeep;
if (minLedger <= rng->minSequence)
return false;
auto bind = [this](auto& params) {
@@ -680,18 +747,19 @@ CassandraBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
std::condition_variable cv;
std::mutex mtx;
std::vector<std::shared_ptr<BulkWriteCallbackData<
std::tuple<ripple::uint256, uint32_t, Blob>,
std::tuple<ripple::uint256, std::uint32_t, Blob>,
typename std::remove_reference<decltype(bind)>::type>>>
cbs;
uint32_t concurrentLimit = 10;
std::uint32_t concurrentLimit = 10;
std::atomic_int numOutstanding = 0;
// iterate through latest ledger, updating TTL
std::optional<ripple::uint256> cursor;
while (true)
{
auto [objects, curCursor, warning] = retryOnTimeout(
[&]() { return fetchLedgerPage(cursor, minLedger, 256); });
auto [objects, curCursor, warning] = retryOnTimeout([&]() {
return fetchLedgerPage(cursor, minLedger, 256, 0, yield);
});
if (warning)
{
BOOST_LOG_TRIVIAL(warning)
@@ -830,9 +898,7 @@ CassandraBackend::open(bool readOnly)
std::string username = getString("username");
if (username.size())
{
BOOST_LOG_TRIVIAL(debug)
<< "user = " << username.c_str()
<< " password = " << getString("password").c_str();
BOOST_LOG_TRIVIAL(debug) << "user = " << username.c_str();
cass_cluster_set_credentials(
cluster, username.c_str(), getString("password").c_str());
}
@@ -1282,10 +1348,8 @@ CassandraBackend::open(bool readOnly)
setupPreparedStatements = true;
}
work_.emplace(ioContext_);
ioThread_ = std::thread{[this]() { ioContext_.run(); }};
open_ = true;
BOOST_LOG_TRIVIAL(info) << "Opened CassandraBackend successfully";
} // namespace Backend
}
} // namespace Backend

266
src/backend/CassandraBackend.h
View File

@@ -3,6 +3,8 @@
#include <ripple/basics/base_uint.h>
#include <boost/asio.hpp>
#include <boost/asio/async_result.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/filesystem.hpp>
#include <boost/json.hpp>
#include <boost/log/trivial.hpp>
@@ -97,6 +99,7 @@ public:
curBindingIndex_ = other.curBindingIndex_;
other.curBindingIndex_ = 0;
}
CassandraStatement(CassandraStatement const& other) = delete;
CassStatement*
@@ -125,9 +128,9 @@ public:
}
void
bindNextBytes(const char* data, uint32_t size)
bindNextBytes(const char* data, std::uint32_t const size)
{
bindNextBytes((unsigned char*)data, size);
bindNextBytes((unsigned const char*)(data), size);
}
void
@@ -153,13 +156,13 @@ public:
}
void
bindNextBytes(void const* key, uint32_t size)
bindNextBytes(void const* key, std::uint32_t const size)
{
bindNextBytes(static_cast<const unsigned char*>(key), size);
}
void
bindNextBytes(const unsigned char* data, uint32_t size)
bindNextBytes(const unsigned char* data, std::uint32_t const size)
{
if (!statement_)
throw std::runtime_error(
@@ -181,7 +184,7 @@ public:
}
void
bindNextUInt(uint32_t value)
bindNextUInt(std::uint32_t const value)
{
if (!statement_)
throw std::runtime_error(
@@ -202,9 +205,9 @@ public:
}
void
bindNextInt(uint32_t value)
bindNextInt(std::uint32_t const value)
{
bindNextInt((int64_t)value);
bindNextInt(static_cast<std::int64_t>(value));
}
void
@@ -227,7 +230,7 @@ public:
}
void
bindNextIntTuple(uint32_t first, uint32_t second)
bindNextIntTuple(std::uint32_t const first, std::uint32_t const second)
{
CassTuple* tuple = cass_tuple_new(2);
CassError rc = cass_tuple_set_int64(tuple, 0, first);
@@ -366,26 +369,6 @@ public:
curGetIndex_++;
return {buf, buf + bufSize};
}
/*
uint32_t
getNumBytes()
{
if (!row_)
throw std::runtime_error("CassandraResult::getBytes - no result");
cass_byte_t const* buf;
std::size_t bufSize;
CassError rc = cass_value_get_bytes(
cass_row_get_column(row_, curGetIndex_), &buf, &bufSize);
if (rc != CASS_OK)
{
std::stringstream msg;
msg << "CassandraResult::getBytes - error getting value: " << rc
<< ", " << cass_error_desc(rc);
BOOST_LOG_TRIVIAL(error) << msg.str();
throw std::runtime_error(msg.str());
}
return bufSize;
}*/
ripple::uint256
getUInt256()
@@ -428,13 +411,13 @@ public:
return val;
}
uint32_t
std::uint32_t
getUInt32()
{
return (uint32_t)getInt64();
return static_cast<std::uint32_t>(getInt64());
}
std::pair<int64_t, int64_t>
std::pair<std::int64_t, std::int64_t>
getInt64Tuple()
{
if (!row_)
@@ -446,13 +429,13 @@ public:
throw std::runtime_error(
"CassandraResult::getInt64Tuple - failed to iterate tuple");
CassValue const* value = cass_iterator_get_value(tupleIter);
int64_t first;
std::int64_t first;
cass_value_get_int64(value, &first);
if (!cass_iterator_next(tupleIter))
throw std::runtime_error(
"CassandraResult::getInt64Tuple - failed to iterate tuple");
value = cass_iterator_get_value(tupleIter);
int64_t second;
std::int64_t second;
cass_value_get_int64(value, &second);
++curGetIndex_;
return {first, second};
@@ -506,6 +489,52 @@ isTimeout(CassError rc)
return false;
}
template <typename CompletionToken>
CassError
cass_future_error_code(CassFuture* fut, CompletionToken&& token)
{
using function_type = void(boost::system::error_code, CassError);
using result_type =
boost::asio::async_result<CompletionToken, function_type>;
using handler_type = typename result_type::completion_handler_type;
handler_type handler(std::forward<decltype(token)>(token));
result_type result(handler);
struct HandlerWrapper
{
handler_type handler;
HandlerWrapper(handler_type&& handler_) : handler(std::move(handler_))
{
}
};
auto resume = [](CassFuture* fut, void* data) -> void {
HandlerWrapper* hw = (HandlerWrapper*)data;
boost::asio::post(
boost::asio::get_associated_executor(hw->handler),
[fut, hw, handler = std::move(hw->handler)]() mutable {
delete hw;
handler(
boost::system::error_code{}, cass_future_error_code(fut));
});
};
HandlerWrapper* wrapper = new HandlerWrapper(std::move(handler));
cass_future_set_callback(fut, resume, wrapper);
// Suspend the coroutine until completion handler is called.
// The handler will populate rc, the error code describing
// the state of the cassandra future.
auto rc = result.get();
return rc;
}
class CassandraBackend : public BackendInterface
{
private:
@@ -529,9 +558,6 @@ private:
std::atomic<bool> open_{false};
// mutex used for open() and close()
std::mutex mutex_;
std::unique_ptr<CassSession, void (*)(CassSession*)> session_{
nullptr,
[](CassSession* session) {
@@ -571,17 +597,12 @@ private:
CassandraPreparedStatement selectLatestLedger_;
CassandraPreparedStatement selectLedgerRange_;
// io_context used for exponential backoff for write retries
mutable boost::asio::io_context ioContext_;
std::optional<boost::asio::io_context::work> work_;
std::thread ioThread_;
// maximum number of concurrent in flight requests. New requests will wait
// for earlier requests to finish if this limit is exceeded
uint32_t maxRequestsOutstanding = 10000;
std::uint32_t maxRequestsOutstanding = 10000;
// we keep this small because the indexer runs in the background, and we
// don't want the database to be swamped when the indexer is running
uint32_t indexerMaxRequestsOutstanding = 10;
std::uint32_t indexerMaxRequestsOutstanding = 10;
mutable std::atomic_uint32_t numRequestsOutstanding_ = 0;
// mutex and condition_variable to limit the number of concurrent in flight
@@ -594,22 +615,40 @@ private:
mutable std::mutex syncMutex_;
mutable std::condition_variable syncCv_;
// io_context for read/write retries
mutable boost::asio::io_context ioContext_;
std::optional<boost::asio::io_context::work> work_;
std::thread ioThread_;
boost::json::object config_;
mutable uint32_t ledgerSequence_ = 0;
mutable std::uint32_t ledgerSequence_ = 0;
public:
CassandraBackend(boost::json::object const& config)
CassandraBackend(
boost::asio::io_context& ioc,
boost::json::object const& config)
: BackendInterface(config), config_(config)
{
work_.emplace(ioContext_);
ioThread_ = std::thread([this]() { ioContext_.run(); });
}
~CassandraBackend() override
{
work_.reset();
ioThread_.join();
if (open_)
close();
}
boost::asio::io_context&
getIOContext() const
{
return ioContext_;
}
bool
isOpen()
{
@@ -626,23 +665,19 @@ public:
void
close() override
{
{
std::lock_guard<std::mutex> lock(mutex_);
work_.reset();
ioThread_.join();
}
open_ = false;
}
AccountTransactions
fetchAccountTransactions(
ripple::AccountID const& account,
std::uint32_t limit,
std::uint32_t const limit,
bool forward,
std::optional<AccountTransactionsCursor> const& cursor) const override;
std::optional<AccountTransactionsCursor> const& cursor,
boost::asio::yield_context& yield) const override;
bool
doFinishWrites() override
doFinishWrites() const override
{
// wait for all other writes to finish
sync();
@@ -674,12 +709,12 @@ public:
writeLedger(ripple::LedgerInfo const& ledgerInfo, std::string&& header)
override;
std::optional<uint32_t>
fetchLatestLedgerSequence() const override
std::optional<std::uint32_t>
fetchLatestLedgerSequence(boost::asio::yield_context& yield) const override
{
BOOST_LOG_TRIVIAL(trace) << __func__;
CassandraStatement statement{selectLatestLedger_};
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result.hasResult())
{
BOOST_LOG_TRIVIAL(error)
@@ -690,13 +725,14 @@ public:
}
std::optional<ripple::LedgerInfo>
fetchLedgerBySequence(uint32_t sequence) const override
fetchLedgerBySequence(
std::uint32_t const sequence,
boost::asio::yield_context& yield) const override
{
BOOST_LOG_TRIVIAL(trace) << __func__;
CassandraStatement statement{selectLedgerBySeq_};
statement.bindNextInt(sequence);
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result)
{
BOOST_LOG_TRIVIAL(error) << __func__ << " - no rows";
@@ -707,46 +743,57 @@ public:
}
std::optional<ripple::LedgerInfo>
fetchLedgerByHash(ripple::uint256 const& hash) const override
fetchLedgerByHash(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const override
{
CassandraStatement statement{selectLedgerByHash_};
statement.bindNextBytes(hash);
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result.hasResult())
{
BOOST_LOG_TRIVIAL(debug) << __func__ << " - no rows returned";
return {};
}
std::uint32_t sequence = result.getInt64();
std::uint32_t const sequence = result.getInt64();
return fetchLedgerBySequence(sequence);
return fetchLedgerBySequence(sequence, yield);
}
std::optional<LedgerRange>
hardFetchLedgerRange() const override;
hardFetchLedgerRange(boost::asio::yield_context& yield) const override;
std::vector<TransactionAndMetadata>
fetchAllTransactionsInLedger(uint32_t ledgerSequence) const override;
fetchAllTransactionsInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
std::vector<ripple::uint256>
fetchAllTransactionHashesInLedger(uint32_t ledgerSequence) const override;
fetchAllTransactionHashesInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
// Synchronously fetch the object with key key, as of ledger with sequence
// sequence
std::optional<Blob>
doFetchLedgerObject(ripple::uint256 const& key, uint32_t sequence)
const override;
doFetchLedgerObject(
ripple::uint256 const& key,
std::uint32_t const sequence,
boost::asio::yield_context& yield) const override;
std::optional<int64_t>
getToken(void const* key) const
getToken(void const* key, boost::asio::yield_context& yield) const
{
BOOST_LOG_TRIVIAL(trace) << "Fetching from cassandra";
CassandraStatement statement{getToken_};
statement.bindNextBytes(key, 32);
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result)
{
BOOST_LOG_TRIVIAL(error) << __func__ << " - no rows";
@@ -760,12 +807,15 @@ public:
}
std::optional<TransactionAndMetadata>
fetchTransaction(ripple::uint256 const& hash) const override
fetchTransaction(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const override
{
BOOST_LOG_TRIVIAL(trace) << __func__;
CassandraStatement statement{selectTransaction_};
statement.bindNextBytes(hash);
CassandraResult result = executeSyncRead(statement);
CassandraResult result = executeAsyncRead(statement, yield);
if (!result)
{
BOOST_LOG_TRIVIAL(error) << __func__ << " - no rows";
@@ -777,29 +827,40 @@ public:
result.getUInt32(),
result.getUInt32()}};
}
std::optional<ripple::uint256>
doFetchSuccessorKey(ripple::uint256 key, uint32_t ledgerSequence)
const override;
doFetchSuccessorKey(
ripple::uint256 key,
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
std::vector<TransactionAndMetadata>
fetchTransactions(
std::vector<ripple::uint256> const& hashes) const override;
std::vector<ripple::uint256> const& hashes,
boost::asio::yield_context& yield) const override;
std::vector<Blob>
doFetchLedgerObjects(
std::vector<ripple::uint256> const& keys,
uint32_t sequence) const override;
std::uint32_t const sequence,
boost::asio::yield_context& yield) const override;
std::vector<LedgerObject>
fetchLedgerDiff(uint32_t ledgerSequence) const override;
fetchLedgerDiff(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
void
doWriteLedgerObject(std::string&& key, uint32_t seq, std::string&& blob)
override;
doWriteLedgerObject(
std::string&& key,
std::uint32_t const seq,
std::string&& blob) override;
void
writeSuccessor(std::string&& key, uint32_t seq, std::string&& successor)
override;
writeSuccessor(
std::string&& key,
std::uint32_t const seq,
std::string&& successor) override;
void
writeAccountTransactions(
@@ -808,13 +869,13 @@ public:
void
writeTransaction(
std::string&& hash,
uint32_t seq,
uint32_t date,
std::uint32_t const seq,
std::uint32_t const date,
std::string&& transaction,
std::string&& metadata) override;
void
startWrites() override
startWrites() const override
{
}
@@ -825,14 +886,11 @@ public:
syncCv_.wait(lck, [this]() { return finishedAllRequests(); });
}
bool
doOnlineDelete(uint32_t numLedgersToKeep) const override;
boost::asio::io_context&
getIOContext() const
{
return ioContext_;
}
bool
doOnlineDelete(
std::uint32_t const numLedgersToKeep,
boost::asio::yield_context& yield) const override;
inline void
incremementOutstandingRequestCount() const
@@ -904,8 +962,10 @@ public:
cass_future_set_callback(
fut, callback, static_cast<void*>(&callbackData));
cass_future_free(fut);
}
template <class T, class S>
void
executeAsyncWrite(
@@ -918,6 +978,7 @@ public:
incremementOutstandingRequestCount();
executeAsyncHelper(statement, callback, callbackData);
}
template <class T, class S>
void
executeAsyncRead(
@@ -927,6 +988,7 @@ public:
{
executeAsyncHelper(statement, callback, callbackData);
}
void
executeSyncWrite(CassandraStatement const& statement) const
{
@@ -1003,18 +1065,32 @@ public:
}
CassandraResult
executeSyncRead(CassandraStatement const& statement) const
executeAsyncRead(
CassandraStatement const& statement,
boost::asio::yield_context& yield) const
{
using result = boost::asio::async_result<
boost::asio::yield_context,
void(boost::system::error_code, CassError)>;
CassFuture* fut;
CassError rc;
do
{
fut = cass_session_execute(session_.get(), statement.get());
rc = cass_future_error_code(fut);
boost::system::error_code ec;
rc = cass_future_error_code(fut, yield[ec]);
if (ec)
{
BOOST_LOG_TRIVIAL(error)
<< "Cannot read async cass_future_error_code";
}
if (rc != CASS_OK)
{
std::stringstream ss;
ss << "Cassandra executeSyncRead error";
ss << "Cassandra executeAsyncRead error";
ss << ": " << cass_error_desc(rc);
BOOST_LOG_TRIVIAL(error) << ss.str();
}
@@ -1030,6 +1106,8 @@ public:
}
} while (rc != CASS_OK);
// The future should have returned at the earlier cass_future_error_code
// so we can use the sync version of this function.
CassResult const* res = cass_future_get_result(fut);
cass_future_free(fut);
return {res};

151
src/backend/DBHelpers.cpp
View File

@@ -1,151 +0,0 @@
#include <boost/format.hpp>
#include <backend/DBHelpers.h>
#include <memory>
static bool
writeToLedgersDB(ripple::LedgerInfo const& info, PgQuery& pgQuery)
{
BOOST_LOG_TRIVIAL(debug) << __func__;
auto cmd = boost::format(
R"(INSERT INTO ledgers
VALUES (%u,'\x%s', '\x%s',%u,%u,%u,%u,%u,'\x%s','\x%s'))");
auto ledgerInsert = boost::str(
cmd % info.seq % ripple::strHex(info.hash) %
ripple::strHex(info.parentHash) % info.drops.drops() %
info.closeTime.time_since_epoch().count() %
info.parentCloseTime.time_since_epoch().count() %
info.closeTimeResolution.count() % info.closeFlags %
ripple::strHex(info.accountHash) % ripple::strHex(info.txHash));
BOOST_LOG_TRIVIAL(trace) << __func__ << " : "
<< " : "
<< "query string = " << ledgerInsert;
auto res = pgQuery(ledgerInsert.data());
return res;
}
/*
bool
writeBooks(std::vector<BookDirectoryData> const& bookDirData, PgQuery& pg)
{
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " : "
<< "Writing " << bookDirData.size() << "books to Postgres";
try
{
std::stringstream booksCopyBuffer;
for (auto const& data : bookDirData)
{
std::string directoryIndex = ripple::strHex(data.directoryIndex);
std::string bookIndex = ripple::strHex(data.bookIndex);
auto ledgerSeq = data.ledgerSequence;
booksCopyBuffer << "\\\\x" << directoryIndex << '\t'
<< std::to_string(ledgerSeq) << '\t' << "\\\\x"
<< bookIndex << '\n';
}
pg.bulkInsert("books", booksCopyBuffer.str());
BOOST_LOG_TRIVIAL(info) << __func__ << " : "
<< "Successfully inserted books";
return true;
}
catch (std::exception& e)
{
BOOST_LOG_TRIVIAL(error)
<< __func__ << "Caught exception inserting books : " << e.what();
assert(false);
return false;
}
}
*/
/*
bool
writeToPostgres(
ripple::LedgerInfo const& info,
std::vector<AccountTransactionsData> const& accountTxData,
std::shared_ptr<PgPool> const& pgPool)
{
BOOST_LOG_TRIVIAL(debug) << __func__ << " : "
<< "Beginning write to Postgres";
try
{
// Create a PgQuery object to run multiple commands over the
// same connection in a single transaction block.
PgQuery pg(pgPool);
auto res = pg("BEGIN");
if (!res || res.status() != PGRES_COMMAND_OK)
{
std::stringstream msg;
msg << "bulkWriteToTable : Postgres insert error: " << res.msg();
throw std::runtime_error(msg.str());
}
// Writing to the ledgers db fails if the ledger already
// exists in the db. In this situation, the ETL process has
// detected there is another writer, and falls back to only
// publishing
if (!writeToLedgersDB(info, pg))
{
BOOST_LOG_TRIVIAL(warning)
<< __func__ << " : "
<< "Failed to write to ledgers database.";
return false;
}
std::stringstream transactionsCopyBuffer;
std::stringstream accountTransactionsCopyBuffer;
for (auto const& data : accountTxData)
{
std::string txHash = ripple::strHex(data.txHash);
std::string nodestoreHash = ripple::strHex(data.nodestoreHash);
auto idx = data.transactionIndex;
auto ledgerSeq = data.ledgerSequence;
transactionsCopyBuffer << std::to_string(ledgerSeq) << '\t'
<< std::to_string(idx) << '\t' << "\\\\x"
<< txHash << '\t' << "\\\\x" << nodestoreHash
<< '\n';
for (auto const& a : data.accounts)
{
std::string acct = ripple::strHex(a);
accountTransactionsCopyBuffer
<< "\\\\x" << acct << '\t' << std::to_string(ledgerSeq)
<< '\t' << std::to_string(idx) << '\n';
}
}
pg.bulkInsert("transactions", transactionsCopyBuffer.str());
pg.bulkInsert(
"account_transactions", accountTransactionsCopyBuffer.str());
res = pg("COMMIT");
if (!res || res.status() != PGRES_COMMAND_OK)
{
std::stringstream msg;
msg << "bulkWriteToTable : Postgres insert error: " << res.msg();
assert(false);
throw std::runtime_error(msg.str());
}
BOOST_LOG_TRIVIAL(info) << __func__ << " : "
<< "Successfully wrote to Postgres";
return true;
}
catch (std::exception& e)
{
BOOST_LOG_TRIVIAL(error)
<< __func__
<< "Caught exception writing to Postgres : " << e.what();
assert(false);
return false;
}
}
*/

11
src/backend/DBHelpers.h
View File

@@ -14,8 +14,8 @@
struct AccountTransactionsData
{
boost::container::flat_set<ripple::AccountID> accounts;
uint32_t ledgerSequence;
uint32_t transactionIndex;
std::uint32_t ledgerSequence;
std::uint32_t transactionIndex;
ripple::uint256 txHash;
AccountTransactionsData(
@@ -39,6 +39,7 @@ isOffer(T const& object)
short offer_bytes = (object[1] << 8) | object[2];
return offer_bytes == 0x006f;
}
template <class T>
inline bool
isOfferHex(T const& object)
@@ -51,6 +52,7 @@ isOfferHex(T const& object)
}
return false;
}
template <class T>
inline bool
isDirNode(T const& object)
@@ -58,6 +60,7 @@ isDirNode(T const& object)
short spaceKey = (object.data()[1] << 8) | object.data()[2];
return spaceKey == 0x0064;
}
template <class T, class R>
inline bool
isBookDir(T const& key, R const& object)
@@ -69,6 +72,7 @@ isBookDir(T const& key, R const& object)
ripple::SerialIter{object.data(), object.size()}, key};
return !sle[~ripple::sfOwner].has_value();
}
template <class T>
inline ripple::uint256
getBook(T const& offer)
@@ -115,11 +119,12 @@ deserializeHeader(ripple::Slice data)
return info;
}
inline std::string
uint256ToString(ripple::uint256 const& uint)
{
return {reinterpret_cast<const char*>(uint.data()), uint.size()};
}
static constexpr uint32_t rippleEpochStart = 946684800;
static constexpr std::uint32_t rippleEpochStart = 946684800;
#endif

557
src/backend/Pg.cpp
View File

@@ -14,6 +14,7 @@
#include <boost/log/trivial.hpp>
#include <algorithm>
#include <array>
#include <backend/BackendInterface.h>
#include <backend/Pg.h>
#include <cassert>
#include <cstdlib>
@@ -61,28 +62,33 @@ PgResult::msg() const
https://www.postgresql.org/docs/10/libpq-connect.html
*/
void
Pg::connect()
Pg::connect(boost::asio::yield_context& yield)
{
std::function<PostgresPollingStatusType(PGconn*)> poller;
if (conn_)
{
// Nothing to do if we already have a good connection.
if (PQstatus(conn_.get()) == CONNECTION_OK)
return;
/* Try resetting connection. */
PQreset(conn_.get());
/* Try resetting connection, or disconnect and retry if that fails.
PQfinish() is synchronous so first try to asynchronously reset. */
if (PQresetStart(conn_.get()))
poller = PQresetPoll;
else
disconnect();
}
else // Make new connection.
if (!conn_)
{
conn_.reset(PQconnectdbParams(
conn_.reset(PQconnectStartParams(
reinterpret_cast<char const* const*>(&config_.keywordsIdx[0]),
reinterpret_cast<char const* const*>(&config_.valuesIdx[0]),
0));
if (!conn_)
throw std::runtime_error("No db connection struct");
poller = PQconnectPoll;
}
/** Results from a synchronous connection attempt can only be either
* CONNECTION_OK or CONNECTION_BAD. */
if (!conn_)
throw std::runtime_error("No db connection object");
if (PQstatus(conn_.get()) == CONNECTION_BAD)
{
std::stringstream ss;
@@ -91,59 +97,271 @@ Pg::connect()
throw std::runtime_error(ss.str());
}
// Log server session console messages.
PQsetNoticeReceiver(conn_.get(), noticeReceiver, nullptr);
try
{
socket_ = getSocket(yield);
/* Asynchronously connecting entails several messages between
* client and server. */
PostgresPollingStatusType poll = PGRES_POLLING_WRITING;
while (poll != PGRES_POLLING_OK)
{
switch (poll)
{
case PGRES_POLLING_FAILED: {
std::stringstream ss;
ss << "DB connection failed";
char* err = PQerrorMessage(conn_.get());
if (err)
ss << ":" << err;
else
ss << '.';
throw std::runtime_error(ss.str());
}
case PGRES_POLLING_READING:
socket_->async_wait(
boost::asio::ip::tcp::socket::wait_read, yield);
break;
case PGRES_POLLING_WRITING:
socket_->async_wait(
boost::asio::ip::tcp::socket::wait_write, yield);
break;
default: {
assert(false);
std::stringstream ss;
ss << "unknown DB polling status: " << poll;
throw std::runtime_error(ss.str());
}
}
poll = poller(conn_.get());
}
}
catch (std::exception const& e)
{
BOOST_LOG_TRIVIAL(error) << __func__ << " "
<< "error, polling connection"
<< "error = " << e.what();
// Sever connection upon any error.
disconnect();
std::stringstream ss;
ss << "polling connection error: " << e.what();
throw std::runtime_error(ss.str());
}
/* Enable asynchronous writes. */
if (PQsetnonblocking(conn_.get(), 1) == -1)
{
std::stringstream ss;
char* err = PQerrorMessage(conn_.get());
if (err)
ss << "Error setting connection to non-blocking: " << err;
else
ss << "Unknown error setting connection to non-blocking";
throw std::runtime_error(ss.str());
}
if (PQstatus(conn_.get()) != CONNECTION_OK)
{
std::stringstream ss;
ss << "bad connection" << std::to_string(PQstatus(conn_.get()));
char* err = PQerrorMessage(conn_.get());
if (err)
ss << ": " << err;
else
ss << '.';
throw std::runtime_error(ss.str());
}
}
inline void
Pg::flush(boost::asio::yield_context& yield)
{
// non-blocking connection requires manually flushing write.
int flushed;
do
{
flushed = PQflush(conn_.get());
if (flushed == 1)
{
socket_->async_wait(
boost::asio::ip::tcp::socket::wait_write, yield);
}
else if (flushed == -1)
{
std::stringstream ss;
ss << "error flushing query " << PQerrorMessage(conn_.get());
throw std::runtime_error(ss.str());
}
} while (flushed);
}
inline PgResult
Pg::waitForStatus(boost::asio::yield_context& yield, ExecStatusType expected)
{
PgResult ret;
while (true)
{
if (PQisBusy(conn_.get()))
{
socket_->async_wait(boost::asio::ip::tcp::socket::wait_read, yield);
}
if (!PQconsumeInput(conn_.get()))
{
std::stringstream ss;
ss << "query consume input error: " << PQerrorMessage(conn_.get());
throw std::runtime_error(ss.str());
}
if (PQisBusy(conn_.get()))
continue;
pg_result_type res{PQgetResult(conn_.get()), [](PGresult* result) {
PQclear(result);
}};
if (!res)
break;
auto status = PQresultStatus(res.get());
ret = PgResult(std::move(res));
if (status == expected)
break;
}
return ret;
}
inline asio_socket_type
Pg::getSocket(boost::asio::yield_context& yield)
{
asio_socket_type s{
new boost::asio::ip::tcp::socket(
boost::asio::get_associated_executor(yield),
boost::asio::ip::tcp::v4(),
PQsocket(conn_.get())),
[](boost::asio::ip::tcp::socket* socket) {
socket->cancel();
socket->release();
delete socket;
}};
return std::move(s);
}
PgResult
Pg::query(char const* command, std::size_t nParams, char const* const* values)
Pg::query(
char const* command,
std::size_t const nParams,
char const* const* values,
boost::asio::yield_context& yield)
{
// The result object must be freed using the libpq API PQclear() call.
pg_result_type ret{nullptr, [](PGresult* result) { PQclear(result); }};
// Connect then submit query.
while (true)
try
{
connect(yield);
int sent;
if (nParams)
{
std::lock_guard<std::mutex> lock(mutex_);
if (stop_)
return PgResult();
// PQexecParams can process only a single command.
sent = PQsendQueryParams(
conn_.get(),
command,
nParams,
nullptr,
values,
nullptr,
nullptr,
0);
}
try
else
{
connect();
if (nParams)
{
// PQexecParams can process only a single command.
ret.reset(PQexecParams(
conn_.get(),
command,
nParams,
nullptr,
values,
nullptr,
nullptr,
0));
}
else
{
// PQexec can process multiple commands separated by
// semi-colons. Returns the response from the last
// command processed.
ret.reset(PQexec(conn_.get(), command));
}
if (!ret)
throw std::runtime_error("no result structure returned");
break;
// PQexec can process multiple commands separated by
// semi-colons. Returns the response from the last
// command processed.
sent = PQsendQuery(conn_.get(), command);
}
catch (std::exception const& e)
if (!sent)
{
// Sever connection and retry until successful.
disconnect();
BOOST_LOG_TRIVIAL(error)
<< "database error, retrying: " << e.what();
std::this_thread::sleep_for(std::chrono::seconds(1));
std::stringstream ss;
ss << "Can't send query: " << PQerrorMessage(conn_.get());
throw std::runtime_error(ss.str());
}
flush(yield);
/* Only read response if query was submitted successfully.
Only a single response is expected, but the API requires
responses to be read until nullptr is returned.
It is possible for pending reads on the connection to interfere
with the current query. For simplicity, this implementation
only flushes pending writes and assumes there are no pending reads.
To avoid this, all pending reads from each query must be consumed,
and all connections with any type of error be severed. */
while (true)
{
if (PQisBusy(conn_.get()))
socket_->async_wait(
boost::asio::ip::tcp::socket::wait_read, yield);
if (!PQconsumeInput(conn_.get()))
{
std::stringstream ss;
ss << "query consume input error: "
<< PQerrorMessage(conn_.get());
throw std::runtime_error(ss.str());
}
if (PQisBusy(conn_.get()))
continue;
pg_result_type res{PQgetResult(conn_.get()), [](PGresult* result) {
PQclear(result);
}};
if (!res)
break;
ret.reset(res.release());
// ret is never null in these cases, so need to break.
bool copyStatus = false;
switch (PQresultStatus(ret.get()))
{
case PGRES_COPY_IN:
case PGRES_COPY_OUT:
case PGRES_COPY_BOTH:
copyStatus = true;
break;
default:;
}
if (copyStatus)
break;
}
}
catch (std::exception const& e)
{
BOOST_LOG_TRIVIAL(error) << __func__ << " "
<< "error, severing connection "
<< "error = " << e.what();
// Sever connection upon any error.
disconnect();
std::stringstream ss;
ss << "query error: " << e.what();
throw std::runtime_error(ss.str());
}
if (!ret)
throw std::runtime_error("no result structure returned");
// Ensure proper query execution.
switch (PQresultStatus(ret.get()))
@@ -161,6 +379,7 @@ Pg::query(char const* command, std::size_t nParams, char const* const* values)
<< ", number of tuples: " << PQntuples(ret.get())
<< ", number of fields: " << PQnfields(ret.get());
BOOST_LOG_TRIVIAL(error) << ss.str();
PgResult retRes(ret.get(), conn_.get());
disconnect();
@@ -206,7 +425,7 @@ formatParams(pg_params const& dbParams)
}
PgResult
Pg::query(pg_params const& dbParams)
Pg::query(pg_params const& dbParams, boost::asio::yield_context& yield)
{
char const* const& command = dbParams.first;
auto const formattedParams = formatParams(dbParams);
@@ -215,18 +434,21 @@ Pg::query(pg_params const& dbParams)
formattedParams.size(),
formattedParams.size()
? reinterpret_cast<char const* const*>(&formattedParams[0])
: nullptr);
: nullptr,
yield);
}
void
Pg::bulkInsert(char const* table, std::string const& records)
Pg::bulkInsert(
char const* table,
std::string const& records,
boost::asio::yield_context& yield)
{
// https://www.postgresql.org/docs/12/libpq-copy.html#LIBPQ-COPY-SEND
assert(conn_.get());
auto copyCmd = boost::format(R"(COPY %s FROM stdin)");
auto formattedCmd = boost::str(copyCmd % table);
BOOST_LOG_TRIVIAL(debug) << __func__ << " " << formattedCmd;
auto res = query(formattedCmd.c_str());
auto res = query(formattedCmd.c_str(), yield);
if (!res || res.status() != PGRES_COPY_IN)
{
std::stringstream ss;
@@ -238,39 +460,105 @@ Pg::bulkInsert(char const* table, std::string const& records)
throw std::runtime_error(ss.str());
}
if (PQputCopyData(conn_.get(), records.c_str(), records.size()) == -1)
try
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". PQputCopyData error: " << PQerrorMessage(conn_.get());
disconnect();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
throw std::runtime_error(ss.str());
}
while (true)
{
std::int32_t const putCopy =
PQputCopyData(conn_.get(), records.c_str(), records.size());
if (PQputCopyEnd(conn_.get(), nullptr) == -1)
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". PQputCopyEnd error: " << PQerrorMessage(conn_.get());
disconnect();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
throw std::runtime_error(ss.str());
}
if (putCopy == -1)
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". PQputCopyData error: " << PQerrorMessage(conn_.get());
disconnect();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
throw std::runtime_error(ss.str());
}
// The result object must be freed using the libpq API PQclear() call.
pg_result_type copyEndResult{
nullptr, [](PGresult* result) { PQclear(result); }};
copyEndResult.reset(PQgetResult(conn_.get()));
ExecStatusType status = PQresultStatus(copyEndResult.get());
if (status != PGRES_COMMAND_OK)
else if (putCopy == 0)
// If the value is zero, wait for write-ready and try again.
socket_->async_wait(
boost::asio::ip::tcp::socket::wait_write, yield);
else
break;
}
flush(yield);
auto copyRes = waitForStatus(yield, PGRES_COPY_IN);
if (!copyRes || copyRes.status() != PGRES_COPY_IN)
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". Postgres insert error: " << copyRes.msg();
if (res)
ss << ". CopyPut status not PGRES_COPY_IN: "
<< copyRes.status();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
throw std::runtime_error(ss.str());
}
std::int32_t copyEnd;
do
{
copyEnd = PQputCopyEnd(conn_.get(), nullptr);
if (copyEnd == -1)
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". PQputCopyEnd error: " << PQerrorMessage(conn_.get());
disconnect();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
throw std::runtime_error(ss.str());
}
// If the value is zero, wait for write-ready and try again.
if (copyEnd == 0)
socket_->async_wait(
boost::asio::ip::tcp::socket::wait_write, yield);
} while (copyEnd == 0);
flush(yield);
auto endRes = waitForStatus(yield, PGRES_COMMAND_OK);
if (!endRes || endRes.status() != PGRES_COMMAND_OK)
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". Postgres insert error: " << endRes.msg();
if (res)
ss << ". CopyEnd status not PGRES_COMMAND_OK: "
<< endRes.status();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
throw std::runtime_error(ss.str());
}
pg_result_type finalRes{PQgetResult(conn_.get()), [](PGresult* result) {
PQclear(result);
}};
if (finalRes)
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". Postgres insert error: " << res.msg();
if (res)
ss << ". Query status not NULL: " << res.status();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
throw std::runtime_error(ss.str());
}
}
catch (std::exception const& e)
{
std::stringstream ss;
ss << "bulkInsert to " << table
<< ". PQputCopyEnd status not PGRES_COMMAND_OK: " << status
<< " message = " << PQerrorMessage(conn_.get());
BOOST_LOG_TRIVIAL(error) << __func__ << " "
<< "error, bulk insertion"
<< "error = " << e.what();
// Sever connection upon any error.
disconnect();
BOOST_LOG_TRIVIAL(error) << __func__ << " " << records;
std::stringstream ss;
ss << "query error: " << e.what();
throw std::runtime_error(ss.str());
}
}
@@ -306,12 +594,21 @@ Pg::clear()
}
} while (res && conn_);
try
{
socket_->cancel();
}
catch (std::exception const& e)
{
}
return conn_ != nullptr;
}
//-----------------------------------------------------------------------------
PgPool::PgPool(boost::json::object const& config)
PgPool::PgPool(boost::asio::io_context& ioc, boost::json::object const& config)
: ioc_(ioc)
{
// Make sure that boost::asio initializes the SSL library.
{
@@ -477,7 +774,7 @@ PgPool::PgPool(boost::json::object const& config)
config_.valuesIdx.push_back(nullptr);
if (config.contains("max_connections"))
config_.max_connections = config.at("max_connections").as_uint64();
config_.max_connections = config.at("max_connections").as_int64();
std::size_t timeout;
if (config.contains("timeout"))
config_.timeout =
@@ -516,32 +813,6 @@ PgPool::onStop()
BOOST_LOG_TRIVIAL(info) << "stopped";
}
void
PgPool::idleSweeper()
{
std::size_t before, after;
{
std::lock_guard<std::mutex> lock(mutex_);
before = idle_.size();
if (config_.timeout != std::chrono::seconds(0))
{
auto const found =
idle_.upper_bound(clock_type::now() - config_.timeout);
for (auto it = idle_.begin(); it != found;)
{
it = idle_.erase(it);
--connections_;
}
}
after = idle_.size();
}
BOOST_LOG_TRIVIAL(info)
<< "Idle sweeper. connections: " << connections_
<< ". checked out: " << connections_ - after
<< ". idle before, after sweep: " << before << ", " << after;
}
std::unique_ptr<Pg>
PgPool::checkout()
{
@@ -563,7 +834,7 @@ PgPool::checkout()
else if (connections_ < config_.max_connections)
{
++connections_;
ret = std::make_unique<Pg>(config_, stop_, mutex_);
ret = std::make_unique<Pg>(config_, ioc_, stop_, mutex_);
}
// Otherwise, wait until a connection becomes available or we stop.
else
@@ -585,6 +856,7 @@ PgPool::checkin(std::unique_ptr<Pg>& pg)
std::lock_guard<std::mutex> lock(mutex_);
if (!stop_ && pg->clear())
{
pg->clear();
idle_.emplace(clock_type::now(), std::move(pg));
}
else
@@ -600,11 +872,11 @@ PgPool::checkin(std::unique_ptr<Pg>& pg)
//-----------------------------------------------------------------------------
std::shared_ptr<PgPool>
make_PgPool(boost::json::object const& config)
make_PgPool(boost::asio::io_context& ioc, boost::json::object const& config)
{
try
{
auto ret = std::make_shared<PgPool>(config);
auto ret = std::make_shared<PgPool>(ioc, config);
ret->setup();
return ret;
}
@@ -612,13 +884,18 @@ make_PgPool(boost::json::object const& config)
{
boost::json::object configCopy = config;
configCopy["database"] = "postgres";
auto ret = std::make_shared<PgPool>(configCopy);
auto ret = std::make_shared<PgPool>(ioc, configCopy);
ret->setup();
PgQuery pgQuery{ret};
std::string query = "CREATE DATABASE " +
std::string{config.at("database").as_string().c_str()};
pgQuery(query.c_str());
ret = std::make_shared<PgPool>(config);
Backend::synchronous([&](boost::asio::yield_context yield) {
PgQuery pgQuery(ret);
std::string query = "CREATE DATABASE " +
std::string{config.at("database").as_string().c_str()};
pgQuery(query.c_str(), yield);
});
ret = std::make_shared<PgPool>(ioc, config);
ret->setup();
return ret;
}
@@ -767,7 +1044,7 @@ CREATE TABLE IF NOT EXISTS ledgers (
CREATE TABLE IF NOT EXISTS objects (
key bytea NOT NULL,
ledger_seq bigint NOT NULL REFERENCES ledgers ON DELETE CASCADE,
ledger_seq bigint NOT NULL,
object bytea
) PARTITION BY RANGE (ledger_seq);
@@ -1297,8 +1574,8 @@ void
applySchema(
std::shared_ptr<PgPool> const& pool,
char const* schema,
std::uint32_t currentVersion,
std::uint32_t schemaVersion)
std::uint32_t const currentVersion,
std::uint32_t const schemaVersion)
{
if (currentVersion != 0 && schemaVersion != currentVersion + 1)
{
@@ -1310,7 +1587,11 @@ applySchema(
throw std::runtime_error(ss.str());
}
auto res = PgQuery(pool)({schema, {}});
PgResult res;
Backend::synchronous([&](boost::asio::yield_context yield) {
res = PgQuery(pool)(schema, yield);
});
if (!res)
{
std::stringstream ss;
@@ -1320,7 +1601,10 @@ applySchema(
}
auto cmd = boost::format(R"(SELECT set_schema_version(%u, 0))");
res = PgQuery(pool)({boost::str(cmd % schemaVersion).c_str(), {}});
Backend::synchronous([&](boost::asio::yield_context yield) {
res = PgQuery(pool)(boost::str(cmd % schemaVersion).c_str(), yield);
});
if (!res)
{
std::stringstream ss;
@@ -1333,7 +1617,11 @@ applySchema(
void
initAccountTx(std::shared_ptr<PgPool> const& pool)
{
auto res = PgQuery(pool)(accountTxSchema);
PgResult res;
Backend::synchronous([&](boost::asio::yield_context yield) {
res = PgQuery(pool)(accountTxSchema, yield);
});
if (!res)
{
std::stringstream ss;
@@ -1346,7 +1634,11 @@ void
initSchema(std::shared_ptr<PgPool> const& pool)
{
// Figure out what schema version, if any, is already installed.
auto res = PgQuery(pool)({version_query, {}});
PgResult res;
Backend::synchronous([&](boost::asio::yield_context yield) {
res = PgQuery(pool)(version_query, yield);
});
if (!res)
{
std::stringstream ss;
@@ -1370,7 +1662,10 @@ initSchema(std::shared_ptr<PgPool> const& pool)
// This protects against corruption in an aborted install that is
// followed by a fresh installation attempt with a new schema.
auto cmd = boost::format(R"(SELECT set_schema_version(0, %u))");
res = PgQuery(pool)({boost::str(cmd % freshVersion).c_str(), {}});
Backend::synchronous([&](boost::asio::yield_context yield) {
res = PgQuery(pool)(boost::str(cmd % freshVersion).c_str(), yield);
});
if (!res)
{
std::stringstream ss;
@@ -1405,7 +1700,9 @@ initSchema(std::shared_ptr<PgPool> const& pool)
// @return LedgerInfo
std::optional<ripple::LedgerInfo>
getLedger(
std::variant<std::monostate, ripple::uint256, uint32_t> const& whichLedger,
boost::asio::yield_context yield,
std::variant<std::monostate, ripple::uint256, std::uint32_t> const&
whichLedger,
std::shared_ptr<PgPool>& pgPool)
{
ripple::LedgerInfo lgrInfo;
@@ -1414,9 +1711,9 @@ getLedger(
"total_coins, closing_time, prev_closing_time, close_time_res, "
"close_flags, ledger_seq FROM ledgers ";
uint32_t expNumResults = 1;
std::uint32_t expNumResults = 1;
if (auto ledgerSeq = std::get_if<uint32_t>(&whichLedger))
if (auto ledgerSeq = std::get_if<std::uint32_t>(&whichLedger))
{
sql << "WHERE ledger_seq = " + std::to_string(*ledgerSeq);
}
@@ -1432,7 +1729,7 @@ getLedger(
BOOST_LOG_TRIVIAL(trace) << __func__ << " : sql = " << sql.str();
auto res = PgQuery(pgPool)(sql.str().data());
auto res = PgQuery(pgPool)(sql.str().data(), yield);
if (!res)
{
BOOST_LOG_TRIVIAL(error)

80
src/backend/Pg.h
View File

@@ -4,6 +4,9 @@
#include <ripple/basics/StringUtilities.h>
#include <ripple/basics/chrono.h>
#include <ripple/ledger/ReadView.h>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/icl/closed_interval.hpp>
#include <boost/json.hpp>
#include <boost/lexical_cast.hpp>
@@ -24,6 +27,9 @@
// These postgres structs must be freed only by the postgres API.
using pg_result_type = std::unique_ptr<PGresult, void (*)(PGresult*)>;
using pg_connection_type = std::unique_ptr<PGconn, void (*)(PGconn*)>;
using asio_socket_type = std::unique_ptr<
boost::asio::ip::tcp::socket,
void (*)(boost::asio::ip::tcp::socket*)>;
/** first: command
* second: parameter values
@@ -46,7 +52,7 @@ using pg_formatted_params = std::vector<char const*>;
struct PgConfig
{
/** Maximum connections allowed to db. */
std::size_t max_connections{std::numeric_limits<std::size_t>::max()};
std::size_t max_connections{1000};
/** Close idle connections past this duration. */
std::chrono::seconds timeout{600};
@@ -255,12 +261,24 @@ class Pg
friend class PgQuery;
PgConfig const& config_;
boost::asio::io_context::strand strand_;
bool& stop_;
std::mutex& mutex_;
asio_socket_type socket_{nullptr, [](boost::asio::ip::tcp::socket*) {}};
// The connection object must be freed using the libpq API PQfinish() call.
pg_connection_type conn_{nullptr, [](PGconn* conn) { PQfinish(conn); }};
inline asio_socket_type
getSocket(boost::asio::yield_context& strand);
inline PgResult
waitForStatus(boost::asio::yield_context& yield, ExecStatusType expected);
inline void
flush(boost::asio::yield_context& yield);
/** Clear results from the connection.
*
* Results from previous commands must be cleared before new commands
@@ -280,13 +298,14 @@ class Pg
* or in an errored state, reconnects to the database.
*/
void
connect();
connect(boost::asio::yield_context& yield);
/** Disconnect from postgres. */
void
disconnect()
{
conn_.reset();
socket_.reset();
}
/** Execute postgres query.
@@ -302,7 +321,11 @@ class Pg
* @return Query result object.
*/
PgResult
query(char const* command, std::size_t nParams, char const* const* values);
query(
char const* command,
std::size_t const nParams,
char const* const* values,
boost::asio::yield_context& yield);
/** Execute postgres query with no parameters.
*
@@ -310,9 +333,9 @@ class Pg
* @return Query result object;
*/
PgResult
query(char const* command)
query(char const* command, boost::asio::yield_context& yield)
{
return query(command, 0, nullptr);
return query(command, 0, nullptr, yield);
}
/** Execute postgres query with parameters.
@@ -321,7 +344,7 @@ class Pg
* @return Query result object.
*/
PgResult
query(pg_params const& dbParams);
query(pg_params const& dbParams, boost::asio::yield_context& yield);
/** Insert multiple records into a table using Postgres' bulk COPY.
*
@@ -331,7 +354,10 @@ class Pg
* @param records Records in the COPY IN format.
*/
void
bulkInsert(char const* table, std::string const& records);
bulkInsert(
char const* table,
std::string const& records,
boost::asio::yield_context& yield);
public:
/** Constructor for Pg class.
@@ -341,8 +367,11 @@ public:
* @param stop Reference to connection pool's stop flag.
* @param mutex Reference to connection pool's mutex.
*/
Pg(PgConfig const& config, bool& stop, std::mutex& mutex)
: config_(config), stop_(stop), mutex_(mutex)
Pg(PgConfig const& config,
boost::asio::io_context& ctx,
bool& stop,
std::mutex& mutex)
: config_(config), strand_(ctx), stop_(stop), mutex_(mutex)
{
}
};
@@ -368,6 +397,7 @@ class PgPool
using clock_type = std::chrono::steady_clock;
boost::asio::io_context& ioc_;
PgConfig config_;
std::mutex mutex_;
std::condition_variable cond_;
@@ -406,13 +436,19 @@ public:
* @param j Logger object.
* @param parent Stoppable parent.
*/
PgPool(boost::json::object const& config);
PgPool(boost::asio::io_context& ioc, boost::json::object const& config);
~PgPool()
{
onStop();
}
PgConfig&
config()
{
return config_;
}
/** Initiate idle connection timer.
*
* The PgPool object needs to be fully constructed to support asynchronous
@@ -424,10 +460,6 @@ public:
/** Prepare for process shutdown. (Stoppable) */
void
onStop();
/** Disconnect idle postgres connections. */
void
idleSweeper();
};
//-----------------------------------------------------------------------------
@@ -467,11 +499,11 @@ public:
* @return Result of query, including errors.
*/
PgResult
operator()(pg_params const& dbParams)
operator()(pg_params const& dbParams, boost::asio::yield_context& yield)
{
if (!pg_) // It means we're stopping. Return empty result.
return PgResult();
return pg_->query(dbParams);
return pg_->query(dbParams, yield);
}
/** Execute postgres query with only command statement.
@@ -480,9 +512,9 @@ public:
* @return Result of query, including errors.
*/
PgResult
operator()(char const* command)
operator()(char const* command, boost::asio::yield_context& yield)
{
return operator()(pg_params{command, {}});
return operator()(pg_params{command, {}}, yield);
}
/** Insert multiple records into a table using Postgres' bulk COPY.
@@ -493,9 +525,12 @@ public:
* @param records Records in the COPY IN format.
*/
void
bulkInsert(char const* table, std::string const& records)
bulkInsert(
char const* table,
std::string const& records,
boost::asio::yield_context& yield)
{
pg_->bulkInsert(table, records);
pg_->bulkInsert(table, records, yield);
}
};
@@ -509,7 +544,7 @@ public:
* @return Postgres connection pool manager
*/
std::shared_ptr<PgPool>
make_PgPool(boost::json::object const& pgConfig);
make_PgPool(boost::asio::io_context& ioc, boost::json::object const& pgConfig);
/** Initialize the Postgres schema.
*
@@ -529,7 +564,8 @@ initAccountTx(std::shared_ptr<PgPool> const& pool);
// @return vector of LedgerInfos
std::optional<ripple::LedgerInfo>
getLedger(
std::variant<std::monostate, ripple::uint256, uint32_t> const& whichLedger,
std::variant<std::monostate, ripple::uint256, std::uint32_t> const&
whichLedger,
std::shared_ptr<PgPool>& pgPool);
#endif // RIPPLE_CORE_PG_H_INCLUDED

592
src/backend/PostgresBackend.cpp
View File

@@ -4,9 +4,26 @@
#include <thread>
namespace Backend {
PostgresBackend::PostgresBackend(boost::json::object const& config)
// Type alias for async completion handlers
using completion_token = boost::asio::yield_context;
using function_type = void(boost::system::error_code);
using result_type = boost::asio::async_result<completion_token, function_type>;
using handler_type = typename result_type::completion_handler_type;
struct HandlerWrapper
{
handler_type handler;
HandlerWrapper(handler_type&& handler_) : handler(std::move(handler_))
{
}
};
PostgresBackend::PostgresBackend(
boost::asio::io_context& ioc,
boost::json::object const& config)
: BackendInterface(config)
, pgPool_(make_PgPool(config))
, pgPool_(make_PgPool(ioc, config))
, writeConnection_(pgPool_)
{
if (config.contains("write_interval"))
@@ -19,22 +36,24 @@ PostgresBackend::writeLedger(
ripple::LedgerInfo const& ledgerInfo,
std::string&& ledgerHeader)
{
auto cmd = boost::format(
R"(INSERT INTO ledgers
VALUES (%u,'\x%s', '\x%s',%u,%u,%u,%u,%u,'\x%s','\x%s'))");
synchronous([&](boost::asio::yield_context yield) {
auto cmd = boost::format(
R"(INSERT INTO ledgers
VALUES (%u,'\x%s', '\x%s',%u,%u,%u,%u,%u,'\x%s','\x%s'))");
auto ledgerInsert = boost::str(
cmd % ledgerInfo.seq % ripple::strHex(ledgerInfo.hash) %
ripple::strHex(ledgerInfo.parentHash) % ledgerInfo.drops.drops() %
ledgerInfo.closeTime.time_since_epoch().count() %
ledgerInfo.parentCloseTime.time_since_epoch().count() %
ledgerInfo.closeTimeResolution.count() % ledgerInfo.closeFlags %
ripple::strHex(ledgerInfo.accountHash) %
ripple::strHex(ledgerInfo.txHash));
auto ledgerInsert = boost::str(
cmd % ledgerInfo.seq % ripple::strHex(ledgerInfo.hash) %
ripple::strHex(ledgerInfo.parentHash) % ledgerInfo.drops.drops() %
ledgerInfo.closeTime.time_since_epoch().count() %
ledgerInfo.parentCloseTime.time_since_epoch().count() %
ledgerInfo.closeTimeResolution.count() % ledgerInfo.closeFlags %
ripple::strHex(ledgerInfo.accountHash) %
ripple::strHex(ledgerInfo.txHash));
auto res = writeConnection_(ledgerInsert.data());
abortWrite_ = !res;
inProcessLedger = ledgerInfo.seq;
auto res = writeConnection_(ledgerInsert.data(), yield);
abortWrite_ = !res;
inProcessLedger = ledgerInfo.seq;
});
}
void
@@ -57,65 +76,71 @@ PostgresBackend::writeAccountTransactions(
}
}
}
void
PostgresBackend::doWriteLedgerObject(
std::string&& key,
uint32_t seq,
std::uint32_t const seq,
std::string&& blob)
{
if (abortWrite_)
return;
objectsBuffer_ << "\\\\x" << ripple::strHex(key) << '\t'
<< std::to_string(seq) << '\t' << "\\\\x"
<< ripple::strHex(blob) << '\n';
numRowsInObjectsBuffer_++;
// If the buffer gets too large, the insert fails. Not sure why. So we
// insert after 1 million records
if (numRowsInObjectsBuffer_ % writeInterval_ == 0)
{
BOOST_LOG_TRIVIAL(info)
<< __func__ << " Flushing large buffer. num objects = "
<< numRowsInObjectsBuffer_;
writeConnection_.bulkInsert("objects", objectsBuffer_.str());
BOOST_LOG_TRIVIAL(info) << __func__ << " Flushed large buffer";
objectsBuffer_.str("");
}
synchronous([&](boost::asio::yield_context yield) {
if (abortWrite_)
return;
objectsBuffer_ << "\\\\x" << ripple::strHex(key) << '\t'
<< std::to_string(seq) << '\t' << "\\\\x"
<< ripple::strHex(blob) << '\n';
numRowsInObjectsBuffer_++;
// If the buffer gets too large, the insert fails. Not sure why. So we
// insert after 1 million records
if (numRowsInObjectsBuffer_ % writeInterval_ == 0)
{
BOOST_LOG_TRIVIAL(info)
<< __func__ << " Flushing large buffer. num objects = "
<< numRowsInObjectsBuffer_;
writeConnection_.bulkInsert("objects", objectsBuffer_.str(), yield);
BOOST_LOG_TRIVIAL(info) << __func__ << " Flushed large buffer";
objectsBuffer_.str("");
}
});
}
void
PostgresBackend::writeSuccessor(
std::string&& key,
uint32_t seq,
std::uint32_t const seq,
std::string&& successor)
{
if (range)
{
if (successors_.count(key) > 0)
return;
successors_.insert(key);
}
successorBuffer_ << "\\\\x" << ripple::strHex(key) << '\t'
<< std::to_string(seq) << '\t' << "\\\\x"
<< ripple::strHex(successor) << '\n';
BOOST_LOG_TRIVIAL(trace)
<< __func__ << ripple::strHex(key) << " - " << std::to_string(seq);
numRowsInSuccessorBuffer_++;
if (numRowsInSuccessorBuffer_ % writeInterval_ == 0)
{
BOOST_LOG_TRIVIAL(info)
<< __func__ << " Flushing large buffer. num successors = "
<< numRowsInSuccessorBuffer_;
writeConnection_.bulkInsert("successor", successorBuffer_.str());
BOOST_LOG_TRIVIAL(info) << __func__ << " Flushed large buffer";
successorBuffer_.str("");
}
synchronous([&](boost::asio::yield_context yield) {
if (range)
{
if (successors_.count(key) > 0)
return;
successors_.insert(key);
}
successorBuffer_ << "\\\\x" << ripple::strHex(key) << '\t'
<< std::to_string(seq) << '\t' << "\\\\x"
<< ripple::strHex(successor) << '\n';
BOOST_LOG_TRIVIAL(trace)
<< __func__ << ripple::strHex(key) << " - " << std::to_string(seq);
numRowsInSuccessorBuffer_++;
if (numRowsInSuccessorBuffer_ % writeInterval_ == 0)
{
BOOST_LOG_TRIVIAL(info)
<< __func__ << " Flushing large buffer. num successors = "
<< numRowsInSuccessorBuffer_;
writeConnection_.bulkInsert(
"successor", successorBuffer_.str(), yield);
BOOST_LOG_TRIVIAL(info) << __func__ << " Flushed large buffer";
successorBuffer_.str("");
}
});
}
void
PostgresBackend::writeTransaction(
std::string&& hash,
uint32_t seq,
uint32_t date,
std::uint32_t const seq,
std::uint32_t const date,
std::string&& transaction,
std::string&& metadata)
{
@@ -127,8 +152,8 @@ PostgresBackend::writeTransaction(
<< '\t' << "\\\\x" << ripple::strHex(metadata) << '\n';
}
uint32_t
checkResult(PgResult const& res, uint32_t numFieldsExpected)
std::uint32_t
checkResult(PgResult const& res, std::uint32_t const numFieldsExpected)
{
if (!res)
{
@@ -201,50 +226,62 @@ parseLedgerInfo(PgResult const& res)
info.validated = true;
return info;
}
std::optional<uint32_t>
PostgresBackend::fetchLatestLedgerSequence() const
std::optional<std::uint32_t>
PostgresBackend::fetchLatestLedgerSequence(
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
auto res = pgQuery(
"SELECT ledger_seq FROM ledgers ORDER BY ledger_seq DESC LIMIT 1");
if (checkResult(res, 1))
pgQuery(set_timeout, yield);
auto const query =
"SELECT ledger_seq FROM ledgers ORDER BY ledger_seq DESC LIMIT 1";
if (auto res = pgQuery(query, yield); checkResult(res, 1))
return res.asBigInt(0, 0);
return {};
}
std::optional<ripple::LedgerInfo>
PostgresBackend::fetchLedgerBySequence(uint32_t sequence) const
PostgresBackend::fetchLedgerBySequence(
std::uint32_t const sequence,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT * FROM ledgers WHERE ledger_seq = "
<< std::to_string(sequence);
auto res = pgQuery(sql.str().data());
if (checkResult(res, 10))
if (auto res = pgQuery(sql.str().data(), yield); checkResult(res, 10))
return parseLedgerInfo(res);
return {};
}
std::optional<ripple::LedgerInfo>
PostgresBackend::fetchLedgerByHash(ripple::uint256 const& hash) const
PostgresBackend::fetchLedgerByHash(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT * FROM ledgers WHERE ledger_hash = "
<< ripple::to_string(hash);
auto res = pgQuery(sql.str().data());
if (checkResult(res, 10))
sql << "SELECT * FROM ledgers WHERE ledger_hash = \'\\x"
<< ripple::to_string(hash) << "\'";
if (auto res = pgQuery(sql.str().data(), yield); checkResult(res, 10))
return parseLedgerInfo(res);
return {};
}
std::optional<LedgerRange>
PostgresBackend::hardFetchLedgerRange() const
PostgresBackend::hardFetchLedgerRange(boost::asio::yield_context& yield) const
{
auto range = PgQuery(pgPool_)("SELECT complete_ledgers()");
auto range = PgQuery(pgPool_)("SELECT complete_ledgers()", yield);
if (!range)
return {};
@@ -279,17 +316,19 @@ PostgresBackend::hardFetchLedgerRange() const
std::optional<Blob>
PostgresBackend::doFetchLedgerObject(
ripple::uint256 const& key,
uint32_t sequence) const
std::uint32_t const sequence,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT object FROM objects WHERE key = "
<< "\'\\x" << ripple::strHex(key) << "\'"
<< " AND ledger_seq <= " << std::to_string(sequence)
<< " ORDER BY ledger_seq DESC LIMIT 1";
auto res = pgQuery(sql.str().data());
if (checkResult(res, 1))
if (auto res = pgQuery(sql.str().data(), yield); checkResult(res, 1))
{
auto blob = res.asUnHexedBlob(0, 0);
if (blob.size())
@@ -301,16 +340,19 @@ PostgresBackend::doFetchLedgerObject(
// returns a transaction, metadata pair
std::optional<TransactionAndMetadata>
PostgresBackend::fetchTransaction(ripple::uint256 const& hash) const
PostgresBackend::fetchTransaction(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT transaction,metadata,ledger_seq,date FROM transactions "
"WHERE hash = "
<< "\'\\x" << ripple::strHex(hash) << "\'";
auto res = pgQuery(sql.str().data());
if (checkResult(res, 4))
if (auto res = pgQuery(sql.str().data(), yield); checkResult(res, 4))
{
return {
{res.asUnHexedBlob(0, 0),
@@ -322,15 +364,19 @@ PostgresBackend::fetchTransaction(ripple::uint256 const& hash) const
return {};
}
std::vector<TransactionAndMetadata>
PostgresBackend::fetchAllTransactionsInLedger(uint32_t ledgerSequence) const
PostgresBackend::fetchAllTransactionsInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT transaction, metadata, ledger_seq,date FROM transactions "
"WHERE "
<< "ledger_seq = " << std::to_string(ledgerSequence);
auto res = pgQuery(sql.str().data());
auto res = pgQuery(sql.str().data(), yield);
if (size_t numRows = checkResult(res, 4))
{
std::vector<TransactionAndMetadata> txns;
@@ -348,14 +394,17 @@ PostgresBackend::fetchAllTransactionsInLedger(uint32_t ledgerSequence) const
}
std::vector<ripple::uint256>
PostgresBackend::fetchAllTransactionHashesInLedger(
uint32_t ledgerSequence) const
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT hash FROM transactions WHERE "
<< "ledger_seq = " << std::to_string(ledgerSequence);
auto res = pgQuery(sql.str().data());
auto res = pgQuery(sql.str().data(), yield);
if (size_t numRows = checkResult(res, 1))
{
std::vector<ripple::uint256> hashes;
@@ -365,22 +414,26 @@ PostgresBackend::fetchAllTransactionHashesInLedger(
}
return hashes;
}
return {};
}
std::optional<ripple::uint256>
PostgresBackend::doFetchSuccessorKey(
ripple::uint256 key,
uint32_t ledgerSequence) const
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT next FROM successor WHERE key = "
<< "\'\\x" << ripple::strHex(key) << "\'"
<< " AND ledger_seq <= " << std::to_string(ledgerSequence)
<< " ORDER BY ledger_seq DESC LIMIT 1";
auto res = pgQuery(sql.str().data());
if (checkResult(res, 1))
if (auto res = pgQuery(sql.str().data(), yield); checkResult(res, 1))
{
auto next = res.asUInt256(0, 0);
if (next == lastKey)
@@ -393,114 +446,107 @@ PostgresBackend::doFetchSuccessorKey(
std::vector<TransactionAndMetadata>
PostgresBackend::fetchTransactions(
std::vector<ripple::uint256> const& hashes) const
std::vector<ripple::uint256> const& hashes,
boost::asio::yield_context& yield) const
{
std::vector<TransactionAndMetadata> results;
constexpr bool doAsync = true;
if (doAsync)
{
auto start = std::chrono::system_clock::now();
auto end = std::chrono::system_clock::now();
auto duration = ((end - start).count()) / 1000000000.0;
results.resize(hashes.size());
std::condition_variable cv;
std::mutex mtx;
std::atomic_uint numRemaining = hashes.size();
for (size_t i = 0; i < hashes.size(); ++i)
{
auto const& hash = hashes[i];
boost::asio::post(
pool_, [this, &hash, &results, &numRemaining, &cv, &mtx, i]() {
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " getting txn = " << i;
PgQuery pgQuery(pgPool_);
std::stringstream sql;
sql << "SELECT transaction,metadata,ledger_seq,date FROM "
"transactions "
"WHERE HASH = \'\\x"
<< ripple::strHex(hash) << "\'";
if (!hashes.size())
return {};
auto res = pgQuery(sql.str().data());
if (size_t numRows = checkResult(res, 4))
{
results[i] = {
res.asUnHexedBlob(0, 0),
res.asUnHexedBlob(0, 1),
res.asBigInt(0, 2),
res.asBigInt(0, 3)};
}
if (--numRemaining == 0)
{
std::unique_lock lck(mtx);
cv.notify_one();
}
});
}
std::unique_lock lck(mtx);
cv.wait(lck, [&numRemaining]() { return numRemaining == 0; });
auto end2 = std::chrono::system_clock::now();
duration = ((end2 - end).count()) / 1000000000.0;
BOOST_LOG_TRIVIAL(info)
<< __func__ << " fetched " << std::to_string(hashes.size())
<< " transactions with threadpool. took "
<< std::to_string(duration);
}
else
std::vector<TransactionAndMetadata> results;
results.resize(hashes.size());
handler_type handler(std::forward<decltype(yield)>(yield));
result_type result(handler);
auto hw = new HandlerWrapper(std::move(handler));
auto start = std::chrono::system_clock::now();
auto end = std::chrono::system_clock::now();
auto duration = ((end - start).count()) / 1000000000.0;
std::atomic_uint numRemaining = hashes.size();
for (size_t i = 0; i < hashes.size(); ++i)
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
std::stringstream sql;
for (size_t i = 0; i < hashes.size(); ++i)
{
auto const& hash = hashes[i];
sql << "SELECT transaction,metadata,ledger_seq,date FROM "
"transactions "
"WHERE HASH = \'\\x"
<< ripple::strHex(hash) << "\'";
if (i + 1 < hashes.size())
sql << " UNION ALL ";
}
auto start = std::chrono::system_clock::now();
auto res = pgQuery(sql.str().data());
auto end = std::chrono::system_clock::now();
auto duration = ((end - start).count()) / 1000000000.0;
BOOST_LOG_TRIVIAL(info)
<< __func__ << " fetched " << std::to_string(hashes.size())
<< " transactions with union all. took "
<< std::to_string(duration);
if (size_t numRows = checkResult(res, 3))
{
for (size_t i = 0; i < numRows; ++i)
results.push_back(
{res.asUnHexedBlob(i, 0),
res.asUnHexedBlob(i, 1),
res.asBigInt(i, 2),
res.asBigInt(i, 3)});
}
auto const& hash = hashes[i];
boost::asio::spawn(
get_associated_executor(yield),
[this, &hash, &results, hw, &numRemaining, i](
boost::asio::yield_context yield) {
BOOST_LOG_TRIVIAL(trace) << __func__ << " getting txn = " << i;
PgQuery pgQuery(pgPool_);
std::stringstream sql;
sql << "SELECT transaction,metadata,ledger_seq,date FROM "
"transactions "
"WHERE HASH = \'\\x"
<< ripple::strHex(hash) << "\'";
auto res = pgQuery(sql.str().data(), yield);
if (size_t numRows = checkResult(res, 4))
{
results[i] = {
res.asUnHexedBlob(0, 0),
res.asUnHexedBlob(0, 1),
res.asBigInt(0, 2),
res.asBigInt(0, 3)};
}
if (--numRemaining == 0)
{
handler_type h(std::move(hw->handler));
h(boost::system::error_code{});
}
});
}
// Yields the worker to the io_context until handler is called.
result.get();
delete hw;
auto end2 = std::chrono::system_clock::now();
duration = ((end2 - end).count()) / 1000000000.0;
BOOST_LOG_TRIVIAL(info)
<< __func__ << " fetched " << std::to_string(hashes.size())
<< " transactions with threadpool. took " << std::to_string(duration);
return results;
}
std::vector<Blob>
PostgresBackend::doFetchLedgerObjects(
std::vector<ripple::uint256> const& keys,
uint32_t sequence) const
std::uint32_t const sequence,
boost::asio::yield_context& yield) const
{
if (!keys.size())
return {};
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::vector<Blob> results;
results.resize(keys.size());
std::condition_variable cv;
std::mutex mtx;
handler_type handler(std::forward<decltype(yield)>(yield));
result_type result(handler);
auto hw = new HandlerWrapper(std::move(handler));
std::atomic_uint numRemaining = keys.size();
auto start = std::chrono::system_clock::now();
for (size_t i = 0; i < keys.size(); ++i)
{
auto const& key = keys[i];
boost::asio::post(
pool_,
[this, &key, &results, &numRemaining, &cv, &mtx, i, sequence]() {
boost::asio::spawn(
boost::asio::get_associated_executor(yield),
[this, &key, &results, &numRemaining, hw, i, sequence](
boost::asio::yield_context yield) {
PgQuery pgQuery(pgPool_);
std::stringstream sql;
sql << "SELECT object FROM "
"objects "
@@ -509,37 +555,48 @@ PostgresBackend::doFetchLedgerObjects(
<< " AND ledger_seq <= " << std::to_string(sequence)
<< " ORDER BY ledger_seq DESC LIMIT 1";
auto res = pgQuery(sql.str().data());
auto res = pgQuery(sql.str().data(), yield);
if (size_t numRows = checkResult(res, 1))
{
results[i] = res.asUnHexedBlob();
}
if (--numRemaining == 0)
{
std::unique_lock lck(mtx);
cv.notify_one();
handler_type h(std::move(hw->handler));
h(boost::system::error_code{});
}
});
}
std::unique_lock lck(mtx);
cv.wait(lck, [&numRemaining]() { return numRemaining == 0; });
// Yields the worker to the io_context until handler is called.
result.get();
delete hw;
auto end = std::chrono::system_clock::now();
auto duration = ((end - start).count()) / 1000000000.0;
BOOST_LOG_TRIVIAL(info)
<< __func__ << " fetched " << std::to_string(keys.size())
<< " objects with threadpool. took " << std::to_string(duration);
return results;
}
std::vector<LedgerObject>
PostgresBackend::fetchLedgerDiff(uint32_t ledgerSequence) const
PostgresBackend::fetchLedgerDiff(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
std::stringstream sql;
sql << "SELECT key,object FROM objects "
"WHERE "
<< "ledger_seq = " << std::to_string(ledgerSequence);
auto res = pgQuery(sql.str().data());
auto res = pgQuery(sql.str().data(), yield);
if (size_t numRows = checkResult(res, 2))
{
std::vector<LedgerObject> objects;
@@ -549,18 +606,20 @@ PostgresBackend::fetchLedgerDiff(uint32_t ledgerSequence) const
}
return objects;
}
return {};
}
AccountTransactions
PostgresBackend::fetchAccountTransactions(
ripple::AccountID const& account,
std::uint32_t limit,
std::uint32_t const limit,
bool forward,
std::optional<AccountTransactionsCursor> const& cursor) const
std::optional<AccountTransactionsCursor> const& cursor,
boost::asio::yield_context& yield) const
{
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
pgQuery(set_timeout, yield);
pg_params dbParams;
char const*& command = dbParams.first;
@@ -587,7 +646,7 @@ PostgresBackend::fetchAccountTransactions(
}
auto start = std::chrono::system_clock::now();
auto res = pgQuery(dbParams);
auto res = pgQuery(dbParams, yield);
auto end = std::chrono::system_clock::now();
auto duration = ((end - start).count()) / 1000000000.0;
@@ -595,6 +654,7 @@ PostgresBackend::fetchAccountTransactions(
<< __func__ << " : executed stored_procedure in "
<< std::to_string(duration)
<< " num records = " << std::to_string(checkResult(res, 1));
checkResult(res, 1);
char const* resultStr = res.c_str();
@@ -618,13 +678,13 @@ PostgresBackend::fetchAccountTransactions(
if (responseObj.contains("cursor"))
{
return {
fetchTransactions(hashes),
fetchTransactions(hashes, yield),
{{responseObj.at("cursor").at("ledger_sequence").as_int64(),
responseObj.at("cursor")
.at("transaction_index")
.as_int64()}}};
}
return {fetchTransactions(hashes), {}};
return {fetchTransactions(hashes, yield), {}};
}
return {{}, {}};
} // namespace Backend
@@ -643,85 +703,93 @@ PostgresBackend::close()
}
void
PostgresBackend::startWrites()
PostgresBackend::startWrites() const
{
numRowsInObjectsBuffer_ = 0;
abortWrite_ = false;
auto res = writeConnection_("BEGIN");
if (!res || res.status() != PGRES_COMMAND_OK)
{
std::stringstream msg;
msg << "Postgres error creating transaction: " << res.msg();
throw std::runtime_error(msg.str());
}
synchronous([&](boost::asio::yield_context yield) {
numRowsInObjectsBuffer_ = 0;
abortWrite_ = false;
auto res = writeConnection_("BEGIN", yield);
if (!res || res.status() != PGRES_COMMAND_OK)
{
std::stringstream msg;
msg << "Postgres error creating transaction: " << res.msg();
throw std::runtime_error(msg.str());
}
});
}
bool
PostgresBackend::doFinishWrites()
PostgresBackend::doFinishWrites() const
{
if (!abortWrite_)
{
std::string txStr = transactionsBuffer_.str();
writeConnection_.bulkInsert("transactions", txStr);
writeConnection_.bulkInsert(
"account_transactions", accountTxBuffer_.str());
std::string objectsStr = objectsBuffer_.str();
if (objectsStr.size())
writeConnection_.bulkInsert("objects", objectsStr);
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " objects size = " << objectsStr.size()
<< " txns size = " << txStr.size();
std::string successorStr = successorBuffer_.str();
if (successorStr.size())
writeConnection_.bulkInsert("successor", successorStr);
successors_.clear();
if (!range)
synchronous([&](boost::asio::yield_context yield) {
if (!abortWrite_)
{
std::stringstream indexCreate;
indexCreate
<< "CREATE INDEX diff ON objects USING hash(ledger_seq) "
"WHERE NOT "
"ledger_seq = "
<< std::to_string(inProcessLedger);
writeConnection_(indexCreate.str().data());
std::string txStr = transactionsBuffer_.str();
writeConnection_.bulkInsert("transactions", txStr, yield);
writeConnection_.bulkInsert(
"account_transactions", accountTxBuffer_.str(), yield);
std::string objectsStr = objectsBuffer_.str();
if (objectsStr.size())
writeConnection_.bulkInsert("objects", objectsStr, yield);
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " objects size = " << objectsStr.size()
<< " txns size = " << txStr.size();
std::string successorStr = successorBuffer_.str();
if (successorStr.size())
writeConnection_.bulkInsert("successor", successorStr, yield);
if (!range)
{
std::stringstream indexCreate;
indexCreate
<< "CREATE INDEX diff ON objects USING hash(ledger_seq) "
"WHERE NOT "
"ledger_seq = "
<< std::to_string(inProcessLedger);
writeConnection_(indexCreate.str().data(), yield);
}
}
}
auto res = writeConnection_("COMMIT");
if (!res || res.status() != PGRES_COMMAND_OK)
{
std::stringstream msg;
msg << "Postgres error committing transaction: " << res.msg();
throw std::runtime_error(msg.str());
}
transactionsBuffer_.str("");
transactionsBuffer_.clear();
objectsBuffer_.str("");
objectsBuffer_.clear();
successorBuffer_.str("");
successorBuffer_.clear();
accountTxBuffer_.str("");
accountTxBuffer_.clear();
numRowsInObjectsBuffer_ = 0;
auto res = writeConnection_("COMMIT", yield);
if (!res || res.status() != PGRES_COMMAND_OK)
{
std::stringstream msg;
msg << "Postgres error committing transaction: " << res.msg();
throw std::runtime_error(msg.str());
}
transactionsBuffer_.str("");
transactionsBuffer_.clear();
objectsBuffer_.str("");
objectsBuffer_.clear();
successorBuffer_.str("");
successorBuffer_.clear();
successors_.clear();
accountTxBuffer_.str("");
accountTxBuffer_.clear();
numRowsInObjectsBuffer_ = 0;
});
return !abortWrite_;
}
bool
PostgresBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
PostgresBackend::doOnlineDelete(
std::uint32_t const numLedgersToKeep,
boost::asio::yield_context& yield) const
{
auto rng = fetchLedgerRange();
if (!rng)
return false;
uint32_t minLedger = rng->maxSequence - numLedgersToKeep;
std::uint32_t minLedger = rng->maxSequence - numLedgersToKeep;
if (minLedger <= rng->minSequence)
return false;
uint32_t limit = 2048;
std::uint32_t limit = 2048;
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 0");
pgQuery("SET statement_timeout TO 0", yield);
std::optional<ripple::uint256> cursor;
while (true)
{
auto [objects, curCursor, warning] = retryOnTimeout(
[&]() { return fetchLedgerPage(cursor, minLedger, 256); });
auto [objects, curCursor, warning] = retryOnTimeout([&]() {
return fetchLedgerPage(cursor, minLedger, 256, 0, yield);
});
if (warning)
{
BOOST_LOG_TRIVIAL(warning) << __func__
@@ -739,7 +807,7 @@ PostgresBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
<< std::to_string(minLedger) << '\t' << "\\\\x"
<< ripple::strHex(obj.blob) << '\n';
}
pgQuery.bulkInsert("objects", objectsBuffer.str());
pgQuery.bulkInsert("objects", objectsBuffer.str(), yield);
cursor = curCursor;
if (!cursor)
break;
@@ -749,7 +817,7 @@ PostgresBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
std::stringstream sql;
sql << "DELETE FROM ledgers WHERE ledger_seq < "
<< std::to_string(minLedger);
auto res = pgQuery(sql.str().data());
auto res = pgQuery(sql.str().data(), yield);
if (res.msg() != "ok")
throw std::runtime_error("Error deleting from ledgers table");
}
@@ -757,7 +825,7 @@ PostgresBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
std::stringstream sql;
sql << "DELETE FROM keys WHERE ledger_seq < "
<< std::to_string(minLedger);
auto res = pgQuery(sql.str().data());
auto res = pgQuery(sql.str().data(), yield);
if (res.msg() != "ok")
throw std::runtime_error("Error deleting from keys table");
}
@@ -765,7 +833,7 @@ PostgresBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
std::stringstream sql;
sql << "DELETE FROM books WHERE ledger_seq < "
<< std::to_string(minLedger);
auto res = pgQuery(sql.str().data());
auto res = pgQuery(sql.str().data(), yield);
if (res.msg() != "ok")
throw std::runtime_error("Error deleting from books table");
}

90
src/backend/PostgresBackend.h
View File

@@ -16,62 +16,84 @@ private:
std::shared_ptr<PgPool> pgPool_;
mutable PgQuery writeConnection_;
mutable bool abortWrite_ = false;
mutable boost::asio::thread_pool pool_{16};
uint32_t writeInterval_ = 1000000;
uint32_t inProcessLedger = 0;
std::unordered_set<std::string> successors_;
std::uint32_t writeInterval_ = 1000000;
std::uint32_t inProcessLedger = 0;
mutable std::unordered_set<std::string> successors_;
const char* const set_timeout = "SET statement_timeout TO 10000";
public:
PostgresBackend(boost::json::object const& config);
PostgresBackend(
boost::asio::io_context& ioc,
boost::json::object const& config);
std::optional<uint32_t>
fetchLatestLedgerSequence() const override;
std::optional<std::uint32_t>
fetchLatestLedgerSequence(boost::asio::yield_context& yield) const override;
std::optional<ripple::LedgerInfo>
fetchLedgerBySequence(uint32_t sequence) const override;
fetchLedgerBySequence(
std::uint32_t const sequence,
boost::asio::yield_context& yield) const override;
std::optional<ripple::LedgerInfo>
fetchLedgerByHash(ripple::uint256 const& hash) const override;
fetchLedgerByHash(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const override;
std::optional<Blob>
doFetchLedgerObject(ripple::uint256 const& key, uint32_t sequence)
const override;
doFetchLedgerObject(
ripple::uint256 const& key,
std::uint32_t const sequence,
boost::asio::yield_context& yield) const override;
// returns a transaction, metadata pair
std::optional<TransactionAndMetadata>
fetchTransaction(ripple::uint256 const& hash) const override;
fetchTransaction(
ripple::uint256 const& hash,
boost::asio::yield_context& yield) const override;
std::vector<TransactionAndMetadata>
fetchAllTransactionsInLedger(uint32_t ledgerSequence) const override;
fetchAllTransactionsInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
std::vector<ripple::uint256>
fetchAllTransactionHashesInLedger(uint32_t ledgerSequence) const override;
fetchAllTransactionHashesInLedger(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
std::vector<LedgerObject>
fetchLedgerDiff(uint32_t ledgerSequence) const override;
fetchLedgerDiff(
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
std::optional<LedgerRange>
hardFetchLedgerRange() const override;
hardFetchLedgerRange(boost::asio::yield_context& yield) const override;
std::optional<ripple::uint256>
doFetchSuccessorKey(ripple::uint256 key, uint32_t ledgerSequence)
const override;
doFetchSuccessorKey(
ripple::uint256 key,
std::uint32_t const ledgerSequence,
boost::asio::yield_context& yield) const override;
std::vector<TransactionAndMetadata>
fetchTransactions(
std::vector<ripple::uint256> const& hashes) const override;
std::vector<ripple::uint256> const& hashes,
boost::asio::yield_context& yield) const override;
std::vector<Blob>
doFetchLedgerObjects(
std::vector<ripple::uint256> const& keys,
uint32_t sequence) const override;
std::uint32_t const sequence,
boost::asio::yield_context& yield) const override;
AccountTransactions
fetchAccountTransactions(
ripple::AccountID const& account,
std::uint32_t limit,
std::uint32_t const limit,
bool forward,
std::optional<AccountTransactionsCursor> const& cursor) const override;
std::optional<AccountTransactionsCursor> const& cursor,
boost::asio::yield_context& yield) const override;
void
writeLedger(
@@ -79,18 +101,22 @@ public:
std::string&& ledgerHeader) override;
void
doWriteLedgerObject(std::string&& key, uint32_t seq, std::string&& blob)
override;
doWriteLedgerObject(
std::string&& key,
std::uint32_t const seq,
std::string&& blob) override;
void
writeSuccessor(std::string&& key, uint32_t seq, std::string&& successor)
override;
writeSuccessor(
std::string&& key,
std::uint32_t const seq,
std::string&& successor) override;
void
writeTransaction(
std::string&& hash,
uint32_t seq,
uint32_t date,
std::uint32_t const seq,
std::uint32_t const date,
std::string&& transaction,
std::string&& metadata) override;
@@ -105,13 +131,15 @@ public:
close() override;
void
startWrites() override;
startWrites() const override;
bool
doFinishWrites() override;
doFinishWrites() const override;
bool
doOnlineDelete(uint32_t numLedgersToKeep) const override;
doOnlineDelete(
std::uint32_t const numLedgersToKeep,
boost::asio::yield_context& yield) const override;
};
} // namespace Backend
#endif

12
src/backend/Types.h
View File

@@ -38,8 +38,8 @@ struct TransactionAndMetadata
{
Blob transaction;
Blob metadata;
uint32_t ledgerSequence;
uint32_t date;
std::uint32_t ledgerSequence;
std::uint32_t date;
bool
operator==(const TransactionAndMetadata& other) const
{
@@ -50,8 +50,8 @@ struct TransactionAndMetadata
struct AccountTransactionsCursor
{
uint32_t ledgerSequence;
uint32_t transactionIndex;
std::uint32_t ledgerSequence;
std::uint32_t transactionIndex;
};
struct AccountTransactions
@@ -62,8 +62,8 @@ struct AccountTransactions
struct LedgerRange
{
uint32_t minSequence;
uint32_t maxSequence;
std::uint32_t minSequence;
std::uint32_t maxSequence;
};
constexpr ripple::uint256 firstKey{
"0000000000000000000000000000000000000000000000000000000000000000"};