Remove timing window from RootStoppable (RIPD-1392):

RootStoppable was using two separate flags to identify that it
was stopping.  LoadManager was being notified when one flag was
set, but checking the other flag (not yet set) to see if we were
stopping.  There is no strong motivation for two flags.  The
timing window is closed by removing one flag and moving around
a chunk of code.

src/ripple/core/Stoppable.h

@@ -44,7 +44,7 @@ class RootStoppable;
     provides a set of behaviors for ensuring that the start and stop of a
     composite application-style object is well defined.
 
-    Upon the initialization of the composite object these steps are peformed:
+    Upon the initialization of the composite object these steps are performed:
 
     1.  Construct sub-components.
 
@@ -339,14 +339,15 @@ private:
     /*  Notify a root stoppable and children to stop, without waiting.
         Has no effect if the stoppable was already notified.
 
+        Returns true on the first call to stopAsync(), false otherwise.
+
         Thread safety:
             Safe to call from any thread at any time.
     */
-    void stopAsync(beast::Journal j);
+    bool stopAsync(beast::Journal j);
 
     std::atomic<bool> m_prepared;
-    bool m_calledStop;
-    std::atomic<bool> m_calledStopAsync;
+    std::atomic<bool> m_calledStop;
     std::mutex m_;
     std::condition_variable c_;
 };
@@ -362,7 +363,7 @@ RootStoppable::alertable_sleep_for(
     std::unique_lock<std::mutex> lock(m_);
     if (m_calledStop)
         return true;
-    return c_.wait_for(lock, d, [this]{return m_calledStop;});
+    return c_.wait_for(lock, d, [this]{return m_calledStop.load();});
 }
 
 template <class Rep, class Period>

src/ripple/core/impl/Stoppable.cpp

@@ -161,13 +161,12 @@ RootStoppable::RootStoppable (char const* name)
     : Stoppable (name, *this)
     , m_prepared (false)
     , m_calledStop (false)
-    , m_calledStopAsync (false)
 {
 }
 
 bool RootStoppable::isStopping() const
 {
-    return m_calledStopAsync;
+    return m_calledStop;
 }
 
 void RootStoppable::prepare ()
@@ -191,25 +190,30 @@ void RootStoppable::stop (beast::Journal j)
     // Must have a prior call to start()
     assert (m_started);
 
-    {
-        std::lock_guard<std::mutex> lock(m_);
-        if (m_calledStop)
-        {
-            if (auto stream = j.warn())
-                stream << "Stoppable::stop called again";
-            return;
-        }
-        m_calledStop = true;
-        c_.notify_all();
-    }
-    stopAsync (j);
-    stopRecursive (j);
+    if (stopAsync (j))
+        stopRecursive (j);
 }
 
-void RootStoppable::stopAsync(beast::Journal j)
+bool RootStoppable::stopAsync(beast::Journal j)
 {
-    if (m_calledStopAsync.exchange (true) == false)
-        stopAsyncRecursive(j);
+    bool alreadyCalled;
+    {
+        // Even though m_calledStop is atomic, we change its value under a
+        // lock.  This removes a small timing window that occurs if the
+        // waiting thread is handling a spurious wakeup while m_calledStop
+        // changes state.
+        std::unique_lock<std::mutex> lock (m_);
+        alreadyCalled = m_calledStop.exchange (true);
+    }
+    if (alreadyCalled)
+    {
+        if (auto stream = j.warn())
+            stream << "Stoppable::stop called again";
+        return false;
+    }
+    c_.notify_all();
+    stopAsyncRecursive(j);
+    return true;
 }
 
 }

src/test/core/Stoppable_test.cpp

@@ -436,6 +436,13 @@ class Stoppable_test
             Stoppable::stopped();
             test_.expect(--test_.count == 0, "Root::onChildrenStopped called out of order");
         }
+
+        void secondStop()
+        {
+            // Calling stop() a second time should have no negative
+            // consequences.
+            stop({});
+        }
     };
 
 public:
@@ -444,6 +451,7 @@ public:
         {
             Root rt(*this);
             rt.run();
+            rt.secondStop();
         }
         pass();
     }