Revert "Asynchronously write batches to NuDB. (#4503)" (#4882)

This reverts commit 1d9db1bfdd.

This improves the stability of online deletion.

src/ripple/nodestore/Types.h

@@ -36,9 +36,8 @@ 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 = 262144
+    batchWriteLimitSize = 65536
 };
 
 /** Return codes from Backend operations. */

src/ripple/nodestore/backend/NuDBFactory.cpp

@@ -20,7 +20,6 @@
 #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>
@@ -36,7 +35,7 @@
 namespace ripple {
 namespace NodeStore {
 
-class NuDBBackend : public Backend, public BatchWriter::Callback
+class NuDBBackend : public Backend
 {
 public:
     static constexpr std::uint64_t currentType = 1;
@@ -47,7 +46,6 @@ public:
 
     beast::Journal const j_;
     size_t const keyBytes_;
-    BatchWriter batch_;
     std::size_t const burstSize_;
     std::string const name_;
     nudb::store db_;
@@ -62,7 +60,6 @@ public:
         beast::Journal journal)
         : j_(journal)
         , keyBytes_(keyBytes)
-        , batch_(*this, scheduler)
         , burstSize_(burstSize)
         , name_(get(keyValues, "path"))
         , deletePath_(false)
@@ -82,7 +79,6 @@ public:
         beast::Journal journal)
         : j_(journal)
         , keyBytes_(keyBytes)
-        , batch_(*this, scheduler)
         , burstSize_(burstSize)
         , name_(get(keyValues, "path"))
         , db_(context)
@@ -266,7 +262,13 @@ public:
     void
     store(std::shared_ptr<NodeObject> const& no) override
     {
-        batch_.store(no);
+        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);
     }
 
     void
@@ -327,7 +329,7 @@ public:
     int
     getWriteLoad() override
     {
-        return batch_.getWriteLoad();
+        return 0;
     }
 
     void
@@ -355,12 +357,6 @@ public:
             Throw<nudb::system_error>(ec);
     }
 
-    void
-    writeBatch(Batch const& batch) override
-    {
-        storeBatch(batch);
-    }
-
     int
     fdRequired() const override
     {