Merge pull request stellar#4291 from marta-lokhova/minor_improvements

Overlay improvements

Reviewed-by: SirTyson

src/database/Database.cpp

@@ -447,7 +447,7 @@ soci::session&
 Database::getSession()
 {
     // global session can only be used from the main thread
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     return mSession;
 }
 

src/herder/HerderImpl.cpp

@@ -1787,7 +1787,7 @@ HerderImpl::checkAndMaybeReanalyzeQuorumMap()
         mLastQuorumMapIntersectionState.mInterruptFlag = false;
         mLastQuorumMapIntersectionState.mCheckingQuorumMapHash = curr;
         auto& cfg = mApp.getConfig();
-        assertThreadIsMain();
+        releaseAssert(threadIsMain());
         auto seed = gRandomEngine();
         auto qic = QuorumIntersectionChecker::create(
             qmap, cfg, mLastQuorumMapIntersectionState.mInterruptFlag, seed);

src/ledger/LedgerTxn.cpp

@@ -2501,7 +2501,7 @@ LedgerTxnRoot::Impl::addChild(AbstractLedgerTxn& child, TransactionMode mode)
     {
         // Read-only transactions are only allowed on the main thread to ensure
         // we're not competing with writes
-        assertThreadIsMain();
+        releaseAssert(threadIsMain());
     }
 
     mChild = &child;

src/main/ApplicationImpl.cpp

@@ -916,7 +916,7 @@ std::string
 ApplicationImpl::manualClose(std::optional<uint32_t> const& manualLedgerSeq,
                              std::optional<TimePoint> const& manualCloseTime)
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
 
     // Manual close only makes sense for validating nodes
     if (!mConfig.NODE_IS_VALIDATOR)
@@ -1452,7 +1452,7 @@ ApplicationImpl::createDatabase()
 AbstractLedgerTxnParent&
 ApplicationImpl::getLedgerTxnRoot()
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     return mConfig.MODE_USES_IN_MEMORY_LEDGER ? *mNeverCommittingLedgerTxn
                                               : *mLedgerTxnRoot;
 }

src/main/ApplicationImpl.h

@@ -126,11 +126,6 @@ class ApplicationImpl : public Application
 
     virtual void resetDBForInMemoryMode() override;
 
-  protected:
-    std::unique_ptr<LedgerManager>
-        mLedgerManager;              // allow to change that for tests
-    std::unique_ptr<Herder> mHerder; // allow to change that for tests
-
   private:
     VirtualClock& mVirtualClock;
     Config mConfig;
@@ -152,6 +147,12 @@ class ApplicationImpl : public Application
     std::unique_ptr<BucketManager> mBucketManager;
     std::unique_ptr<Database> mDatabase;
     std::unique_ptr<OverlayManager> mOverlayManager;
+
+  protected:
+    std::unique_ptr<LedgerManager>
+        mLedgerManager;              // allow to change that for tests
+    std::unique_ptr<Herder> mHerder; // allow to change that for tests
+  private:
     std::unique_ptr<CatchupManager> mCatchupManager;
     std::unique_ptr<HerderPersistence> mHerderPersistence;
     std::unique_ptr<HistoryArchiveManager> mHistoryArchiveManager;

src/overlay/FlowControl.cpp

