Push cut point out through receiving messages

Push cut point out a tiny bit

todo

Identify where to push back next

Atomic ledger state

Refactor out tx metrics from queue adding

In preparation for moving the cut point

Consistency

Getting close to having this all pushed out

Just need to wire up the final bit

Remove old post to bg thread

Indirection around tx queue storage

Feed tx queues to peer

Wire up background flow

Debug output

Fix tracy build

More debug info around broadcasting

Maybe fix deadlock

Fix deadlock with TCPPeer

Reduce main thread posts on broadcast

Add thread

Though it's not being used anywhere

Actually run on new thread

Another memory leak fix

Fix `maybeExecuteInBackground` to check thread type

Save main thread type

Author: bboston7

src/herder/Herder.h

@@ -225,5 +225,13 @@ class Herder
     virtual TransactionFrameBaseConstPtr getTx(Hash const& hash) const = 0;
 
     virtual void beginApply() = 0;
+
+    // TODO: Docs
+    virtual TransactionQueuesPtr getTransactionQueues() const = 0;
+
+    // TODO: Docs
+    static TransactionQueue::AddResult
+    recvTransaction(TransactionQueuesPtr txQueues, TransactionFrameBasePtr tx,
+                    bool submittedFromSelf);
 };
 }

src/herder/HerderImpl.cpp

@@ -272,14 +272,7 @@ HerderImpl::shutdown()
                    "Shutdown interrupting quorum transitive closure analysis.");
         mLastQuorumMapIntersectionState.mInterruptFlag = true;
     }
-    if (mTransactionQueue)
-    {
-        mTransactionQueue->shutdown();
-    }
-    if (mSorobanTransactionQueue)
-    {
-        mSorobanTransactionQueue->shutdown();
-    }
+    mTransactionQueues->shutdown();
 
     mTxSetGarbageCollectTimer.cancel();
 }
@@ -589,22 +582,35 @@ HerderImpl::emitEnvelope(SCPEnvelope const& envelope)
     broadcast(envelope);
 }
 
+// TODO: Move to Herder.cpp?
 TransactionQueue::AddResult
