Refactored BucketListDB for concurrency

Author: SirTyson

src/bucket/Bucket.cpp

@@ -77,18 +77,6 @@ Bucket::Bucket()
 {
 }
 
-XDRInputFileStream&
-Bucket::getStream()
-{
-    if (!mStream)
-    {
-        mStream = std::make_unique<XDRInputFileStream>();
-        releaseAssertOrThrow(!mFilename.empty());
-        mStream->open(mFilename.string());
-    }
-    return *mStream;
-}
-
 Hash const&
 Bucket::getHash() const
 {
@@ -139,157 +127,6 @@ void
 Bucket::freeIndex()
 {
     mIndex.reset(nullptr);
-    mStream.reset(nullptr);
-}
-
-std::optional<BucketEntry>
-Bucket::getEntryAtOffset(LedgerKey const& k, std::streamoff pos,
-                         size_t pageSize)
-{
-    ZoneScoped;
-    auto& stream = getStream();
-    stream.seek(pos);
-
-    BucketEntry be;
-    if (pageSize == 0)
-    {
-        if (stream.readOne(be))
-        {
-            return std::make_optional(be);
-        }
-    }
-    else if (stream.readPage(be, k, pageSize))
-    {
-        return std::make_optional(be);
-    }
-
-    // Mark entry miss for metrics
-    getIndex().markBloomMiss();
-    return std::nullopt;
-}
-
-std::optional<BucketEntry>
-Bucket::getBucketEntry(LedgerKey const& k)
-{
-    ZoneScoped;
-    auto pos = getIndex().lookup(k);
-    if (pos.has_value())
-    {
-        return getEntryAtOffset(k, pos.value(), getIndex().getPageSize());
-    }
-
-    return std::nullopt;
-}
-
-// When searching for an entry, BucketList calls this function on every bucket.
-// Since the input is sorted, we do a binary search for the first key in keys.
-// If we find the entry, we remove the found key from keys so that later buckets
-// do not load shadowed entries. If we don't find the entry, we do not remove it
-// from keys so that it will be searched for again at a lower level.
-void
-Bucket::loadKeys(std::set<LedgerKey, LedgerEntryIdCmp>& keys,
-                 std::vector<LedgerEntry>& result)
-{
-    ZoneScoped;
-
-    auto currKeyIt = keys.begin();
-    auto const& index = getIndex();
-    auto indexIter = index.begin();
-    while (currKeyIt != keys.end() && indexIter != index.end())
-    {
-        auto [offOp, newIndexIter] = index.scan(indexIter, *currKeyIt);
-        indexIter = newIndexIter;
-        if (offOp)
-        {
-            auto entryOp =
-                getEntryAtOffset(*currKeyIt, *offOp, getIndex().getPageSize());
-            if (entryOp)
-            {
-                if (entryOp->type() != DEADENTRY)
-                {
-                    result.push_back(entryOp->liveEntry());
-                }
-
-                currKeyIt = keys.erase(currKeyIt);
-                continue;
-            }
-        }
-
-        ++currKeyIt;
-    }
-}
-
-void
-Bucket::loadPoolShareTrustLinessByAccount(
-    AccountID const& accountID, UnorderedSet<LedgerKey>& deadTrustlines,
-    UnorderedMap<LedgerKey, LedgerEntry>& liquidityPoolKeyToTrustline,
-    LedgerKeySet& liquidityPoolKeys)
-{
-    ZoneScoped;
-
-    // Takes a LedgerKey or LedgerEntry::_data_t, returns true if entry is a
-    // poolshare trusline for the given accountID
-    auto trustlineCheck = [&accountID](auto const& entry) {
-        return entry.type() == TRUSTLINE &&
-               entry.trustLine().asset.type() == ASSET_TYPE_POOL_SHARE &&
-               entry.trustLine().accountID == accountID;
-    };
-
-    // Get upper and lower bound for poolshare trustline range associated
-    // with this account
-    auto searchRange = getIndex().getPoolshareTrustlineRange(accountID);
-    if (searchRange.first == 0)
-    {
-        // No poolshare trustlines, exit
-        return;
-    }
-
-    BucketEntry be;
-    auto& stream = getStream();
-    stream.seek(searchRange.first);
-    while (stream && stream.pos() < searchRange.second && stream.readOne(be))
-    {
-        LedgerEntry entry;
-        switch (be.type())
-        {
-        case LIVEENTRY:
-        case INITENTRY:
-            entry = be.liveEntry();
-            break;
-        case DEADENTRY:
-        {
-            auto key = be.deadEntry();
-
-            // If we find a valid trustline key and we have not seen the
-            // key yet, mark it as dead so we do not load a shadowed version
-            // later
-            if (trustlineCheck(key))
-            {
-                deadTrustlines.emplace(key);
-            }
-            continue;
-        }
-        case METAENTRY:
-        default:
-            throw std::invalid_argument("Indexed METAENTRY");
-        }
-
-        // If this is a pool share trustline that matches the accountID and
-        // is not shadowed, add it to results
-        if (trustlineCheck(entry.data) &&
-            deadTrustlines.find(LedgerEntryKey(entry)) == deadTrustlines.end())
-        {
-            auto const& poolshareID =
-                entry.data.trustLine().asset.liquidityPoolID();
-
-            LedgerKey key;
-            key.type(LIQUIDITY_POOL);
-            key.liquidityPool().liquidityPoolID = poolshareID;
-
-            liquidityPoolKeyToTrustline.emplace(key, entry);
-            liquidityPoolKeys.emplace(key);
-        }
-    }
 }
 
 #ifdef BUILD_TESTS
@@ -837,12 +674,12 @@ mergeCasesWithEqualKeys(MergeCounters& mc, BucketInputIterator& oi,
 }
 
 bool
-Bucket::scanForEviction(AbstractLedgerTxn& ltx, EvictionIterator& iter,
-                        uint64_t& bytesToScan, uint32_t& maxEntriesToEvict,
-                        uint32_t ledgerSeq,
-                        medida::Counter& entriesEvictedCounter,
-                        medida::Counter& bytesScannedForEvictionCounter,
-                        std::optional<EvictionMetrics>& metrics)
+Bucket::scanForEvictionLegacySQL(
+    AbstractLedgerTxn& ltx, EvictionIterator& iter, uint64_t& bytesToScan,
+    uint32_t& maxEntriesToEvict, uint32_t ledgerSeq,
+    medida::Counter& entriesEvictedCounter,
+    medida::Counter& bytesScannedForEvictionCounter,
+    std::optional<EvictionStatistics>& stats) const
 {
     ZoneScoped;
     if (isEmpty())
@@ -857,7 +694,8 @@ Bucket::scanForEviction(AbstractLedgerTxn& ltx, EvictionIterator& iter,
         return true;
     }
 
-    auto& stream = getStream();
+    XDRInputFileStream stream{};
+    stream.open(mFilename);
     stream.seek(iter.bucketFileOffset);
 
     BucketEntry be;
@@ -898,10 +736,10 @@ Bucket::scanForEviction(AbstractLedgerTxn& ltx, EvictionIterator& iter,
                 if (shouldEvict())
                 {
                     ZoneNamedN(evict, "evict entry", true);
-                    if (metrics.has_value())
+                    if (stats.has_value())
                     {
-                        ++metrics->numEntriesEvicted;
-                        metrics->evictedEntriesAgeSum +=
+                        ++stats->numEntriesEvicted;
+                        stats->evictedEntriesAgeSum +=
                             ledgerSeq - liveUntilLedger;
                     }
 

src/bucket/Bucket.h

@@ -38,7 +38,7 @@ namespace stellar
 class AbstractLedgerTxn;
 class Application;
 class BucketManager;
-struct EvictionMetrics;
+struct EvictionStatistics;
 
 class Bucket : public std::enable_shared_from_this<Bucket>,
                public NonMovableOrCopyable
@@ -49,22 +49,9 @@ class Bucket : public std::enable_shared_from_this<Bucket>,
 
     std::unique_ptr<BucketIndex const> mIndex{};
 
-    // Lazily-constructed and retained for read path.
-    std::unique_ptr<XDRInputFileStream> mStream;
-
     // Returns index, throws if index not yet initialized
     BucketIndex const& getIndex() const;
 
-    // Returns (lazily-constructed) file stream for bucket file. Note
-    // this might be in some random position left over from a previous read --
-    // must be seek()'ed before use.
-    XDRInputFileStream& getStream();
-
-    // Loads the bucket entry for LedgerKey k. Starts at file offset pos and
-    // reads until key is found or the end of the page.
-    std::optional<BucketEntry>
-    getEntryAtOffset(LedgerKey const& k, std::streamoff pos, size_t pageSize);
-
     static std::string randomFileName(std::string const& tmpDir,
                                       std::string ext);
 
@@ -98,25 +85,6 @@ class Bucket : public std::enable_shared_from_this<Bucket>,
     // Sets index, throws if index is already set
     void setIndex(std::unique_ptr<BucketIndex const>&& index);
 
-    // Loads bucket entry for LedgerKey k.
-    std::optional<BucketEntry> getBucketEntry(LedgerKey const& k);
-
-    // Loads LedgerEntry's for given keys. When a key is found, the
-    // entry is added to result and the key is removed from keys.
-    void loadKeys(std::set<LedgerKey, LedgerEntryIdCmp>& keys,
-                  std::vector<LedgerEntry>& result);
-
-    // Loads all poolshare trustlines for the given account. Trustlines are
-    // stored with their corresponding liquidity pool key in
-    // liquidityPoolKeyToTrustline. All liquidity pool keys corresponding to
-    // loaded trustlines are also reduntantly stored in liquidityPoolKeys.
-    // If a trustline key is in deadTrustlines, it is not loaded. Whenever a
-    // dead trustline is found, its key is added to deadTrustlines.
-    void loadPoolShareTrustLinessByAccount(
-        AccountID const& accountID, UnorderedSet<LedgerKey>& deadTrustlines,
-        UnorderedMap<LedgerKey, LedgerEntry>& liquidityPoolKeyToTrustline,
-        LedgerKeySet& liquidityPoolKeys);
-
     // At version 11, we added support for INITENTRY and METAENTRY. Before this
     // we were only supporting LIVEENTRY and DEADENTRY.
     static constexpr ProtocolVersion
@@ -137,18 +105,6 @@ class Bucket : public std::enable_shared_from_this<Bucket>,
     static std::string randomBucketName(std::string const& tmpDir);
     static std::string randomBucketIndexName(std::string const& tmpDir);
 
-    // Returns false if eof reached, true otherwise. Modifies iter as the bucket
-    // is scanned. Also modifies bytesToScan and maxEntriesToEvict such that
-    // after this function returns:
-    // bytesToScan -= amount_bytes_scanned
-    // maxEntriesToEvict -= entries_evicted
-    bool scanForEviction(AbstractLedgerTxn& ltx, EvictionIterator& iter,
-                         uint64_t& bytesToScan, uint32_t& maxEntriesToEvict,
-                         uint32_t ledgerSeq,
-                         medida::Counter& entriesEvictedCounter,
-                         medida::Counter& bytesScannedForEvictionCounter,
-                         std::optional<EvictionMetrics>& metrics);
-
 #ifdef BUILD_TESTS
     // "Applies" the bucket to the database. For each entry in the bucket,
     // if the entry is init or live, creates or updates the corresponding
@@ -164,6 +120,19 @@ class Bucket : public std::enable_shared_from_this<Bucket>,
 
 #endif // BUILD_TESTS
 
+    // Returns false if eof reached, true otherwise. Modifies iter as the bucket
+    // is scanned. Also modifies bytesToScan and maxEntriesToEvict such that
+    // after this function returns:
+    // bytesToScan -= amount_bytes_scanned
+    // maxEntriesToEvict -= entries_evicted
+    bool
+    scanForEvictionLegacySQL(AbstractLedgerTxn& ltx, EvictionIterator& iter,
+                             uint64_t& bytesToScan, uint32_t& maxEntriesToEvict,
+                             uint32_t ledgerSeq,
+                             medida::Counter& entriesEvictedCounter,
+                             medida::Counter& bytesScannedForEvictionCounter,
+                             std::optional<EvictionStatistics>& stats) const;
+
     // Create a fresh bucket from given vectors of init (created) and live
     // (updated) LedgerEntries, and dead LedgerEntryKeys. The bucket will
     // be sorted, hashed, and adopted in the provided BucketManager.
@@ -196,5 +165,7 @@ class Bucket : public std::enable_shared_from_this<Bucket>,
     static uint32_t getBucketVersion(std::shared_ptr<Bucket> const& bucket);
     static uint32_t
     getBucketVersion(std::shared_ptr<Bucket const> const& bucket);
+
+    friend class SearchableBucketSnapshot;
 };
 }