ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2025-11-21 04:05:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

add types to make it harder to mix up book index and key index

Browse Source
This commit is contained in:
CJ Cobb
2021-05-11 19:06:02 +00:00
parent e7b212a05c
commit ca886fe2c8
6 changed files with 156 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
reporting/BackendIndexer.cpp
View File

@@ -105,8 +105,8 @@ writeKeyFlagLedger(
}
auto start = std::chrono::system_clock::now();
backend.writeKeys(keys, nextFlag, true);
backend.writeKeys({zero}, nextFlag, true);
backend.writeKeys(keys, KeyIndex{nextFlag}, true);
backend.writeKeys({zero}, KeyIndex{nextFlag}, true);
auto end = std::chrono::system_clock::now();
BOOST_LOG_TRIVIAL(info)
<< __func__
@@ -134,8 +134,9 @@ writeBookFlagLedger(
<< " books.size() = " << std::to_string(books.size());
auto start = std::chrono::system_clock::now();
backend.writeBooks(books, nextFlag, true);
backend.writeBooks({{zero, {zero}}}, nextFlag, true);
backend.writeBooks(books, BookIndex{nextFlag}, true);
backend.writeBooks({{zero, {zero}}}, BookIndex{nextFlag}, true);
auto end = std::chrono::system_clock::now();
BOOST_LOG_TRIVIAL(info)
@@ -435,30 +436,28 @@ BackendIndexer::finish(uint32_t ledgerSequence, BackendInterface const& backend)
<< __func__
<< " starting. sequence = " << std::to_string(ledgerSequence);
bool isFirst = false;
uint32_t keyIndex = getKeyIndexOfSeq(ledgerSequence);
uint32_t bookIndex = getBookIndexOfSeq(ledgerSequence);
auto keyIndex = getKeyIndexOfSeq(ledgerSequence);
auto bookIndex = getBookIndexOfSeq(ledgerSequence);
auto rng = backend.fetchLedgerRangeNoThrow();
if (!rng || rng->minSequence == ledgerSequence)
{
isFirst = true;
keyIndex = bookIndex = ledgerSequence;
keyIndex = KeyIndex{ledgerSequence};
bookIndex = BookIndex{ledgerSequence};
}
backend.writeKeys(keys, keyIndex);
backend.writeBooks(books, bookIndex);
if (isFirst)
{
// write completion record
ripple::uint256 zero = {};
backend.writeBooks({{zero, {zero}}}, ledgerSequence);
backend.writeKeys({zero}, ledgerSequence);
writeBookFlagLedgerAsync(ledgerSequence, backend);
writeKeyFlagLedgerAsync(ledgerSequence, backend);
backend.writeBooks({{zero, {zero}}}, bookIndex);
backend.writeKeys({zero}, keyIndex);
}
keys = {};
books = {};
BOOST_LOG_TRIVIAL(info)
<< __func__
<< " finished. sequence = " << std::to_string(ledgerSequence);
}
} // namespace Backend

54
reporting/BackendInterface.h
View File