@@ -43,6 +43,8 @@ FlowControl::FlowControl(Application& app)
 void
 FlowControl::sendSendMore(uint32_t numMessages, std::shared_ptr<Peer> peer)
 {
+    releaseAssert(threadIsMain());
+
     ZoneScoped;
     StellarMessage m;
     m.type(SEND_MORE);
@@ -55,6 +57,8 @@ void
 FlowControl::sendSendMore(uint32_t numMessages, uint32_t numBytes,
                           std::shared_ptr<Peer> peer)
 {
+    releaseAssert(threadIsMain());
+
     ZoneScoped;
     StellarMessage m;
     m.type(SEND_MORE_EXTENDED);
@@ -68,6 +72,7 @@ FlowControl::sendSendMore(uint32_t numMessages, uint32_t numBytes,
 bool
 FlowControl::hasOutboundCapacity(StellarMessage const& msg) const
 {
+    releaseAssert(threadIsMain());
     releaseAssert(mFlowControlCapacity);
     return mFlowControlCapacity->hasOutboundCapacity(msg) &&
            (!mFlowControlBytesCapacity ||
@@ -80,6 +85,8 @@ FlowControl::start(std::weak_ptr<Peer> peer,
                    std::function<void(StellarMessage const&)> sendCb,
                    bool enableFCBytes)
 {
+    releaseAssert(threadIsMain());
+
     auto peerPtr = peer.lock();
     if (!peerPtr)
     {
@@ -242,7 +249,7 @@ bool
 FlowControl::maybeSendMessage(std::shared_ptr<StellarMessage const> msg)
 {
     ZoneScoped;
-    if (mApp.getOverlayManager().isFloodMessage(*msg))
+    if (OverlayManager::isFloodMessage(*msg))
     {
         addMsgAndMaybeTrimQueue(msg);
         maybeSendNextBatch();
@@ -281,6 +288,7 @@ FlowControl::endMessageProcessing(StellarMessage const& msg,
                                   std::weak_ptr<Peer> peer)
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
 
     mFloodDataProcessed += mFlowControlCapacity->releaseLocalCapacity(msg);
     if (mFlowControlBytesCapacity)
@@ -339,6 +347,7 @@ bool
 FlowControl::isSendMoreValid(StellarMessage const& msg,
                              std::string& errorMsg) const
 {
+    releaseAssert(threadIsMain());
     bool sendMoreExtendedType =
         mFlowControlBytesCapacity && msg.type() == SEND_MORE_EXTENDED;
     bool sendMoreType = !mFlowControlBytesCapacity && msg.type() == SEND_MORE;
@@ -385,6 +394,7 @@ bool
 dropMessageAfterTimeout(FlowControl::QueuedOutboundMessage const& queuedMsg,
                         VirtualClock::time_point now)
 {
+    releaseAssert(threadIsMain());
     auto const& msg = *(queuedMsg.mMessage);
     bool dropType = msg.type() == TRANSACTION || msg.type() == FLOOD_ADVERT ||
                     msg.type() == FLOOD_DEMAND;
@@ -554,6 +564,8 @@ FlowControl::addMsgAndMaybeTrimQueue(std::shared_ptr<StellarMessage const> msg)
 Json::Value
 FlowControl::getFlowControlJsonInfo(bool compact) const
 {
+    releaseAssert(threadIsMain());
+
     Json::Value res;
     if (mFlowControlCapacity->getCapacity().mTotalCapacity)
     {
@@ -608,5 +620,6 @@ FlowControl::FlowControlMetrics::FlowControlMetrics()
                                               Peer::PEER_METRICS_RATE_UNIT,
                                               Peer::PEER_METRICS_WINDOW_SIZE))
 {
+    releaseAssert(threadIsMain());
 }
 }

src/overlay/FlowControlCapacity.cpp

@@ -133,7 +133,7 @@ void
 FlowControlCapacity::lockOutboundCapacity(StellarMessage const& msg)
 {
     ZoneScoped;
-    if (mApp.getOverlayManager().isFloodMessage(msg))
+    if (OverlayManager::isFloodMessage(msg))
     {
         releaseAssert(hasOutboundCapacity(msg));
         mOutboundCapacity -= getMsgResourceCount(msg);
@@ -152,7 +152,7 @@ FlowControlCapacity::lockLocalCapacity(StellarMessage const& msg)
         *mCapacity.mTotalCapacity -= msgResources;
     }
 
-    if (mApp.getOverlayManager().isFloodMessage(msg))
+    if (OverlayManager::isFloodMessage(msg))
     {
         // No capacity to process flood message
         if (mCapacity.mFloodCapacity < msgResources)
@@ -182,7 +182,7 @@ FlowControlCapacity::releaseLocalCapacity(StellarMessage const& msg)
         *mCapacity.mTotalCapacity += resourcesFreed;
     }
 
-    if (mApp.getOverlayManager().isFloodMessage(msg))
+    if (OverlayManager::isFloodMessage(msg))
     {
         if (mCapacity.mFloodCapacity == 0)
         {

src/overlay/OverlayManager.h

@@ -66,6 +66,7 @@ class OverlayManager
 
     // Drop all PeerRecords from the Database
     static void dropAll(Database& db);
+    static bool isFloodMessage(StellarMessage const& msg);
 
     // Flush all FloodGate and ItemFetcher state for ledgers older than
     // `ledgerSeq`.
@@ -141,8 +142,6 @@ class OverlayManager
     virtual bool isPossiblyPreferred(std::string const& ip) const = 0;
     virtual bool haveSpaceForConnection(std::string const& ip) const = 0;
 
-    virtual bool isFloodMessage(StellarMessage const& msg) = 0;
-
     // Return the current in-memory set of inbound pending peers.
     virtual std::vector<Peer::pointer> const&
     getInboundPendingPeers() const = 0;

src/overlay/OverlayManagerImpl.cpp

@@ -1071,7 +1071,7 @@ OverlayManagerImpl::isPreferred(Peer* peer) const
 }
 
 bool
-OverlayManagerImpl::isFloodMessage(StellarMessage const& msg)
+OverlayManager::isFloodMessage(StellarMessage const& msg)
 {
     return msg.type() == SCP_MESSAGE || msg.type() == TRANSACTION ||
            msg.type() == FLOOD_DEMAND || msg.type() == FLOOD_ADVERT;

src/overlay/OverlayManagerImpl.h

@@ -151,7 +151,6 @@ class OverlayManagerImpl : public OverlayManager
 
     bool acceptAuthenticatedPeer(Peer::pointer peer) override;
     bool isPreferred(Peer* peer) const override;
-    bool isFloodMessage(StellarMessage const& msg) override;
     std::vector<Peer::pointer> const& getInboundPendingPeers() const override;
     std::vector<Peer::pointer> const& getOutboundPendingPeers() const override;
     std::vector<Peer::pointer> getPendingPeers() const override;

src/overlay/PeerManager.cpp

@@ -172,6 +172,7 @@ PeerManager::removePeersWithManyFailures(size_t minNumFailures,
                                          PeerBareAddress const* address)
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
     try
     {
         auto& db = mApp.getDatabase();

src/overlay/TCPPeer.cpp

@@ -52,7 +52,7 @@ TCPPeer::initiate(Application& app, PeerBareAddress const& address)
     releaseAssert(address.getType() == PeerBareAddress::Type::IPv4);
 
     CLOG_DEBUG(Overlay, "TCPPeer:initiate to {}", address.toString());
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     auto socket = make_shared<SocketType>(app.getClock().getIOContext(), BUFSZ);
     auto result = make_shared<TCPPeer>(app, WE_CALLED_REMOTE, socket);
     result->mAddress = address;
@@ -90,7 +90,7 @@ TCPPeer::initiate(Application& app, PeerBareAddress const& address)
 TCPPeer::pointer
 TCPPeer::accept(Application& app, shared_ptr<TCPPeer::SocketType> socket)
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
 
     auto extractIP = [](shared_ptr<SocketType> socket) {
         std::string result;
@@ -152,7 +152,7 @@ TCPPeer::accept(Application& app, shared_ptr<TCPPeer::SocketType> socket)
 
 TCPPeer::~TCPPeer()
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     Peer::shutdown();
     if (mRole == REMOTE_CALLED_US)
     {
@@ -181,7 +181,7 @@ TCPPeer::sendMessage(xdr::msg_ptr&& xdrBytes)
         return;
     }
 
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
 
     TimestampedMessage msg;
     msg.mEnqueuedTime = mApp.getClock().now();
@@ -263,7 +263,7 @@ void
 TCPPeer::messageSender()
 {
     ZoneScoped;
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
 
     // if nothing to do, mark progress and return.
     if (mWriteQueue.empty())
@@ -372,7 +372,7 @@ TCPPeer::writeHandler(asio::error_code const& error,
                       size_t messages_transferred)
 {
     ZoneScoped;
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     mLastWrite = mApp.getClock().now();
 
     if (error)
@@ -473,7 +473,7 @@ TCPPeer::scheduleRead()
 
     releaseAssert(canRead());
 
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     if (shouldAbort())
     {
         return;
@@ -488,7 +488,7 @@ void
 TCPPeer::startRead()
 {
     ZoneScoped;
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     releaseAssert(canRead());
     if (shouldAbort())
     {
@@ -637,7 +637,7 @@ void
 TCPPeer::readHeaderHandler(asio::error_code const& error,
                            std::size_t bytes_transferred)
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     if (error)
     {
         noteErrorReadHeader(bytes_transferred, error);
@@ -669,7 +669,7 @@ TCPPeer::readBodyHandler(asio::error_code const& error,
                          std::size_t bytes_transferred,
                          std::size_t expected_length)
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
 
     if (error)
     {
@@ -706,7 +706,7 @@ void
 TCPPeer::recvMessage()
 {
     ZoneScoped;
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     releaseAssert(canRead());
 
     try
@@ -737,7 +737,7 @@ void
 TCPPeer::drop(std::string const& reason, DropDirection dropDirection,
               DropMode dropMode)
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
     if (shouldAbort())
     {
         return;

src/test/test.cpp

@@ -96,7 +96,7 @@ struct TestContextListener : Catch::TestEventListenerBase
     {
         if (gTestTxMetaMode != TestTxMetaMode::META_TEST_IGNORE)
         {
-            assertThreadIsMain();
+            releaseAssert(threadIsMain());
             releaseAssert(!sTestCtx.has_value());
             sTestCtx.emplace(testInfo);
         }
@@ -106,7 +106,7 @@ struct TestContextListener : Catch::TestEventListenerBase
     {
         if (gTestTxMetaMode != TestTxMetaMode::META_TEST_IGNORE)
         {
-            assertThreadIsMain();
+            releaseAssert(threadIsMain());
             releaseAssert(sTestCtx.has_value());
             releaseAssert(sSectCtx.empty());
             sTestCtx.reset();
@@ -117,7 +117,7 @@ struct TestContextListener : Catch::TestEventListenerBase
     {
         if (gTestTxMetaMode != TestTxMetaMode::META_TEST_IGNORE)
         {
-            assertThreadIsMain();
+            releaseAssert(threadIsMain());
             sSectCtx.emplace_back(sectionInfo);
         }
     }
@@ -126,7 +126,7 @@ struct TestContextListener : Catch::TestEventListenerBase
     {
         if (gTestTxMetaMode != TestTxMetaMode::META_TEST_IGNORE)
         {
-            assertThreadIsMain();
+            releaseAssert(threadIsMain());
             sSectCtx.pop_back();
         }
     }
@@ -584,7 +584,7 @@ logFatalAndThrow(std::string const& msg)
 static std::pair<stdfs::path, std::string>
 getCurrentTestContext()
 {
-    assertThreadIsMain();
+    releaseAssert(threadIsMain());
 
     releaseAssert(TestContextListener::sTestCtx.has_value());
     auto& tc = TestContextListener::sTestCtx.value();

src/transactions/TransactionFrame.cpp

@@ -83,6 +83,7 @@ TransactionFrame::TransactionFrame(Hash const& networkID,
 Hash const&
 TransactionFrame::getFullHash() const
 {
+    ZoneScoped;
     if (isZero(mFullHash))
     {
         mFullHash = xdrSha256(mEnvelope);
@@ -93,6 +94,7 @@ TransactionFrame::getFullHash() const
 Hash const&
 TransactionFrame::getContentsHash() const
 {
+    ZoneScoped;
 #ifdef _DEBUG
     // force recompute
     Hash oldHash;