Added medium priority threads and cleanup

Author: SirTyson

src/bucket/BucketManagerImpl.cpp

@@ -972,8 +972,9 @@ BucketManagerImpl::startBackgroundEvictionScan(uint32_t ledgerSeq)
         });
 
     mEvictionFuture = task->get_future();
-    mApp.postOnBackgroundThread(bind(&task_t::operator(), task),
-                                "SearchableBucketListSnapshot: eviction scan");
+    mApp.postOnEvictionBackgroundThread(
+        bind(&task_t::operator(), task),
+        "SearchableBucketListSnapshot: eviction scan");
 }
 
 void
@@ -982,13 +983,13 @@ BucketManagerImpl::resolveBackgroundEvictionScan(
     LedgerKeySet const& modifiedKeys)
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
 
     if (!mEvictionFuture.valid())
     {
         startBackgroundEvictionScan(ledgerSeq);
     }
 
-    mEvictionFuture.wait();
     auto evictionCandidates = mEvictionFuture.get();
 
     auto const& networkConfig =
@@ -1000,7 +1001,6 @@ BucketManagerImpl::resolveBackgroundEvictionScan(
                                     networkConfig.stateArchivalSettings()))
     {
         startBackgroundEvictionScan(ledgerSeq);
-        mEvictionFuture.wait();
         evictionCandidates = mEvictionFuture.get();
     }
 

src/bucket/BucketManagerImpl.h

@@ -62,17 +62,6 @@ class BucketManagerImpl : public BucketManager
 
     std::future<EvictionResult> mEvictionFuture{};
 
-    // Lock for managing raw Bucket files or the bucket directory. This lock is
-    // only required for file access, but is not required for logical changes to
-    // the BucketList (i.e. addBatch).
-    mutable std::recursive_mutex mBucketFileMutex;
-
-    // Lock for logical BucketList changes and snapshots (i.e. addBatch,
-    // getSearchableSnapshot). This lock is not required for raw Bucket file
-    // management.
-    mutable std::recursive_mutex mBucketSnapshotMutex;
-
-
     bool const mDeleteEntireBucketDirInDtor;
 
     // Records bucket-merges that are currently _live_ in some FutureBucket, in

src/main/Application.h

@@ -226,12 +226,18 @@ class Application
     // this io_context will execute in parallel with the calling thread, so use
     // with caution.
     virtual asio::io_context& getWorkerIOContext() = 0;
+    virtual asio::io_context& getEvictionIOContext() = 0;
 
     virtual void postOnMainThread(
         std::function<void()>&& f, std::string&& name,
         Scheduler::ActionType type = Scheduler::ActionType::NORMAL_ACTION) = 0;
+
+    // While both are lower priority than the main thread, eviction threads have
+    // more priority than regular worker background threads
     virtual void postOnBackgroundThread(std::function<void()>&& f,
                                         std::string jobName) = 0;
+    virtual void postOnEvictionBackgroundThread(std::function<void()>&& f,
+                                                std::string jobName) = 0;
 
     // Perform actions necessary to transition from BOOTING_STATE to other
     // states. In particular: either reload or reinitialize the database, and

src/main/ApplicationImpl.cpp

@@ -81,9 +81,20 @@ namespace stellar
 ApplicationImpl::ApplicationImpl(VirtualClock& clock, Config const& cfg)
     : mVirtualClock(clock)
     , mConfig(cfg)
-    , mWorkerIOContext(mConfig.WORKER_THREADS)
+    // Allocate one worker to eviction when background eviction enabled
+    , mWorkerIOContext(mConfig.EXPERIMENTAL_BACKGROUND_EVICTION_SCAN
+                           ? mConfig.WORKER_THREADS - 1
+                           : mConfig.WORKER_THREADS)
+    , mEvictionIOContext(mConfig.EXPERIMENTAL_BACKGROUND_EVICTION_SCAN
+                             ? std::make_optional<asio::io_context>(1)
+                             : std::nullopt)
     , mWork(std::make_unique<asio::io_context::work>(mWorkerIOContext))
+    , mEvictionWork(
+          mEvictionIOContext
+              ? std::make_unique<asio::io_context::work>(*mEvictionIOContext)
+              : nullptr)
     , mWorkerThreads()
+    , mEvictionThread()
     , mStopSignals(clock.getIOContext(), SIGINT)
     , mStarted(false)
     , mStopping(false)
@@ -135,6 +146,21 @@ ApplicationImpl::ApplicationImpl(VirtualClock& clock, Config const& cfg)
 
     auto t = mConfig.WORKER_THREADS;
     LOG_DEBUG(DEFAULT_LOG, "Application constructing (worker threads: {})", t);