@@ -53,6 +53,19 @@ struct LedgerRange
uint32_t maxSequence;
};
// The below two structs exist to prevent developers from accidentally mixing up
// the two indexes.
struct BookIndex
{
uint32_t bookIndex;
explicit BookIndex(uint32_t v) : bookIndex(v){};
};
struct KeyIndex
{
uint32_t keyIndex;
explicit KeyIndex(uint32_t v) : keyIndex(v){};
};
class DatabaseTimeout : public std::exception
{
const char*
@@ -148,26 +161,33 @@ public:
{
return keyShift_;
}
uint32_t
KeyIndex
getKeyIndexOfSeq(uint32_t seq) const
{
if (isKeyFlagLedger(seq))
return seq;
return KeyIndex{seq};
auto incr = (1 << keyShift_);
return (seq >> keyShift_ << keyShift_) + incr;
KeyIndex index{(seq >> keyShift_ << keyShift_) + incr};
assert(isKeyFlagLedger(index.keyIndex));
return index;
}
bool
isKeyFlagLedger(uint32_t ledgerSequence) const
{
return (ledgerSequence % (1 << keyShift_)) == 0;
}
uint32_t
BookIndex
getBookIndexOfSeq(uint32_t seq) const
{
if (isBookFlagLedger(seq))
return seq;
return BookIndex{seq};
auto incr = (1 << bookShift_);
return (seq >> bookShift_ << bookShift_) + incr;
BookIndex index{(seq >> bookShift_ << bookShift_) + incr};
assert(isBookFlagLedger(index.bookIndex));
assert(
bookShift_ == keyShift_ || !isKeyFlagLedger(index.bookIndex) ||
!isKeyFlagLedger(index.bookIndex + incr));
return index;
}
bool
isBookFlagLedger(uint32_t ledgerSequence) const
@@ -193,28 +213,28 @@ public:
return indexer_;
}
std::optional<uint32_t>
std::optional<KeyIndex>
getKeyIndexOfSeq(uint32_t seq) const
{
if (indexer_.isKeyFlagLedger(seq))
return seq;
return KeyIndex{seq};
auto rng = fetchLedgerRange();
if (!rng)
return {};
if (rng->minSequence == seq)
return seq;
return KeyIndex{seq};
return indexer_.getKeyIndexOfSeq(seq);
}
std::optional<uint32_t>
std::optional<BookIndex>
getBookIndexOfSeq(uint32_t seq) const
{
if (indexer_.isBookFlagLedger(seq))
return seq;
return BookIndex{seq};
auto rng = fetchLedgerRange();
if (!rng)
return {};
if (rng->minSequence == seq)
return seq;
return BookIndex{seq};
return indexer_.getBookIndexOfSeq(seq);
}
@@ -225,9 +245,11 @@ public:
auto commitRes = doFinishWrites();
if (commitRes)
{
if (indexer_.isBookFlagLedger(ledgerSequence))
bool isFirst =
fetchLedgerRangeNoThrow()->minSequence == ledgerSequence;
if (indexer_.isBookFlagLedger(ledgerSequence) || isFirst)
indexer_.writeBookFlagLedgerAsync(ledgerSequence, *this);
if (indexer_.isKeyFlagLedger(ledgerSequence))
if (indexer_.isKeyFlagLedger(ledgerSequence) || isFirst)
indexer_.writeKeyFlagLedgerAsync(ledgerSequence, *this);
}
return commitRes;
@@ -381,14 +403,14 @@ public:
virtual bool
writeKeys(
std::unordered_set<ripple::uint256> const& keys,
uint32_t ledgerSequence,
KeyIndex const& index,
bool isAsync = false) const = 0;
virtual bool
writeBooks(
std::unordered_map<
ripple::uint256,
std::unordered_set<ripple::uint256>> const& books,
uint32_t ledgerSequence,
BookIndex const& index,
bool isAsync = false) const = 0;
virtual ~BackendInterface()

74
reporting/CassandraBackend.cpp
View File