-HerderImpl::recvTransaction(TransactionFrameBasePtr tx, bool submittedFromSelf)
+Herder::recvTransaction(TransactionQueuesPtr txQueues,
+                            TransactionFrameBasePtr tx, bool submittedFromSelf)
 {
     ZoneScoped;
+    ClassicTransactionQueue& classicTxQueue =
+        txQueues->getClassicTransactionQueue();
     TransactionQueue::AddResult result(
         TransactionQueue::AddResultCode::ADD_STATUS_COUNT);
 
     // Allow txs of the same kind to reach the tx queue in case it can be
     // replaced by fee
+    // TODO: Is there a potential TOCTOU issue here as sourceAccountPending
+    // could change before adding? I think no because the other competing thread
+    // would be whatever is handling ledger close. However, that will only
+    // decrease the sourceAccountPending value, which means this erroneously
+    // rejects (which is safe). I guess it's possible for a user-submitted
+    // transaction to come in and conflict with the overlay thread, but that
+    // would require them to be simultaneously running two clients and
+    // submitting from both of them. Still, it might be safest to use some kind
+    // of atomic function that handles both this check AND the add.
     bool hasSoroban =
-        mSorobanTransactionQueue &&
-        mSorobanTransactionQueue->sourceAccountPending(tx->getSourceID()) &&
+        txQueues->hasSorobanTransactionQueue() &&
+        txQueues->getSorobanTransactionQueue().sourceAccountPending(
+            tx->getSourceID()) &&
         !tx->isSoroban();
-    bool hasClassic =
-        mTransactionQueue->sourceAccountPending(tx->getSourceID()) &&
-        tx->isSoroban();
+    bool hasClassic = classicTxQueue.sourceAccountPending(tx->getSourceID()) &&
+                      tx->isSoroban();
     if (hasSoroban || hasClassic)
     {
         CLOG_DEBUG(Herder,
@@ -617,31 +623,12 @@ HerderImpl::recvTransaction(TransactionFrameBasePtr tx, bool submittedFromSelf)
     }
     else if (!tx->isSoroban())
     {
-        if (mApp.getConfig().BACKGROUND_TX_QUEUE && !submittedFromSelf)
-        {
-            mApp.postOnOverlayThread(
-                [this, tx]() { mTransactionQueue->tryAdd(tx, false); },
-                "try add tx");
-            result.code = TransactionQueue::AddResultCode::ADD_STATUS_UNKNOWN;
-        }
-        else
-        {
-            result = mTransactionQueue->tryAdd(tx, submittedFromSelf);
-        }
+        result = classicTxQueue.tryAdd(tx, submittedFromSelf);
     }
-    else if (mSorobanTransactionQueue)
+    else if (txQueues->hasSorobanTransactionQueue())
     {
-        if (mApp.getConfig().BACKGROUND_TX_QUEUE && !submittedFromSelf)
-        {
-            mApp.postOnOverlayThread(
-                [this, tx]() { mSorobanTransactionQueue->tryAdd(tx, false); },
-                "try add tx");
-            result.code = TransactionQueue::AddResultCode::ADD_STATUS_UNKNOWN;
-        }
-        else
-        {
-            result = mSorobanTransactionQueue->tryAdd(tx, submittedFromSelf);
-        }
+        result = txQueues->getSorobanTransactionQueue().tryAdd(
+            tx, submittedFromSelf);
     }
     else
     {
@@ -661,6 +648,13 @@ HerderImpl::recvTransaction(TransactionFrameBasePtr tx, bool submittedFromSelf)
     return result;
 }
 
+TransactionQueue::AddResult
+HerderImpl::recvTransaction(TransactionFrameBasePtr tx, bool submittedFromSelf)
+{
+    ZoneScoped;
+    return Herder::recvTransaction(mTransactionQueues, tx, submittedFromSelf);
+}
+
 bool
 HerderImpl::checkCloseTime(SCPEnvelope const& envelope, bool enforceRecent)
 {
@@ -943,13 +937,7 @@ HerderImpl::externalizeValue(TxSetXDRFrameConstPtr txSet, uint32_t ledgerSeq,
 bool
 HerderImpl::sourceAccountPending(AccountID const& accountID) const
 {
-    bool accPending = mTransactionQueue->sourceAccountPending(accountID);
-    if (mSorobanTransactionQueue)
-    {
-        accPending = accPending ||
-                     mSorobanTransactionQueue->sourceAccountPending(accountID);
-    }
-    return accPending;
+    return mTransactionQueues->sourceAccountPending(accountID);
 }
 
 #endif
@@ -1112,13 +1100,12 @@ HerderImpl::getPendingEnvelopes()
 ClassicTransactionQueue&
 HerderImpl::getTransactionQueue()
 {
-    return *mTransactionQueue;
+    return mTransactionQueues->getClassicTransactionQueue();
 }
 SorobanTransactionQueue&
 HerderImpl::getSorobanTransactionQueue()
 {
-    releaseAssert(mSorobanTransactionQueue);
-    return *mSorobanTransactionQueue;
+    return mTransactionQueues->getSorobanTransactionQueue();
 }
 #endif
 
@@ -1411,14 +1398,16 @@ HerderImpl::triggerNextLedger(uint32_t ledgerSeqToTrigger,
     // it's guaranteed to be up-to-date
     auto const& lcl = mLedgerManager.getLastClosedLedgerHeader();
     PerPhaseTransactionList txPhases;
-    txPhases.emplace_back(mTransactionQueue->getTransactions(lcl.header));
+    txPhases.emplace_back(
+        mTransactionQueues->getClassicTransactionQueue().getTransactions(
+            lcl.header));
 
     if (protocolVersionStartsFrom(lcl.header.ledgerVersion,
                                   SOROBAN_PROTOCOL_VERSION))
     {
-        releaseAssert(mSorobanTransactionQueue);
         txPhases.emplace_back(
-            mSorobanTransactionQueue->getTransactions(lcl.header));
+            mTransactionQueues->getSorobanTransactionQueue().getTransactions(
+                lcl.header));
     }
 
     // We pick as next close time the current time unless it's before the last
@@ -1485,12 +1474,11 @@ HerderImpl::triggerNextLedger(uint32_t ledgerSeqToTrigger,
     if (protocolVersionStartsFrom(lcl.header.ledgerVersion,
                                   SOROBAN_PROTOCOL_VERSION))
     {
-        releaseAssert(mSorobanTransactionQueue);
-        mSorobanTransactionQueue->ban(
+        mTransactionQueues->getSorobanTransactionQueue().ban(
             invalidTxPhases[static_cast<size_t>(TxSetPhase::SOROBAN)]);
     }
 
-    mTransactionQueue->ban(
+    mTransactionQueues->getClassicTransactionQueue().ban(
         invalidTxPhases[static_cast<size_t>(TxSetPhase::CLASSIC)]);
 
     auto txSetHash = proposedSet->getContentsHash();
@@ -2190,18 +2178,18 @@ HerderImpl::maybeSetupSorobanQueue(uint32_t protocolVersion)
 {
     if (protocolVersionStartsFrom(protocolVersion, SOROBAN_PROTOCOL_VERSION))
     {
-        if (!mSorobanTransactionQueue)
+        if (!mTransactionQueues->hasSorobanTransactionQueue())
         {
             releaseAssert(mTxQueueBucketSnapshot);
-            mSorobanTransactionQueue =
+            mTransactionQueues->setSorobanTransactionQueue(
                 std::make_unique<SorobanTransactionQueue>(
                     mApp, mTxQueueBucketSnapshot,
                     TRANSACTION_QUEUE_TIMEOUT_LEDGERS,
                     TRANSACTION_QUEUE_BAN_LEDGERS,
-                    SOROBAN_TRANSACTION_QUEUE_SIZE_MULTIPLIER);
+                    SOROBAN_TRANSACTION_QUEUE_SIZE_MULTIPLIER));
         }
     }
-    else if (mSorobanTransactionQueue)
+    else if (mTransactionQueues->hasSorobanTransactionQueue())
     {
         throw std::runtime_error(
             "Invalid state: Soroban queue initialized before v20");
@@ -2215,10 +2203,10 @@ HerderImpl::start()
     mTxQueueBucketSnapshot = mApp.getBucketManager()
                                  .getBucketSnapshotManager()
                                  .copySearchableLiveBucketListSnapshot();
-    releaseAssert(!mTransactionQueue);
-    mTransactionQueue = std::make_unique<ClassicTransactionQueue>(
-        mApp, mTxQueueBucketSnapshot, TRANSACTION_QUEUE_TIMEOUT_LEDGERS,
-        TRANSACTION_QUEUE_BAN_LEDGERS, TRANSACTION_QUEUE_SIZE_MULTIPLIER);
+    mTransactionQueues->setClassicTransactionQueue(
+        std::make_unique<ClassicTransactionQueue>(
+            mApp, mTxQueueBucketSnapshot, TRANSACTION_QUEUE_TIMEOUT_LEDGERS,
+            TRANSACTION_QUEUE_BAN_LEDGERS, TRANSACTION_QUEUE_SIZE_MULTIPLIER));
 
     mMaxTxSize = mApp.getHerder().getMaxClassicTxSize();
     {
@@ -2378,18 +2366,18 @@ HerderImpl::updateTransactionQueue(TxSetXDRFrameConstPtr externalizedTxSet)
         .maybeCopySearchableBucketListSnapshot(mTxQueueBucketSnapshot);
     if (txsPerPhase.size() > static_cast<size_t>(TxSetPhase::CLASSIC))
     {
-        mTransactionQueue->update(
+        mTransactionQueues->getClassicTransactionQueue().update(
             txsPerPhase[static_cast<size_t>(TxSetPhase::CLASSIC)], lhhe.header,
             mTxQueueBucketSnapshot, filterInvalidTxs);
     }
 
     // Even if we're in protocol 20, still check for number of phases, in case
     // we're dealing with the upgrade ledger that contains old-style transaction
     // set
-    if (mSorobanTransactionQueue != nullptr &&
+    if (mTransactionQueues->hasSorobanTransactionQueue() &&
         txsPerPhase.size() > static_cast<size_t>(TxSetPhase::SOROBAN))
     {
-        mSorobanTransactionQueue->update(
+        mTransactionQueues->getSorobanTransactionQueue().update(
             txsPerPhase[static_cast<size_t>(TxSetPhase::SOROBAN)], lhhe.header,
             mTxQueueBucketSnapshot, filterInvalidTxs);
     }
@@ -2508,37 +2496,34 @@ HerderImpl::isNewerNominationOrBallotSt(SCPStatement const& oldSt,
 size_t
 HerderImpl::getMaxQueueSizeOps() const
 {
-    return mTransactionQueue->getMaxQueueSizeOps();
+    return mTransactionQueues->getClassicTransactionQueue()
+        .getMaxQueueSizeOps();
 }
 
 size_t
 HerderImpl::getMaxQueueSizeSorobanOps() const
 {
-    return mSorobanTransactionQueue
-               ? mSorobanTransactionQueue->getMaxQueueSizeOps()
+    return mTransactionQueues->hasSorobanTransactionQueue()
+               ? mTransactionQueues->getSorobanTransactionQueue()
+                     .getMaxQueueSizeOps()
                : 0;
 }
 
 bool
 HerderImpl::isBannedTx(Hash const& hash) const
 {
-    auto banned = mTransactionQueue->isBanned(hash);
-    if (mSorobanTransactionQueue)
-    {
-        banned = banned || mSorobanTransactionQueue->isBanned(hash);
-    }
-    return banned;
+    return mTransactionQueues->isBanned(hash);
 }
 
 TransactionFrameBaseConstPtr
 HerderImpl::getTx(Hash const& hash) const
 {
-    auto classic = mTransactionQueue->getTx(hash);
-    if (!classic && mSorobanTransactionQueue)
-    {
-        return mSorobanTransactionQueue->getTx(hash);
-    }
-    return classic;
+    return mTransactionQueues->getTx(hash);
+}
+
+TransactionQueuesPtr HerderImpl::getTransactionQueues() const {
+    releaseAssert(mTransactionQueues);
+    return mTransactionQueues;
 }
 
 }

src/herder/HerderImpl.h

@@ -198,6 +198,8 @@ class HerderImpl : public Herder
 
     virtual void beginApply() override;
 
+    TransactionQueuesPtr getTransactionQueues() const override;
+
     void startTxSetGCTimer();
 
 #ifdef BUILD_TESTS
@@ -248,8 +250,11 @@ class HerderImpl : public Herder
     void purgeOldPersistedTxSets();
     void writeDebugTxSet(LedgerCloseData const& lcd);
 
-    std::unique_ptr<ClassicTransactionQueue> mTransactionQueue;
-    std::unique_ptr<SorobanTransactionQueue> mSorobanTransactionQueue;
+    // TODO: Need some way to get these queues
+    // TODO: Maybe something else should create this and pass it in somehow,
+    // either via Application or explicitly in the constructor for HerderImpl.
+    TransactionQueuesPtr const mTransactionQueues =
+        std::make_shared<TransactionQueues>();
 
     void updateTransactionQueue(TxSetXDRFrameConstPtr txSet);
     void maybeSetupSorobanQueue(uint32_t protocolVersion);