Asynchronously write batches to NuDB. (XRPLF#4503)

Author: mtrippled

src/ripple/nodestore/Types.h

@@ -36,8 +36,9 @@ enum {
     // This sets a limit on the maximum number of writes
     // in a batch. Actual usage can be twice this since
     // we have a new batch growing as we write the old.
+    // The main goal is to avoid delays while persisting the ledger.
     //
-    batchWriteLimitSize = 65536
+    batchWriteLimitSize = 262144
 };
 
 /** Return codes from Backend operations. */

src/ripple/nodestore/backend/NuDBFactory.cpp

@@ -20,6 +20,7 @@
 #include <ripple/basics/contract.h>
 #include <ripple/nodestore/Factory.h>
 #include <ripple/nodestore/Manager.h>
+#include <ripple/nodestore/impl/BatchWriter.h>
 #include <ripple/nodestore/impl/DecodedBlob.h>
 #include <ripple/nodestore/impl/EncodedBlob.h>
 #include <ripple/nodestore/impl/codec.h>
@@ -35,7 +36,7 @@
 namespace ripple {
 namespace NodeStore {
 
-class NuDBBackend : public Backend
+class NuDBBackend : public Backend, public BatchWriter::Callback
 {
 public:
     static constexpr std::uint64_t currentType = 1;
@@ -46,6 +47,7 @@ class NuDBBackend : public Backend
 
     beast::Journal const j_;
     size_t const keyBytes_;
+    BatchWriter batch_;
     std::size_t const burstSize_;
     std::string const name_;
     nudb::store db_;
@@ -60,6 +62,7 @@ class NuDBBackend : public Backend
         beast::Journal journal)
         : j_(journal)
         , keyBytes_(keyBytes)
+        , batch_(*this, scheduler)
         , burstSize_(burstSize)
         , name_(get(keyValues, "path"))
         , deletePath_(false)
@@ -79,6 +82,7 @@ class NuDBBackend : public Backend
         beast::Journal journal)
         : j_(journal)
         , keyBytes_(keyBytes)
+        , batch_(*this, scheduler)
         , burstSize_(burstSize)
         , name_(get(keyValues, "path"))
         , db_(context)
@@ -262,13 +266,7 @@ class NuDBBackend : public Backend
     void
     store(std::shared_ptr<NodeObject> const& no) override
     {
-        BatchWriteReport report;
-        report.writeCount = 1;
-        auto const start = std::chrono::steady_clock::now();
-        do_insert(no);
-        report.elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
-            std::chrono::steady_clock::now() - start);
-        scheduler_.onBatchWrite(report);
+        batch_.store(no);
     }
 
     void
@@ -329,7 +327,7 @@ class NuDBBackend : public Backend
     int
     getWriteLoad() override
     {
-        return 0;
+        return batch_.getWriteLoad();
     }
 
     void
@@ -357,6 +355,12 @@ class NuDBBackend : public Backend
             Throw<nudb::system_error>(ec);
     }
 
+    void
+    writeBatch(Batch const& batch) override
+    {
+        storeBatch(batch);
+    }
+
     int
     fdRequired() const override
     {