Improved performance monitoring of NodeStore backends:

* Add multi-sample interface to LoadMonitor
* Instrument fetch operations and completions for reporting in server_info

src/ripple_app/main/NodeStoreScheduler.cpp

@@ -58,4 +58,18 @@ void NodeStoreScheduler::doTask (NodeStore::Task& task, Job&)
         stopped();
 }
 
+void NodeStoreScheduler::onFetch (NodeStore::FetchReport const& report)
+{
+    if (report.wentToDisk)
+        m_jobQueue->addLoadEvents (
+            report.isAsync ? jtNS_ASYNC_READ : jtNS_SYNC_READ,
+                1, report.elapsed);
+}
+
+void NodeStoreScheduler::onBatchWrite (NodeStore::BatchWriteReport const& report)
+{
+    m_jobQueue->addLoadEvents (jtNS_WRITE,
+        report.writeCount, report.elapsed);
+}
+
 } // ripple

src/ripple_app/main/NodeStoreScheduler.h

@@ -38,6 +38,8 @@ public:
     void onStop ();
     void onChildrenStopped ();
     void scheduleTask (NodeStore::Task& task);
+    void onFetch (NodeStore::FetchReport const& report) override;
+    void onBatchWrite (NodeStore::BatchWriteReport const& report) override;
 
 private:
     void doTask (NodeStore::Task& task, Job&);

src/ripple_core/functional/Job.h

@@ -68,6 +68,11 @@ enum JobType
     jtHO_READ       ,
     jtHO_WRITE      ,
     jtGENERIC       ,   // Used just to measure time
+
+    // Node store monitoring
+    jtNS_SYNC_READ  ,
+    jtNS_ASYNC_READ ,
+    jtNS_WRITE      ,
 };
 
 class Job

src/ripple_core/functional/JobQueue.cpp

@@ -267,6 +267,14 @@ public:
             new LoadEvent (iter-> second.load (), name, true));
     }
 
+    void addLoadEvents (JobType t,
+        int count, std::chrono::milliseconds elapsed)
+    {
+        JobDataMap::iterator iter (m_jobData.find (t));
+        assert (iter != m_jobData.end ());
+        iter->second.load().addSamples (count, elapsed);
+    }
+
     bool isOverloaded ()
     {
         int count = 0;

src/ripple_core/functional/JobQueue.h

@@ -63,6 +63,10 @@ public:
     //          
     virtual LoadEvent::autoptr getLoadEventAP (JobType t, const std::string& name) = 0;
 
+    // Add multiple load events
+    virtual void addLoadEvents (JobType t,
+        int count, std::chrono::milliseconds elapsed) = 0;
+
     virtual bool isOverloaded () = 0;
 
     virtual Json::Value getJson (int c = 0) = 0;

src/ripple_core/functional/JobTypes.h

@@ -158,6 +158,14 @@ public:
 
         add (jtGENERIC,       "generic",
             0,        false,  true,  0,     0);
+
+        add (jtNS_SYNC_READ,  "SyncReadNode",
+            0,        false,  true,  0,     0);
+        add (jtNS_ASYNC_READ, "AsyncReadNode",
+            0,        false,  true,  0,     0);
+        add (jtNS_WRITE,      "WriteNode",
+            0,        false,  true,  0,     0);
+
     }
 
     JobTypeInfo const& get (JobType jt) const

src/ripple_core/functional/LoadMonitor.cpp

@@ -122,7 +122,7 @@ void LoadMonitor::addLatency (int latency)
     mLatencyMSAvg += latency;
     mLatencyMSPeak += latency;
 
-    // VFALCO NOTE Why are we multiplying by 4?
+    // Units are quarters of a millisecond
     int const latencyPeak = mLatencyEvents * latency * 4;
 
     if (mLatencyMSPeak < latencyPeak)
@@ -168,6 +168,26 @@ void LoadMonitor::addLoadSample (LoadEvent const& sample)
         mLatencyMSPeak = latencyPeak;
 }
 
