From ea2008086a14d204ff0ae1e3681a835ffb75e3bf Mon Sep 17 00:00:00 2001
From: Mark Travis <mtrippled@users.noreply.github.com>
Date: Mon, 11 Sep 2023 15:46:17 -0700
Subject: [PATCH] Asynchronously write batches to NuDB. (#4503)

---
 src/ripple/nodestore/Types.h                 |  3 ++-
 src/ripple/nodestore/backend/NuDBFactory.cpp | 22 ++++++++++++--------
 2 files changed, 15 insertions(+), 10 deletions(-)

diff --git a/src/ripple/nodestore/Types.h b/src/ripple/nodestore/Types.h
index 6d8583ed9d1..08bcac2d25c 100644
--- a/src/ripple/nodestore/Types.h
+++ b/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. */
diff --git a/src/ripple/nodestore/backend/NuDBFactory.cpp b/src/ripple/nodestore/backend/NuDBFactory.cpp
index 30b848e82af..74afe2ed7f1 100644
--- a/src/ripple/nodestore/backend/NuDBFactory.cpp
+++ b/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
     {