Redesign stoppable objects

src/ripple/app/ledger/LedgerMaster.h

@@ -31,10 +31,10 @@
 #include <ripple/app/misc/CanonicalTXSet.h>
 #include <ripple/basics/RangeSet.h>
 #include <ripple/basics/StringUtilities.h>
+#include <ripple/basics/UptimeClock.h>
 #include <ripple/basics/chrono.h>
 #include <ripple/beast/insight/Collector.h>
 #include <ripple/beast/utility/PropertyStream.h>
-#include <ripple/core/Stoppable.h>
 #include <ripple/protocol/Protocol.h>
 #include <ripple/protocol/RippleLedgerHash.h>
 #include <ripple/protocol/STValidation.h>
@@ -69,7 +69,7 @@ class ReportingShouldProxy : public std::runtime_error
 // Tracks the current ledger and any ledgers in the process of closing
 // Tracks ledger history
 // Tracks held transactions
-class LedgerMaster : public Stoppable, public AbstractFetchPackContainer
+class LedgerMaster : public AbstractFetchPackContainer
 {
 public:
     // Age for last validated ledger if the process has yet to validate.

src/ripple/app/ledger/OrderBookDB.cpp

@@ -27,11 +27,8 @@
 
 namespace ripple {
 
-OrderBookDB::OrderBookDB(Application& app, Stoppable& parent)
-    : Stoppable("OrderBookDB", parent)
-    , app_(app)
-    , mSeq(0)
-    , j_(app.journal("OrderBookDB"))
+OrderBookDB::OrderBookDB(Application& app)
+    : app_(app), mSeq(0), j_(app.journal("OrderBookDB"))
 {
 }
 
@@ -101,7 +98,7 @@ OrderBookDB::update(std::shared_ptr<ReadView const> const& ledger)
     {
         for (auto& sle : ledger->sles)
         {
-            if (isStopping())
+            if (app_.isStopping())
             {
                 JLOG(j_.info())
                     << "OrderBookDB::update exiting due to isStopping";

src/ripple/app/ledger/OrderBookDB.h

@@ -28,10 +28,10 @@
 
 namespace ripple {
 
-class OrderBookDB : public Stoppable
+class OrderBookDB
 {
 public:
-    OrderBookDB(Application& app, Stoppable& parent);
+    OrderBookDB(Application& app);
 
     void
     setup(std::shared_ptr<ReadView const> const& ledger);

src/ripple/app/ledger/impl/LedgerCleaner.cpp

@@ -51,8 +51,8 @@ class LedgerCleanerImp : public LedgerCleaner
 
     std::thread thread_;
 
-    enum class State : char { readyToClean = 0, startCleaning, cleaning };
-    State state_ = State::readyToClean;
+    enum class State : char { notCleaning = 0, cleaning };
+    State state_ = State::notCleaning;
     bool shouldExit_ = false;
 
     // The lowest ledger in the range we're checking.
@@ -108,6 +108,7 @@ class LedgerCleanerImp : public LedgerCleaner
             wakeup_.notify_one();
         }
         thread_.join();
+        stopped();
     }
 
     //--------------------------------------------------------------------------
@@ -214,11 +215,8 @@ class LedgerCleanerImp : public LedgerCleaner
             if (params.isMember(jss::stop) && params[jss::stop].asBool())
                 minRange_ = maxRange_ = 0;
 
-            if (state_ == State::readyToClean)
-            {
-                state_ = State::startCleaning;
-                wakeup_.notify_one();
-            }
+            state_ = State::cleaning;
+            wakeup_.notify_one();
         }
     }
 
@@ -228,36 +226,26 @@ class LedgerCleanerImp : public LedgerCleaner
     //
     //--------------------------------------------------------------------------
 private:
-    void
-    init()
-    {
-        JLOG(j_.debug()) << "Initializing";
-    }
-
     void
     run()
     {
         beast::setCurrentThreadName("LedgerCleaner");
         JLOG(j_.debug()) << "Started";
 
-        init();
-
         while (true)
         {
             {
                 std::unique_lock<std::mutex> lock(mutex_);
+                state_ = State::notCleaning;
                 wakeup_.wait(lock, [this]() {
-                    return (shouldExit_ || state_ == State::startCleaning);
+                    return (shouldExit_ || state_ == State::cleaning);
                 });
                 if (shouldExit_)
                     break;
-
-                state_ = State::cleaning;
+                assert(state_ == State::cleaning);
             }
             doLedgerCleaner();
         }
-
-        stopped();
     }
 
     // VFALCO TODO This should return boost::optional<uint256>
@@ -413,12 +401,11 @@ class LedgerCleanerImp : public LedgerCleaner
             bool doNodes;
             bool doTxns;
 
-            while (app_.getFeeTrack().isLoadedLocal())
+            if (app_.getFeeTrack().isLoadedLocal())
             {
                 JLOG(j_.debug()) << "Waiting for load to subside";
                 std::this_thread::sleep_for(std::chrono::seconds(5));
-                if (shouldExit())
-                    return;
+                continue;
             }
 
             {
@@ -427,7 +414,6 @@ class LedgerCleanerImp : public LedgerCleaner
                     (minRange_ == 0))
                 {
                     minRange_ = maxRange_ = 0;
-                    state_ = State::readyToClean;
                     return;
                 }
                 ledgerIndex = maxRange_;

src/ripple/app/ledger/impl/LedgerMaster.cpp

@@ -184,12 +184,13 @@ LedgerMaster::LedgerMaster(
     Stoppable& parent,
     beast::insight::Collector::ptr const& collector,
     beast::Journal journal)
-    : Stoppable("LedgerMaster", parent)
-    , app_(app)
+    : app_(app)
     , m_journal(journal)
     , mLedgerHistory(collector, app)
-    , mLedgerCleaner(
-          detail::make_LedgerCleaner(app, *this, app_.journal("LedgerCleaner")))
+    , mLedgerCleaner(detail::make_LedgerCleaner(
+          app,
+          parent,
+          app_.journal("LedgerCleaner")))
     , standalone_(app_.config().standalone())
     , fetch_depth_(
           app_.getSHAMapStore().clampFetchDepth(app_.config().FETCH_DEPTH))
@@ -1557,7 +1558,7 @@ LedgerMaster::newPFWork(
     }
     // If we're stopping don't give callers the expectation that their
     // request will be fulfilled, even if it may be serviced.
-    return mPathFindThread > 0 && !isStopping();
+    return mPathFindThread > 0 && !app_.isStopping();
 }
 
 std::recursive_mutex&

src/ripple/app/main/Application.cpp

@@ -167,6 +167,9 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
 #ifdef RIPPLED_REPORTING
     std::shared_ptr<PgPool> pgPool_;
 #endif
+
+    std::unique_ptr<CollectorManager> m_collectorManager;
+    std::unique_ptr<JobQueue> m_jobQueue;
     NodeStoreScheduler m_nodeStoreScheduler;
     std::unique_ptr<SHAMapStore> m_shaMapStore;
     PendingSaves pendingSaves_;
@@ -175,15 +178,13 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
 
     // These are not Stoppable-derived
     NodeCache m_tempNodeCache;
-    std::unique_ptr<CollectorManager> m_collectorManager;
     CachedSLEs cachedSLEs_;
     std::pair<PublicKey, SecretKey> nodeIdentity_;
     ValidatorKeys const validatorKeys_;
 
     std::unique_ptr<Resource::Manager> m_resourceManager;
 
     // These are Stoppable-related
-    std::unique_ptr<JobQueue> m_jobQueue;
     std::unique_ptr<NodeStore::Database> m_nodeStore;
     NodeFamily nodeFamily_;
     std::unique_ptr<NodeStore::DatabaseShard> shardStore_;
@@ -288,7 +289,22 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
                                    : nullptr)
 #endif
 
-        , m_nodeStoreScheduler(*this)
+        , m_collectorManager(CollectorManager::New(
+              config_->section(SECTION_INSIGHT),
+              logs_->journal("Collector")))
+
+        // The JobQueue has to come pretty early since
+        // almost everything is a Stoppable child of the JobQueue.
+        //
+        , m_jobQueue(std::make_unique<JobQueue>(
+              m_collectorManager->group("jobq"),
+              *this,
+              logs_->journal("JobQueue"),
+              *logs_,
+              *perfLog_))
+
+        , m_nodeStoreScheduler(*m_jobQueue)
+
         , m_shaMapStore(make_SHAMapStore(
               *this,
               *this,
@@ -304,26 +320,13 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
               stopwatch(),
               logs_->journal("TaggedCache"))
 
-        , m_collectorManager(CollectorManager::New(
-              config_->section(SECTION_INSIGHT),
-              logs_->journal("Collector")))
         , cachedSLEs_(std::chrono::minutes(1), stopwatch())
         , validatorKeys_(*config_, m_journal)
 
         , m_resourceManager(Resource::make_Manager(
               m_collectorManager->collector(),
               logs_->journal("Resource")))
 
-        // The JobQueue has to come pretty early since
-        // almost everything is a Stoppable child of the JobQueue.
-        //
-        , m_jobQueue(std::make_unique<JobQueue>(
-              m_collectorManager->group("jobq"),
-              m_nodeStoreScheduler,
-              logs_->journal("JobQueue"),
-              *logs_,
-              *perfLog_))
-
         , m_nodeStore(m_shaMapStore->makeNodeStore("NodeStore.main", 4))
 
         , nodeFamily_(*this, *m_collectorManager)
@@ -336,7 +339,7 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
               4,
               logs_->journal("ShardStore")))
 