+/* Add multiple samples
+   @param count The number of samples to add
+   @param latencyMS The total number of milliseconds
+*/
+void LoadMonitor::addSamples (int count, std::chrono::milliseconds latency)
+{
+    ScopedLockType sl (mLock);
+
+    update ();
+    mCounts += count;
+    mLatencyEvents += count;
+    mLatencyMSAvg += latency.count();
+    mLatencyMSPeak += latency.count();
+
+    int const latencyPeak = mLatencyEvents * latency.count() * 4 / count;
+
+    if (mLatencyMSPeak < latencyPeak)
+        mLatencyMSPeak = latencyPeak;
+}
+
 void LoadMonitor::setTargetLatency (std::uint64_t avg, std::uint64_t pk)
 {
     mTargetLatencyAvg  = avg;

src/ripple_core/functional/LoadMonitor.h

@@ -37,6 +37,8 @@ public:
 
     void addLoadSample (LoadEvent const& sample);
 
+    void addSamples (int count, std::chrono::milliseconds latency);
+
     void setTargetLatency (std::uint64_t avg, std::uint64_t pk);
 
     bool isOverTarget (std::uint64_t avg, std::uint64_t peak);

src/ripple_core/nodestore/api/DummyScheduler.h

@@ -31,6 +31,8 @@ public:
     ~DummyScheduler ();
     void scheduleTask (Task& task);
     void scheduledTasksStopped ();
+    void onFetch (FetchReport const& report) override;
+    void onBatchWrite (BatchWriteReport const& report) override;
 };
 
 }

src/ripple_core/nodestore/api/Scheduler.h

@@ -23,6 +23,22 @@
 namespace ripple {
 namespace NodeStore {
 
+/** Contains information about a fetch operation. */
+struct FetchReport
+{
+    std::chrono::milliseconds elapsed;
+    bool isAsync;
+    bool wentToDisk;
+    bool wasFound;
+};
+
+/** Contains information about a batch write operation. */
+struct BatchWriteReport
+{
+    std::chrono::milliseconds elapsed;
+    int writeCount;
+};
+
 /** Scheduling for asynchronous backend activity
     
     For improved performance, a backend has the option of performing writes
@@ -42,6 +58,16 @@ public:
         foreign thread.
     */
     virtual void scheduleTask (Task& task) = 0;
+
+    /** Reports completion of a fetch
+        Allows the scheduler to monitor the node store's performance
+    */
+    virtual void onFetch (FetchReport const& report) = 0;
+
+    /** Reports the completion of a batch write
+        Allows the scheduler to monitor the node store's performance
+    */
+    virtual void onBatchWrite (BatchWriteReport const& report) = 0;
 };
 
 }

src/ripple_core/nodestore/impl/BatchWriter.cpp

@@ -90,7 +90,16 @@ BatchWriter::writeBatch ()
 
         }
 
+        BatchWriteReport report;
+        report.writeCount = set.size();
+        auto const before = std::chrono::steady_clock::now();
+
         m_callback.writeBatch (set);
+
+        report.elapsed = std::chrono::duration_cast <std::chrono::milliseconds>
+            (std::chrono::steady_clock::now() - before);
+
+        m_scheduler.onBatchWrite (report);
     }
 }
 

src/ripple_core/nodestore/impl/DatabaseImp.h

@@ -24,6 +24,7 @@
 
 #include <thread>
 #include <condition_variable>
+#include <chrono>
 
 namespace ripple {
 namespace NodeStore {
@@ -135,7 +136,30 @@ public:
         return m_cache.getTargetSize() / asyncDivider;
     }
 
-    NodeObject::Ptr fetch (uint256 const& hash)
+    NodeObject::Ptr fetch (uint256 const& hash) override
+    {
+        return doTimedFetch (hash, false);
+    }
+
+    /** Perform a fetch and report the time it took */
+    NodeObject::Ptr doTimedFetch (uint256 const& hash, bool isAsync)
+    {
+        FetchReport report;
+        report.isAsync = isAsync;
+        report.wentToDisk = false;
+
+        auto const before = std::chrono::steady_clock::now();
+        NodeObject::Ptr ret = doFetch (hash, report);
+        report.elapsed = std::chrono::duration_cast <std::chrono::milliseconds>
+            (std::chrono::steady_clock::now() - before);
+
+        report.wasFound = (ret != nullptr);
+        m_scheduler.onFetch (report);
+
+        return ret;
+    }
+
+    NodeObject::Ptr doFetch (uint256 const& hash, FetchReport &report)
     {
         // See if the object already exists in the cache
         //
@@ -150,6 +174,7 @@ public:
         // Check the database(s).
 
         bool foundInFastBackend = false;
+        report.wentToDisk = true;
 
         // Check the fast backend database if we have one
         //
@@ -324,7 +349,7 @@ public:
             }
 
             // Perform the read
-            fetch (hash);
+            doTimedFetch (hash, true);
          }
      }
 

src/ripple_core/nodestore/impl/DummyScheduler.cpp

@@ -40,5 +40,15 @@ DummyScheduler::scheduledTasksStopped ()
 {
 }
 
+void
+DummyScheduler::onFetch (const FetchReport& report)
+{
+}
+
+void
+DummyScheduler::onBatchWrite (const BatchWriteReport& report)
+{
+}
+
 }
 }