+
+    if (mConfig.EXPERIMENTAL_BACKGROUND_EVICTION_SCAN)
+    {
+        releaseAssert(mConfig.WORKER_THREADS > 0);
+        releaseAssert(mEvictionIOContext);
+
+        // Allocate one thread for Eviction scan
+        mEvictionThread = std::thread{[this]() {
+            runCurrentThreadWithMediumPriority();
+            mEvictionIOContext->run();
+        }};
+
+        --t;
+    }
+
     while (t--)
     {
         auto thread = std::thread{[this]() {
@@ -760,12 +786,21 @@ ApplicationImpl::validateAndLogConfig()
             "stellar-core new-db.");
     }
 
-    if (mConfig.EXPERIMENTAL_BACKGROUND_EVICTION_SCAN &&
-        !mConfig.isUsingBucketListDB())
+    if (mConfig.EXPERIMENTAL_BACKGROUND_EVICTION_SCAN)
     {
-        throw std::invalid_argument(
-            "EXPERIMENTAL_BUCKETLIST_DB must be enabled to use "
-            "EXPERIMENTAL_BACKGROUND_EVICTION_SCAN");
+        if (!mConfig.isUsingBucketListDB())
+        {
+            throw std::invalid_argument(
+                "EXPERIMENTAL_BUCKETLIST_DB must be enabled to use "
+                "EXPERIMENTAL_BACKGROUND_EVICTION_SCAN");
+        }
+
+        if (mConfig.WORKER_THREADS < 2)
+        {
+            throw std::invalid_argument(
+                "EXPERIMENTAL_BACKGROUND_EVICTION_SCAN requires "
+                "WORKER_THREADS > 1");
+        }
     }
 
     if (isNetworkedValidator && mConfig.isInMemoryMode())
@@ -917,6 +952,18 @@ ApplicationImpl::joinAllThreads()
     {
         w.join();
     }
+
+    if (mEvictionWork)
+    {
+        mEvictionWork.reset();
+    }
+
+    if (mEvictionThread)
+    {
+        LOG_DEBUG(DEFAULT_LOG, "Joining eviction thread");
+        mEvictionThread->join();
+    }
+
     LOG_DEBUG(DEFAULT_LOG, "Joined all {} threads", mWorkerThreads.size());
 }
 
@@ -1386,6 +1433,13 @@ ApplicationImpl::getWorkerIOContext()
     return mWorkerIOContext;
 }
 
+asio::io_context&
+ApplicationImpl::getEvictionIOContext()
+{
+    releaseAssert(mEvictionIOContext);
+    return *mEvictionIOContext;
+}
+
 void
 ApplicationImpl::postOnMainThread(std::function<void()>&& f, std::string&& name,
                                   Scheduler::ActionType type)
@@ -1418,6 +1472,18 @@ ApplicationImpl::postOnBackgroundThread(std::function<void()>&& f,
     });
 }
 