-        , m_orderBookDB(*this, *m_jobQueue)
+        , m_orderBookDB(*this)
 
         , m_pathRequests(std::make_unique<PathRequests>(
               *this,
@@ -457,7 +460,7 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
               std::chrono::milliseconds(100),
               get_io_service())
         , grpcServer_(std::make_unique<GRPCServer>(*this, *m_jobQueue))
-        , reportingETL_(std::make_unique<ReportingETL>(*this, *m_ledgerMaster))
+        , reportingETL_(std::make_unique<ReportingETL>(*this, *this))
     {
         add(m_resourceManager.get());
 
@@ -476,9 +479,6 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
         //  started in this constructor.
         //
 
-        // VFALCO HACK
-        m_nodeStoreScheduler.setJobQueue(*m_jobQueue);
-
         add(m_ledgerMaster->getPropertySource());
     }
 
@@ -498,6 +498,8 @@ class ApplicationImp : public Application, public RootStoppable, public BasicApp
     checkSigs() const override;
     void
     checkSigs(bool) override;
+    bool
+    isStopping() const override;
     int
     fdRequired() const override;
 
@@ -1768,6 +1770,12 @@ ApplicationImp::checkSigs(bool check)
     checkSigs_ = check;
 }
 
+bool
+ApplicationImp::isStopping() const
+{
+    return Stoppable::isStopping();
+}
+
 int
 ApplicationImp::fdRequired() const
 {

src/ripple/app/main/Application.h

@@ -135,6 +135,8 @@ class Application : public beast::PropertyStream::Source
     checkSigs() const = 0;
     virtual void
     checkSigs(bool) = 0;
+    virtual bool
+    isStopping() const = 0;
 
     //
     // ---

src/ripple/app/main/LoadManager.cpp

@@ -39,7 +39,6 @@ LoadManager::LoadManager(
     , journal_(journal)
     , deadLock_()
     , armed_(false)
-    , stop_(false)
 {
 }
 
@@ -89,13 +88,15 @@ LoadManager::onStart()
 void
 LoadManager::onStop()
 {
+    {
+        std::lock_guard<std::mutex> sl(mutex_);
+        stop_ = true;
+        // There is at most one thread waiting on this condition.
+        cv_.notify_all();
+    }
     if (thread_.joinable())
     {
         JLOG(journal_.debug()) << "Stopping";
-        {
-            std::lock_guard sl(mutex_);
-            stop_ = true;
-        }
         thread_.join();
     }
     stopped();
@@ -109,19 +110,22 @@ LoadManager::run()
     beast::setCurrentThreadName("LoadManager");
 
     using namespace std::chrono_literals;
-    using clock_type = std::chrono::system_clock;
+    using clock_type = std::chrono::steady_clock;
 
     auto t = clock_type::now();
-    bool stop = false;
 
-    while (!(stop || isStopping()))
+    while (true)
     {
         {
-            // Copy out shared data under a lock.  Use copies outside lock.
+            t += 1s;
             std::unique_lock<std::mutex> sl(mutex_);
+            if (cv_.wait_until(sl, t, [this] { return stop_; }))
+            {
+                break;
+            }
+            // Copy out shared data under a lock.  Use copies outside lock.
             auto const deadLock = deadLock_;
             auto const armed = armed_;
-            stop = stop_;
             sl.unlock();
 
             // Measure the amount of time we have been deadlocked, in seconds.
@@ -192,22 +196,7 @@ LoadManager::run()
             // subscribe in NetworkOPs or Application.
             app_.getOPs().reportFeeChange();
         }
-
-        t += 1s;
-        auto const duration = t - clock_type::now();
-
-        if ((duration < 0s) || (duration > 1s))
-        {
-            JLOG(journal_.warn()) << "time jump";
-            t = clock_type::now();
-        }
-        else
-        {
-            alertable_sleep_until(t);
-        }
     }
-
-    stopped();
 }
 
 //------------------------------------------------------------------------------