@@ -405,12 +405,12 @@ CassandraBackend::fetchLedgerPage(
LedgerPage page;
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " ledgerSequence = " << std::to_string(ledgerSequence)
<< " index = " << std::to_string(*index);
<< " index = " << std::to_string(index->keyIndex);
if (cursor)
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " - Cursor = " << ripple::strHex(*cursor);
CassandraStatement statement{selectKeys_};
statement.bindInt(*index);
statement.bindInt(index->keyIndex);
if (cursor)
statement.bindBytes(*cursor);
else
@@ -508,10 +508,10 @@ CassandraBackend::fetchBookOffers(
return {{}, {}};
auto readBooks =
[this, &book, &limit, &limitTuningFactor]
(std::uint32_t sequence)
-> std::pair<bool, std::vector<std::pair<std::uint64_t, ripple::uint256>>>
{
[this, &book, &limit, &limitTuningFactor](std::uint32_t sequence)
-> std::pair<
bool,
std::vector<std::pair<std::uint64_t, ripple::uint256>>> {
CassandraStatement completeQuery{completeBook_};
completeQuery.bindInt(sequence);
CassandraResult completeResult = executeSyncRead(completeQuery);
@@ -523,8 +523,9 @@ CassandraBackend::fetchBookOffers(
statement.bindBytes(book.data(), 24);
statement.bindInt(sequence);
BOOST_LOG_TRIVIAL(info) << __func__ << " upper = " << std::to_string(sequence)
<< " book = " << ripple::strHex(std::string((char*)book.data(), 24));
BOOST_LOG_TRIVIAL(info)
<< __func__ << " upper = " << std::to_string(sequence) << " book = "
<< ripple::strHex(std::string((char*)book.data(), 24));
ripple::uint256 zero = beast::zero;
statement.bindBytes(zero.data(), 8);
@@ -560,8 +561,8 @@ CassandraBackend::fetchBookOffers(
return {complete, keys};
};
auto upper = indexer_.getBookIndexOfSeq(ledgerSequence);
auto [complete, quality_keys] = readBooks(upper);
auto upper = getBookIndexOfSeq(ledgerSequence);
auto [complete, quality_keys] = readBooks(upper->bookIndex);
BOOST_LOG_TRIVIAL(debug)
<< __func__ << " - populated keys. num keys = " << quality_keys.size();
@@ -573,7 +574,7 @@ CassandraBackend::fetchBookOffers(
BOOST_LOG_TRIVIAL(info) << "May be incomplete. Fetching other page";
auto bookShift = indexer_.getBookShift();
std::uint32_t lower = upper - (1 << bookShift);
std::uint32_t lower = upper->bookIndex - (1 << bookShift);
auto originalKeys = std::move(quality_keys);
auto [lowerComplete, otherKeys] = readBooks(lower);
@@ -581,17 +582,19 @@ CassandraBackend::fetchBookOffers(
std::vector<std::pair<std::uint64_t, ripple::uint256>> merged_keys;
merged_keys.reserve(originalKeys.size() + otherKeys.size());
std::merge(originalKeys.begin(), originalKeys.end(),
otherKeys.begin(), otherKeys.end(),
std::merge(
originalKeys.begin(),
originalKeys.end(),
otherKeys.begin(),
otherKeys.end(),
std::back_inserter(merged_keys),
[](auto pair1, auto pair2)
{
return pair1.first < pair2.first;
});
[](auto pair1, auto pair2) { return pair1.first < pair2.first; });
}
std::vector<ripple::uint256> merged(quality_keys.size());
std::transform(quality_keys.begin(), quality_keys.end(),
std::transform(
quality_keys.begin(),
quality_keys.end(),
std::back_inserter(merged),
[](auto pair) { return pair.second; });
@@ -605,8 +608,8 @@ CassandraBackend::fetchBookOffers(
auto end = std::chrono::system_clock::now();
auto duration = ((end - start).count()) / 1000000000.0;
BOOST_LOG_TRIVIAL(info) << "Book object fetch took "
<< std::to_string(duration) << " seconds.";
BOOST_LOG_TRIVIAL(info)
<< "Book object fetch took " << std::to_string(duration) << " seconds.";
std::vector<LedgerObject> results;
for (size_t i = 0; i < objs.size(); ++i)
@@ -616,7 +619,7 @@ CassandraBackend::fetchBookOffers(
}
return {results, {}, warning};
}
} // namespace Backend
struct WriteBookCallbackData
{
CassandraBackend const& backend;
@@ -775,14 +778,9 @@ writeKeyCallback(CassFuture* fut, void* cbData)
bool
CassandraBackend::writeKeys(
std::unordered_set<ripple::uint256> const& keys,
uint32_t ledgerSequence,
KeyIndex const& index,
bool isAsync) const
{
BOOST_LOG_TRIVIAL(info)
<< __func__ << " Ledger = " << std::to_string(ledgerSequence)
<< " . num keys = " << std::to_string(keys.size())
<< " . concurrentLimit = "
<< std::to_string(indexerMaxRequestsOutstanding);
std::atomic_uint32_t numRemaining = keys.size();
std::condition_variable cv;
std::mutex mtx;
@@ -790,11 +788,16 @@ CassandraBackend::writeKeys(
cbs.reserve(keys.size());
uint32_t concurrentLimit =
isAsync ? indexerMaxRequestsOutstanding : keys.size();
BOOST_LOG_TRIVIAL(info)
<< __func__ << " Ledger = " << std::to_string(index.keyIndex)
<< " . num keys = " << std::to_string(keys.size())
<< " . concurrentLimit = "
<< std::to_string(indexerMaxRequestsOutstanding);
uint32_t numSubmitted = 0;
for (auto& key : keys)
{
cbs.push_back(std::make_shared<WriteKeyCallbackData>(
*this, key, ledgerSequence, cv, mtx, numRemaining));
*this, key, index.keyIndex, cv, mtx, numRemaining));
writeKey(*cbs.back());
++numSubmitted;
BOOST_LOG_TRIVIAL(trace) << __func__ << "Submitted a write request";
@@ -828,11 +831,11 @@ CassandraBackend::writeBooks(
std::unordered_map<
ripple::uint256,
std::unordered_set<ripple::uint256>> const& books,
uint32_t ledgerSequence,
BookIndex const& index,
bool isAsync) const
{
BOOST_LOG_TRIVIAL(info)
<< __func__ << " Ledger = " << std::to_string(ledgerSequence)
<< __func__ << " Ledger = " << std::to_string(index.bookIndex)
<< " . num books = " << std::to_string(books.size());
std::condition_variable cv;
std::mutex mtx;
@@ -852,7 +855,7 @@ CassandraBackend::writeBooks(
*this,
book.first,
offer,
ledgerSequence,
index.bookIndex,
cv,
mtx,
numOutstanding));
@@ -1100,7 +1103,7 @@ CassandraBackend::runIndexer(uint32_t ledgerSequence) const
*/
}
bool
CassandraBackend::doOnlineDelete(uint32_t minLedgerToKeep) const
CassandraBackend::doOnlineDelete(uint32_t numLedgersToKeep) const
{
throw std::runtime_error("doOnlineDelete : unimplemented");
return false;
@@ -1386,8 +1389,10 @@ CassandraBackend::open(bool readOnly)
query.str("");
query << "CREATE TABLE IF NOT EXISTS " << tablePrefix << "books"
<< " ( book blob, sequence bigint, quality_key tuple<blob, blob>, PRIMARY KEY "
"((book, sequence), quality_key)) WITH CLUSTERING ORDER BY (quality_key "
<< " ( book blob, sequence bigint, quality_key tuple<blob, "
"blob>, PRIMARY KEY "
"((book, sequence), quality_key)) WITH CLUSTERING ORDER BY "
"(quality_key "
"ASC)";
if (!executeSimpleStatement(query.str()))
continue;
@@ -1568,7 +1573,6 @@ CassandraBackend::open(bool readOnly)
if (!completeBook_.prepareStatement(query, session_.get()))
continue;
query.str("");
query << " INSERT INTO " << tablePrefix << "account_tx"
<< " (account, seq_idx, hash) "

4
reporting/CassandraBackend.h
View File

@@ -1014,14 +1014,14 @@ public:
bool
writeKeys(
std::unordered_set<ripple::uint256> const& keys,
uint32_t ledgerSequence,
KeyIndex const& index,
bool isAsync = false) const;
bool
writeBooks(
std::unordered_map<
ripple::uint256,
std::unordered_set<ripple::uint256>> const& books,
uint32_t ledgerSequence,
BookIndex const& index,
bool isAsync = false) const override;
BookOffersPage
fetchBookOffers(

57
reporting/PostgresBackend.cpp
View File

@@ -335,7 +335,8 @@ PostgresBackend::fetchLedgerPage(
PgQuery pgQuery(pgPool_);
pgQuery("SET statement_timeout TO 10000");
std::stringstream sql;
sql << "SELECT key FROM keys WHERE ledger_seq = " << std::to_string(*index);
sql << "SELECT key FROM keys WHERE ledger_seq = "
<< std::to_string(index->keyIndex);
if (cursor)
sql << " AND key > \'\\x" << ripple::strHex(*cursor) << "\'";
sql << " ORDER BY key ASC LIMIT " << std::to_string(limit);
@@ -389,15 +390,13 @@ PostgresBackend::fetchBookOffers(
ripple::uint256 bookEnd = ripple::getQualityNext(bookBase);
using bookKeyPair = std::pair<ripple::uint256, ripple::uint256>;
auto getBooks =
[this, &bookBase, &bookEnd, &limit, &limitTuningFactor]
(std::uint32_t sequence)
-> std::pair<bool, std::vector<bookKeyPair>>
{
auto getBooks = [this, &bookBase, &bookEnd, &limit, &limitTuningFactor](
std::uint32_t sequence)
-> std::pair<bool, std::vector<bookKeyPair>> {
BOOST_LOG_TRIVIAL(info) << __func__ << ": Fetching books between "
<< "0x" << ripple::strHex(bookBase) << " and "
<< "0x" << ripple::strHex(bookEnd) << "at ledger "
<< std::to_string(sequence);
<< "0x" << ripple::strHex(bookEnd)
<< "at ledger " << std::to_string(sequence);
auto start = std::chrono::system_clock::now();
@@ -432,8 +431,7 @@ PostgresBackend::fetchBookOffers(
auto duration = ((end - start).count()) / 1000000000.0;
BOOST_LOG_TRIVIAL(info) << "Postgres book key fetch took "
<< std::to_string(duration)
<< " seconds";
<< std::to_string(duration) << " seconds";
if (size_t numRows = checkResult(res, 2))
{
@@ -453,13 +451,11 @@ PostgresBackend::fetchBookOffers(
};
auto fetchObjects =
[this]
(std::vector<bookKeyPair> const& pairs,
[this](
std::vector<bookKeyPair> const& pairs,
std::uint32_t sequence,
std::uint32_t limit,
std::optional<std::string> warning)
-> BookOffersPage
{
std::optional<std::string> warning) -> BookOffersPage {
std::vector<ripple::uint256> allKeys(pairs.size());
for (auto const& pair : pairs)
allKeys.push_back(pair.second);
@@ -474,11 +470,10 @@ PostgresBackend::fetchBookOffers(
auto end = std::chrono::system_clock::now();
auto duration = ((end - start).count()) / 1000000000.0;
BOOST_LOG_TRIVIAL(info) << "Postgres book objects fetch took "
<< std::to_string(duration)
BOOST_LOG_TRIVIAL(info)
<< "Postgres book objects fetch took " << std::to_string(duration)
<< " seconds. "
<< "Fetched "
<< std::to_string(ledgerEntries.size())
<< "Fetched " << std::to_string(ledgerEntries.size())
<< " ledger entries";
std::vector<LedgerObject> objects;
@@ -492,9 +487,9 @@ PostgresBackend::fetchBookOffers(
};
std::uint32_t bookShift = indexer_.getBookShift();
auto upper = indexer_.getBookIndexOfSeq(ledgerSequence);
auto upper = getBookIndexOfSeq(ledgerSequence);
auto [upperComplete, upperResults] = getBooks(upper);
auto [upperComplete, upperResults] = getBooks(upper->bookIndex);
BOOST_LOG_TRIVIAL(info) << __func__ << ": Upper results found "
<< upperResults.size() << " books.";
@@ -508,7 +503,7 @@ PostgresBackend::fetchBookOffers(
BOOST_LOG_TRIVIAL(info) << "Upper book page is not complete "
<< "fetching again";
auto lower = upper - (1 << bookShift);
auto lower = upper->bookIndex - (1 << bookShift);
if (lower < rng->minSequence)
lower = rng->minSequence;
@@ -521,11 +516,13 @@ PostgresBackend::fetchBookOffers(
std::vector<bookKeyPair> pairs;
pairs.reserve(upperResults.size() + lowerResults.size());
std::merge(upperResults.begin(), upperResults.end(),
lowerResults.begin(), lowerResults.end(),
std::merge(
upperResults.begin(),
upperResults.end(),
lowerResults.begin(),
lowerResults.end(),
std::back_inserter(pairs),
[](bookKeyPair pair1, bookKeyPair pair2) -> bool
{
[](bookKeyPair pair1, bookKeyPair pair2) -> bool {
return pair1.first < pair2.first;
});
@@ -806,7 +803,7 @@ PostgresBackend::doFinishWrites() const
bool
PostgresBackend::writeKeys(
std::unordered_set<ripple::uint256> const& keys,
uint32_t ledgerSequence,
KeyIndex const& index,
bool isAsync) const
{
BOOST_LOG_TRIVIAL(debug) << __func__;
@@ -816,7 +813,7 @@ PostgresBackend::writeKeys(
size_t numRows = 0;
for (auto& key : keys)
{
keysBuffer << std::to_string(ledgerSequence) << '\t' << "\\\\x"
keysBuffer << std::to_string(index.keyIndex) << '\t' << "\\\\x"
<< ripple::strHex(key) << '\n';
numRows++;
// If the buffer gets too large, the insert fails. Not sure why. So we
@@ -841,7 +838,7 @@ PostgresBackend::writeBooks(
std::unordered_map<
ripple::uint256,
std::unordered_set<ripple::uint256>> const& books,
uint32_t ledgerSequence,
BookIndex const& index,
bool isAsync) const
{
BOOST_LOG_TRIVIAL(debug) << __func__;
@@ -854,7 +851,7 @@ PostgresBackend::writeBooks(
{
for (auto& offer : book.second)
{
booksBuffer << std::to_string(ledgerSequence) << '\t' << "\\\\x"
booksBuffer << std::to_string(index.bookIndex) << '\t' << "\\\\x"
<< ripple::strHex(book.first) << '\t' << "\\\\x"
<< ripple::strHex(offer) << '\n';
numRows++;

4
reporting/PostgresBackend.h
View File

@@ -117,14 +117,14 @@ public:
bool
writeKeys(
std::unordered_set<ripple::uint256> const& keys,
uint32_t ledgerSequence,
KeyIndex const& index,
bool isAsync = false) const override;
bool
writeBooks(
std::unordered_map<
ripple::uint256,
std::unordered_set<ripple::uint256>> const& books,
uint32_t ledgerSequence,
BookIndex const& index,
bool isAsync = false) const override;
};
} // namespace Backend