Some cleanup around separate tx queue thread

Author: bboston7

src/herder/HerderImpl.cpp

@@ -600,7 +600,7 @@ Herder::recvTransaction(TransactionQueuesPtr txQueues,
     // 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
+    // transaction to come in and conflict with the tx queue 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.

src/herder/HerderImpl.h

@@ -250,9 +250,6 @@ class HerderImpl : public Herder
     void purgeOldPersistedTxSets();
     void writeDebugTxSet(LedgerCloseData const& lcd);
 
-    // 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>();
 

src/herder/TransactionQueue.h

@@ -67,6 +67,8 @@ class Application;
 //     incoming SCP messages need to wait for tx queue additions to occur, which
 //     is bad.
 //         * My note: Try both approaches and benchmark
+//         * Follow up note: The priority locking scheme also helps to address
+//           this.
 
 enum class TxQueueAddResultCode
 {
@@ -172,6 +174,9 @@ class TransactionQueue
   protected:
     // TODO: Docs?
     // TODO: Move?
+    // TODO: It might be worth benchmarking this against the solution that does
+    // not use priority locking (just uses std::mutex). The added complexity of
+    // this may not be worth it.
     class TxQueueLock : NonMovableOrCopyable
     {
       public:

src/main/ApplicationImpl.cpp

@@ -199,7 +199,10 @@ ApplicationImpl::ApplicationImpl(VirtualClock& clock, Config const& cfg)
     if (mConfig.BACKGROUND_TX_QUEUE)
     {
         // TODO: Keep priority unchanged as tx queue processes time-sensitive
-        // tasks? Or should tx queue priority be downgraded?
+        // tasks? Or should tx queue priority be downgraded? The priority
+        // locking mechanism in TransactionQueue is designed to prevent tx queue
+        // from starving other work, so it may be fine to keep priority
+        // unchanged.
         mTxQueueThread = std::thread{[this]() { mTxQueueIOContext->run(); }};
         mThreadTypes[mTxQueueThread->get_id()] = ThreadType::TX_QUEUE;
     }

src/overlay/OverlayManagerImpl.cpp

@@ -1231,17 +1231,6 @@ OverlayManagerImpl::recvTransaction(StellarMessage const& msg,
 
         // add it to our current set
         // and make sure it is valid
-        // TODO: I think I could pull this one call into Peer::recvTransaction
-        // and move basically all of the rest of this function to a new function
-        // called something like "recordTransactionStats" or something. Then,
-        // Peer:recvTransaction would invoke HerderImpl::recvTransaction in the
-        // background, and then pass the result to the new function on the main
-        // thread. That way I don't have to make OverlayManagerImpl and its
-        // dependencies (Floodgate, Peer, TxDemandsManager, maybe more), or much
-        // of Peer thread safe. Note that the recordTransactionStats function
-        // would probably need to take a shared ptr to the message so that the
-        // message doesn't get deleted before the function is called. The lambda
-        // capture will need to copy this pointer in.
         auto addResult = mApp.getHerder().recvTransaction(transaction, false);
         recordAddTransactionStats(addResult, transaction->getFullHash(), peer,
                                   index);

src/overlay/Peer.cpp

@@ -878,119 +878,94 @@ Peer::recvAuthenticatedMessage(AuthenticatedMessage&& msg)
 {
     ZoneScoped;
     releaseAssert(!threadIsMain() || !useBackgroundThread());
+    RECURSIVE_LOCK_GUARD(mStateMutex, guard);
 
-    // TODO: Remove if I get rid of the special lock scoping vv
-    std::shared_ptr<CapacityTrackedMessage> msgTracker = nullptr;
-
-    // TODO: Move back if I git rid of lock scoping vv
-    Scheduler::ActionType type = Scheduler::ActionType::NORMAL_ACTION;
-    std::string queueName;
+    if (shouldAbort(guard))
     {
-        RECURSIVE_LOCK_GUARD(mStateMutex, guard);
-
-        if (shouldAbort(guard))
-        {
-            return false;
-        }
+        return false;
+    }
 
-        std::string errorMsg;
-        if (getState(guard) >= GOT_HELLO &&
-            msg.v0().message.type() != ERROR_MSG)
+    std::string errorMsg;
+    if (getState(guard) >= GOT_HELLO && msg.v0().message.type() != ERROR_MSG)
+    {
+        if (!mHmac.checkAuthenticatedMessage(msg, errorMsg))
         {
-            if (!mHmac.checkAuthenticatedMessage(msg, errorMsg))
+            if (!threadIsMain())
             {
-                if (!threadIsMain())
-                {
-                    mAppConnector.postOnMainThread(
-                        [self = shared_from_this(), errorMsg]() {
-                            self->sendErrorAndDrop(ERR_AUTH, errorMsg);
-                        },
-                        "Peer::sendErrorAndDrop");
-                }
-                else
-                {
-                    sendErrorAndDrop(ERR_AUTH, errorMsg);
-                }
-                return false;
+                mAppConnector.postOnMainThread(
+                    [self = shared_from_this(), errorMsg]() {
+                        self->sendErrorAndDrop(ERR_AUTH, errorMsg);
+                    },
+                    "Peer::sendErrorAndDrop");
+            }
+            else
+            {
+                sendErrorAndDrop(ERR_AUTH, errorMsg);
             }
+            return false;
         }
+    }
 
-        // NOTE: Additionally, we may use state snapshots to verify TRANSACTION
-        // type messages in the background.
-
-        // Start tracking capacity here, so read throttling is applied
-        // appropriately. Flow control might not be started at that time
-        msgTracker = std::make_shared<CapacityTrackedMessage>(
-            shared_from_this(), msg.v0().message);
-
-        std::string cat;
-
-        switch (msgTracker->getMessage().type())
-        {
-        case HELLO:
-        case AUTH:
-            cat = AUTH_ACTION_QUEUE;
-            break;
-        // control messages
-        case PEERS:
-        case ERROR_MSG:
-        case SEND_MORE:
-        case SEND_MORE_EXTENDED:
-            cat = "CTRL";
-            break;
-        // high volume flooding
-        case TRANSACTION:
-        case FLOOD_ADVERT:
-        case FLOOD_DEMAND:
-        {
-            cat = "TX";
-            type = Scheduler::ActionType::DROPPABLE_ACTION;
-            break;
-        }
+    // NOTE: Additionally, we may use state snapshots to verify TRANSACTION type
+    // messages in the background.
 
-        // consensus, inbound
-        case GET_TX_SET:
-        case GET_SCP_QUORUMSET:
-        case GET_SCP_STATE:
-            cat = "SCPQ";
-            type = Scheduler::ActionType::DROPPABLE_ACTION;
-            break;
+    // Start tracking capacity here, so read throttling is applied
+    // appropriately. Flow control might not be started at that time
+    auto msgTracker = std::make_shared<CapacityTrackedMessage>(
+        shared_from_this(), msg.v0().message);
 
-        // consensus, self
-        case DONT_HAVE:
-        case TX_SET:
-        case GENERALIZED_TX_SET:
-        case SCP_QUORUMSET:
-        case SCP_MESSAGE:
-            cat = "SCP";
-            break;
+    std::string cat;
+    Scheduler::ActionType type = Scheduler::ActionType::NORMAL_ACTION;
 
-        default:
-            cat = "MISC";
-        }
+    switch (msgTracker->getMessage().type())
+    {
+    case HELLO:
+    case AUTH:
+        cat = AUTH_ACTION_QUEUE;
+        break;
+    // control messages
+    case PEERS:
+    case ERROR_MSG:
+    case SEND_MORE:
+    case SEND_MORE_EXTENDED:
+        cat = "CTRL";
+        break;
+    // high volume flooding
+    case TRANSACTION:
+    case FLOOD_ADVERT:
+    case FLOOD_DEMAND:
+    {
+        cat = "TX";
+        type = Scheduler::ActionType::DROPPABLE_ACTION;
+        break;
+    }
 
-        // processing of incoming messages during authenticated must be
-        // in-order, so while not authenticated, place all messages onto
-        // AUTH_ACTION_QUEUE scheduler queue
-        queueName = isAuthenticated(guard) ? cat : AUTH_ACTION_QUEUE;
-        type = isAuthenticated(guard) ? type
-                                      : Scheduler::ActionType::NORMAL_ACTION;
+    // consensus, inbound
+    case GET_TX_SET:
+    case GET_SCP_QUORUMSET:
+    case GET_SCP_STATE:
+        cat = "SCPQ";
+        type = Scheduler::ActionType::DROPPABLE_ACTION;
+        break;
 
-        // TODO: This scope (ending here) exists to ensure this doesn't hold the
-        // state lock upon entry to the transaction queue. This can cause
-        // deadlocks! I think it's safe to release the lock here as there's no
-        // longer any state querying. In practice though, if I end up posting
-        // the tryAdd action onto some tx-queue specific thread, then I can
-        // remove the scoping I added here and the lock will be released upon
-        // return from this function (like it always has).
+    // consensus, self
+    case DONT_HAVE:
+    case TX_SET:
+    case GENERALIZED_TX_SET:
+    case SCP_QUORUMSET:
+    case SCP_MESSAGE:
+        cat = "SCP";
+        break;
 
-        // TODO: Really investigate whether this peer+transaction queue locking
-        // each other issue can come up anywhere else.
+    default:
+        cat = "MISC";
     }
 
-    // TODO: vv Remove asserts if I get rid of the scoping above
-    releaseAssert(msgTracker);
-    releaseAssert(!queueName.empty());
+    // processing of incoming messages during authenticated must be in-order, so
+    // while not authenticated, place all messages onto AUTH_ACTION_QUEUE
+    // scheduler queue
+    auto queueName = isAuthenticated(guard) ? cat : AUTH_ACTION_QUEUE;
+    type = isAuthenticated(guard) ? type : Scheduler::ActionType::NORMAL_ACTION;
 
     // If a message is already scheduled, drop
     if (mAppConnector.checkScheduledAndCache(msgTracker))
@@ -1027,9 +1002,6 @@ Peer::recvAuthenticatedMessage(AuthenticatedMessage&& msg)
                 self->recvMessage(t);
             },
             "Peer::recvMessage"); // TODO: Change message to something better
-        // TODO: If I end up running this on a different thread then I need to
-        // be sure to std::move `msgTracker` into the lambda as-per the note
-        // below.
     }
     else
     {

src/overlay/TCPPeer.cpp

@@ -528,14 +528,10 @@ TCPPeer::startRead()
     ZoneScoped;
     releaseAssert(!threadIsMain() || !useBackgroundThread());
     releaseAssert(canRead());
+    RECURSIVE_LOCK_GUARD(mStateMutex, guard);
+    if (shouldAbort(guard))
     {
-        RECURSIVE_LOCK_GUARD(mStateMutex, guard);
-        if (shouldAbort(guard))
-        {
-            return;
-        }
-        // TODO: Remove this outer scoping if I add separate thread for bg tx
-        // queue
+        return;
     }
 
     mThreadVars.getIncomingHeader().clear();