+void
+ApplicationImpl::postOnEvictionBackgroundThread(std::function<void()>&& f,
+                                                std::string jobName)
+{
+    LogSlowExecution isSlow{std::move(jobName), LogSlowExecution::Mode::MANUAL,
+                            "executed after"};
+    asio::post(getEvictionIOContext(), [this, f = std::move(f), isSlow]() {
+        mPostOnBackgroundThreadDelay.Update(isSlow.checkElapsedTime());
+        f();
+    });
+}
+
 void
 ApplicationImpl::enableInvariantsFromConfig()
 {

src/main/ApplicationImpl.h

@@ -78,10 +78,13 @@ class ApplicationImpl : public Application
     virtual StatusManager& getStatusManager() override;
 
     virtual asio::io_context& getWorkerIOContext() override;
+    virtual asio::io_context& getEvictionIOContext() override;
     virtual void postOnMainThread(std::function<void()>&& f, std::string&& name,
                                   Scheduler::ActionType type) override;
     virtual void postOnBackgroundThread(std::function<void()>&& f,
                                         std::string jobName) override;
+    virtual void postOnEvictionBackgroundThread(std::function<void()>&& f,
+                                                std::string jobName) override;
 
     virtual void start() override;
     void startServices();
@@ -142,7 +145,9 @@ class ApplicationImpl : public Application
     // subsystems.
 
     asio::io_context mWorkerIOContext;
+    std::optional<asio::io_context> mEvictionIOContext;
     std::unique_ptr<asio::io_context::work> mWork;
+    std::unique_ptr<asio::io_context::work> mEvictionWork;
 
     std::unique_ptr<BucketManager> mBucketManager;
     std::unique_ptr<Database> mDatabase;
@@ -188,6 +193,12 @@ class ApplicationImpl : public Application
 
     std::vector<std::thread> mWorkerThreads;
 
+    // Unlike mWorkerThreads (which are low priority), eviction scans require a
+    // medium priority thread. In the future, this may become a more general
+    // higher-priority worker thread type, but for now we only need a single
+    // thread for eviction scans.
+    std::optional<std::thread> mEvictionThread;
+
     asio::signal_set mStopSignals;
 
     bool mStarted;

src/main/Config.cpp

@@ -1331,7 +1331,7 @@ Config::processConfig(std::shared_ptr<cpptoml::table> t)
             }
             else if (item.first == "WORKER_THREADS")
             {
-                WORKER_THREADS = readInt<int>(item, 1, 1000);
+                WORKER_THREADS = readInt<int>(item, 2, 1000);
             }
             else if (item.first == "MAX_CONCURRENT_SUBPROCESSES")
             {

src/test/FuzzerImpl.cpp

@@ -867,7 +867,7 @@ getFuzzConfig(int instanceNumber)
     cfg.ARTIFICIALLY_GENERATE_LOAD_FOR_TESTING = false;
     cfg.ARTIFICIALLY_SET_CLOSE_TIME_FOR_TESTING = UINT32_MAX;
     cfg.HTTP_PORT = 0;
-    cfg.WORKER_THREADS = 1;
+    cfg.WORKER_THREADS = 2;
     cfg.QUORUM_INTERSECTION_CHECKER = false;
     cfg.PREFERRED_PEERS_ONLY = false;
     cfg.RUN_STANDALONE = true;

src/util/Thread.cpp

@@ -19,64 +19,101 @@ namespace stellar
 
 #if defined(_WIN32)
 
-void
-runCurrentThreadWithLowPriority()
+static void
+runCurrentThreadWithPriority(int priority)
 {
     HANDLE curThread = ::GetCurrentThread();
-    BOOL ret = ::SetThreadPriority(curThread, THREAD_PRIORITY_BELOW_NORMAL);
+    BOOL ret = ::SetThreadPriority(curThread, priority);
 
     if (!ret)
     {
         LOG_DEBUG(DEFAULT_LOG, "Unable to set priority for thread: {}", ret);
     }
 }
 
-#elif defined(__linux__)
-
 void
 runCurrentThreadWithLowPriority()
 {
-    constexpr auto const LOW_PRIORITY_NICE = 5;
+    runCurrentThreadWithPriority(THREAD_PRIORITY_LOWEST);
+}
+
+void
+runCurrentThreadWithMediumPriority()
+{
+    runCurrentThreadWithPriority(THREAD_PRIORITY_BELOW_NORMAL);
+}
 
-    auto newNice = nice(LOW_PRIORITY_NICE);
-    if (newNice != LOW_PRIORITY_NICE)
+#elif defined(__linux__)
+
+static void
+runCurrentThreadWithPriority(int priority)
+{
+    auto newNice = nice(priority);
+    if (newNice != priority)
     {
         LOG_DEBUG(DEFAULT_LOG, "Unable to run worker thread with low priority. "
                                "Normal priority will be used.");
     }
 }
 
-#elif defined(__APPLE__)
-
 void
 runCurrentThreadWithLowPriority()
+{
+    runCurrentThreadWithPriority(/*LOW_PRIORITY_NICE*/ 5);
+}
+
+void
+runCurrentThreadWithMediumPriority()
+{
+    runCurrentThreadWithPriority(/*MED_PRIORITY_NICE*/ 3);
+}
+
+#elif defined(__APPLE__)
+
+static void
+runCurrentThreadWithPriority(int priority)
 {
     // Default MacOS priority is 31 in a user-mode band from 0..63, niceing (or
     // other priority-adjustment) usually subtracts from there. Range is +/- 16,
     // with lower meaning lower (i.e. UTILITY class is 20). The standard
     // pthreads API works for adjusting a single thread's priority.
-    constexpr auto const LOW_PRIORITY_NICE = 5;
     struct sched_param sp;
     int policy;
     int ret = pthread_getschedparam(pthread_self(), &policy, &sp);
     if (ret != 0)
     {
         LOG_DEBUG(DEFAULT_LOG, "Unable to get priority for thread: {}", ret);
     }
-    sp.sched_priority -= LOW_PRIORITY_NICE;
+    sp.sched_priority -= priority;
     ret = pthread_setschedparam(pthread_self(), policy, &sp);
     if (ret != 0)
     {
         LOG_DEBUG(DEFAULT_LOG, "Unable to set priority for thread: {}", ret);
     }
 }
 
+void
+runCurrentThreadWithLowPriority()
+{
+    runCurrentThreadWithPriority(/*LOW_PRIORITY_NICE*/ 5);
+}
+
+void
+runCurrentThreadWithMediumPriority()
+{
+    runCurrentThreadWithPriority(/*MED_PRIORITY_NICE*/ 3);
+}
 #else
 
 void
 runCurrentThreadWithLowPriority()
 {
 }
 
+void
+runCurrentThreadWithMediumPriority()
+{
+}
+
 #endif
 }

src/util/Thread.h

@@ -12,6 +12,7 @@ namespace stellar
 {
 
 void runCurrentThreadWithLowPriority();
+void runCurrentThreadWithMediumPriority();
 
 template <typename T>
 bool