WIP introduce scan_pwals thread_context for LevelDB/RocksDB Batch

src/limestone/datastore_snapshot.cpp

@@ -94,7 +94,11 @@ static int comp_twisted_key(const std::string_view& a, const std::string_view& b
 }
 
 [[maybe_unused]]
-static void insert_entry_or_update_to_max(sortdb_wrapper* sortdb, const log_entry& e) {
+static void insert_entry_or_update_to_max(sortdb_wrapper* sortdb, sortdb_wrapper::batch* b, const log_entry& e) {
+    (void)b;
+    // not using batch mode;
+    // 1. this mode is RMW; so need WBWI
+    // 2. this mode is works_with_multi_thread = false, so less need for batch
     bool need_write = true;
     // skip older entry than already inserted
     std::string value;
@@ -114,7 +118,7 @@ static void insert_entry_or_update_to_max(sortdb_wrapper* sortdb, const log_entr
 }
 
 [[maybe_unused]]
-static void insert_twisted_entry(sortdb_wrapper* sortdb, const log_entry& e) {
+static void insert_twisted_entry(sortdb_wrapper* sortdb, sortdb_wrapper::batch* b, const log_entry& e) {
     // key_sid: storage_id[8] key[*], value_etc: epoch[8]LE minor_version[8]LE value[*], type: type[1]
     // db_key: epoch[8]BE minor_version[8]BE storage_id[8] key[*], db_value: type[1] value[*]
     std::string db_key(write_version_size + e.key_sid().size(), '\0');
@@ -123,9 +127,24 @@ static void insert_twisted_entry(sortdb_wrapper* sortdb, const log_entry& e) {
     std::memcpy(&db_key[write_version_size], e.key_sid().data(), e.key_sid().size());
     std::string db_value(1, static_cast<char>(e.type()));
     db_value.append(e.value_etc().substr(write_version_size));
-    sortdb->put(db_key, db_value);
+    sortdb->batch_put(b, db_key, db_value);
 }
 
+class sortdb_batch_context : public dblog_scan::thread_context_base {
+    sortdb_wrapper* sw;
+public:
+    sortdb_batch_context(sortdb_wrapper* sw0) : sw(sw0) {}
+    void* thread_start() override {
+        return sw->batch_start();
+    }
+
+    void thread_end(void*p) override {
+        auto* b = (sortdb_wrapper::batch*)p;
+        sw->flush(b);
+        sw->batch_end(b);
+    }
+};
+
 static std::pair<epoch_id_type, sorting_context> create_sorted_from_wals(
     const boost::filesystem::path& from_dir,
     int num_worker,
@@ -146,11 +165,12 @@ static std::pair<epoch_id_type, sorting_context> create_sorted_from_wals(
     const auto add_entry_to_point = insert_entry_or_update_to_max;
     bool works_with_multi_thread = false;
 #endif
-    auto add_entry = [&sctx, &add_entry_to_point](const log_entry& e){
+    auto add_entry = [&sctx, &add_entry_to_point](void* thread_context, const log_entry& e){
+        (void)thread_context;
         switch (e.type()) {
         case log_entry::entry_type::normal_entry:
         case log_entry::entry_type::remove_entry:
-            add_entry_to_point(sctx.get_sortdb(), e);
+            add_entry_to_point(sctx.get_sortdb(), (sortdb_wrapper::batch*)thread_context, e);
             break;
         case log_entry::entry_type::clear_storage:
         case log_entry::entry_type::remove_storage: {  // remove_storage is treated as clear_storage
@@ -172,8 +192,9 @@ static std::pair<epoch_id_type, sorting_context> create_sorted_from_wals(
         num_worker = 1;
     }
     logscan.set_thread_num(num_worker);
+    sortdb_batch_context bc(sctx.get_sortdb());
     try {
-        epoch_id_type max_appeared_epoch = logscan.scan_pwal_files_throws(ld_epoch, add_entry);
+        epoch_id_type max_appeared_epoch = logscan.scan_pwal_files_throws(ld_epoch, add_entry, &bc);
         return {max_appeared_epoch, std::move(sctx)};
     } catch (std::runtime_error& e) {
         VLOG_LP(log_info) << "failed to scan pwal files: " << e.what();

src/limestone/dblog_scan.cpp

@@ -122,15 +122,16 @@ void dblog_scan::detach_wal_files(bool skip_empty_files) {
 }
 
 epoch_id_type dblog_scan::scan_pwal_files(  // NOLINT(readability-function-cognitive-complexity)
-        epoch_id_type ld_epoch, const std::function<void(log_entry&)>& add_entry,
+        epoch_id_type ld_epoch, const std::function<void(void*, log_entry&)>& add_entry,
+        thread_context_base* tcb,
         const error_report_func_t& report_error, dblog_scan::parse_error::code* max_parse_error_value) {
     std::atomic<epoch_id_type> max_appeared_epoch{ld_epoch};
     if (max_parse_error_value) { *max_parse_error_value = dblog_scan::parse_error::failed; }
     std::atomic<dblog_scan::parse_error::code> max_error_value{dblog_scan::parse_error::code::ok};
-    auto process_file = [&](const boost::filesystem::path& p) {  // NOLINT(readability-function-cognitive-complexity)
+    auto process_file = [&](void* tc, const boost::filesystem::path& p) {  // NOLINT(readability-function-cognitive-complexity)
         if (is_wal(p)) {
             parse_error ec;
-            auto rc = scan_one_pwal_file(p, ld_epoch, add_entry, report_error, ec);
+            auto rc = scan_one_pwal_file(p, ld_epoch, add_entry, report_error, ec, tc);
             epoch_id_type max_epoch_of_file = rc;
             auto ec_value = ec.value();
             switch (ec_value) {
@@ -193,6 +194,8 @@ epoch_id_type dblog_scan::scan_pwal_files(  // NOLINT(readability-function-cogni
     workers.reserve(thread_num_);
     for (int i = 0; i < thread_num_; i++) {
         workers.emplace_back(std::thread([&](){
+            void * tc = nullptr;
+            if (tcb) { tc = tcb->thread_start(); }
             for (;;) {
                 boost::filesystem::path p;
                 {
@@ -203,7 +206,7 @@ epoch_id_type dblog_scan::scan_pwal_files(  // NOLINT(readability-function-cogni
                 }
 
                 try {
-                    process_file(p);
+                    process_file(tc, p);
                 } catch (std::runtime_error& ex) {
                     VLOG(log_info) << "/:limestone catch runtime_error(" << ex.what() << ")";
                     {
@@ -216,6 +219,7 @@ epoch_id_type dblog_scan::scan_pwal_files(  // NOLINT(readability-function-cogni
                     break;
                 }
             }
+            if (tcb) { tcb->thread_end(tc); }
         }));
     }
     for (int i = 0; i < thread_num_; i++) {
@@ -230,12 +234,12 @@ epoch_id_type dblog_scan::scan_pwal_files(  // NOLINT(readability-function-cogni
 
 // called from datastore::create_snapshot
 // db_startup mode
-epoch_id_type dblog_scan::scan_pwal_files_throws(epoch_id_type ld_epoch, const std::function<void(log_entry&)>& add_entry) {
+epoch_id_type dblog_scan::scan_pwal_files_throws(epoch_id_type ld_epoch, const std::function<void(void*, log_entry&)>& add_entry, thread_context_base* tcb) {
     set_fail_fast(true);
     set_process_at_nondurable_epoch_snippet(process_at_nondurable::repair_by_mark);
     set_process_at_truncated_epoch_snippet(process_at_truncated::report);
     set_process_at_damaged_epoch_snippet(process_at_damaged::report);
-    return scan_pwal_files(ld_epoch, add_entry, log_error_and_throw);
+    return scan_pwal_files(ld_epoch, add_entry, tcb, log_error_and_throw);
 }
 
 void dblog_scan::rescan_directory_paths() {

src/limestone/dblog_scan.h

@@ -163,21 +163,29 @@ class dblog_scan {
 
     epoch_id_type last_durable_epoch_in_dir();
 
+class thread_context_base {
+public:
+    virtual void* thread_start() = 0;
+    virtual void thread_end(void*) = 0;
+};
+
     /**
      * @returns max epoch value in directory
      * @throws exception on error
      */
-    epoch_id_type scan_pwal_files_throws(epoch_id_type ld_epoch, const std::function<void(log_entry&)>& add_entry);
+    epoch_id_type scan_pwal_files_throws(epoch_id_type ld_epoch, const std::function<void(void*, log_entry&)>& add_entry, thread_context_base *tcb = nullptr);
+
     /**
      * @returns max epoch value in directory
      */
-    epoch_id_type scan_pwal_files(epoch_id_type ld_epoch, const std::function<void(log_entry&)>& add_entry,
+    epoch_id_type scan_pwal_files(epoch_id_type ld_epoch, const std::function<void(void*, log_entry&)>& add_entry,
+        thread_context_base* tcb,
         const error_report_func_t& report_error, dblog_scan::parse_error::code* max_parse_error_value = nullptr);
 
     epoch_id_type scan_one_pwal_file(const boost::filesystem::path& p, epoch_id_type ld_epoch,
-        const std::function<void(log_entry&)>& add_entry,
+        const std::function<void(void*, log_entry&)>& add_entry,
         const error_report_func_t& report_error,
-        parse_error& pe);
+        parse_error& pe, void*);
 
     static bool is_wal(const boost::filesystem::path& p) { return p.filename().string().rfind(log_channel_prefix, 0) == 0; }
     static bool is_detached_wal(const boost::filesystem::path& p) {

src/limestone/dblogutil/dblogutil.cpp

@@ -75,7 +75,7 @@ void inspect(dblog_scan &ds, std::optional<epoch_id_type> epoch) {
     ds.set_process_at_damaged_epoch_snippet(dblog_scan::process_at_damaged::report);
     ds.set_fail_fast(false);
     dblog_scan::parse_error::code max_ec{};
-    epoch_id_type max_appeared_epoch = ds.scan_pwal_files(epoch.value_or(ld_epoch), [&](log_entry& e){
+    epoch_id_type max_appeared_epoch = ds.scan_pwal_files(epoch.value_or(ld_epoch), [&](void*, log_entry& e){
         if (e.type() == log_entry::entry_type::normal_entry) {
             VLOG(50) << "normal";
             count_normal_entry++;
@@ -85,7 +85,7 @@ void inspect(dblog_scan &ds, std::optional<epoch_id_type> epoch) {
         } else {
             LOG(ERROR) << static_cast<int>(e.type());
         }
-    }, [](log_entry::read_error& ec){
+    }, nullptr, [](log_entry::read_error& ec){
         VLOG(30) << "ERROR " << ec.value() << " : " << ec.message();
         return false;
     }, &max_ec);
@@ -136,7 +136,7 @@ void repair(dblog_scan &ds, std::optional<epoch_id_type> epoch) {
     ds.rescan_directory_paths();
     std::atomic_size_t count_wal_entry = 0;
     dblog_scan::parse_error::code max_ec{};
-    ds.scan_pwal_files(ld_epoch, [&count_wal_entry](log_entry&){ count_wal_entry++; }, [](log_entry::read_error& e) -> bool {
+    ds.scan_pwal_files(ld_epoch, [&count_wal_entry](void*, log_entry&){ count_wal_entry++; }, nullptr, [](log_entry::read_error& e) -> bool {
         LOG_LP(ERROR) << "this pwal file is broken: " << e.message();
         return false;
     }, &max_ec);

src/limestone/parse_wal_file.cpp

@@ -206,9 +206,9 @@ void invalidate_epoch_snippet(boost::filesystem::fstream& strm, std::streampos f
 // scan the file, and check max epoch number in this file
 epoch_id_type dblog_scan::scan_one_pwal_file(  // NOLINT(readability-function-cognitive-complexity)
         const boost::filesystem::path& p, epoch_id_type ld_epoch,
-        const std::function<void(log_entry&)>& add_entry,
+        const std::function<void(void*, log_entry&)>& add_entry,
         const error_report_func_t& report_error,
-        parse_error& pe) {
+        parse_error& pe, void* tc) {
     VLOG_LP(log_info) << "processing pwal file: " << p.filename().string();
     epoch_id_type current_epoch{UINT64_MAX};
     epoch_id_type max_epoch_of_file{0};
@@ -251,7 +251,7 @@ epoch_id_type dblog_scan::scan_one_pwal_file(  // NOLINT(readability-function-co
 // normal_entry | remove_entry | clear_storage | add_storage | remove_storage : (not 1st) { if (valid) process-entry } -> loop
             if (!first) {
                 if (valid) {
-                    add_entry(e);
+                    add_entry(tc, e);
                 }
             } else {
                 err_unexpected();

src/limestone/sortdb_wrapper.h

@@ -85,6 +85,44 @@ class sortdb_wrapper {
         throw std::runtime_error("error in sortdb put");
     }
 
+    using batch = WriteBatch;
+    batch* batch_start() {
+        return new WriteBatch();
+    }
+
+    void batch_end(batch* batch) {
+        if (batch->Count() != 0) {
+            LOG_LP(ERROR) << "programming error, unflushed batch count = " << batch->Count();
+            throw std::runtime_error("error in sortdb batch_end");
+        }
+        delete batch;
+    }
+
+    void batch_put(batch* batch, const std::string& key, const std::string& value) {
+        auto status = batch->Put(key, value);
+        if (!status.ok()) {
+            LOG_LP(ERROR) << "sortdb batch put error, status: " << status.ToString();
+            throw std::runtime_error("error in sortdb batch put");
+        }
+        if (batch->Count() >= 256) {
+            flush(batch);
+        }
+    }
+
+    void flush(batch* batch) {
+        if (batch->Count() != 0) {
+            LOG_LP(INFO) << batch->Count();
+            WriteOptions write_options{};
+            auto status = sortdb_->Write(write_options, batch);
+            if (!status.ok()) {
+                LOG_LP(ERROR) << "sortdb flush error, status: " << status.ToString();
+                throw std::runtime_error("error in sortdb flush");
+            }
+            LOG_LP(INFO) << batch->Count();
+            batch->Clear();
+        }
+    }
+
     bool get(const std::string& key, std::string* value) {
         ReadOptions read_options{};
         auto status = sortdb_->Get(read_options, key, value);

test/limestone/log/dblog_scan_test.cpp

@@ -76,13 +76,13 @@ static constexpr const char* location = "/tmp/dblog_scan_test";
         dblog_scan::parse_error pe;
         std::vector<log_entry::read_error> errors;
 
-        epoch_id_type max_epoch = ds.scan_one_pwal_file(p, 0x100, [](const log_entry& e){
+        epoch_id_type max_epoch = ds.scan_one_pwal_file(p, 0x100, [](void*, const log_entry& e){
             VLOG(30) << static_cast<int>(e.type());
         }, [&errors](const log_entry::read_error& re){
             VLOG(30) << re.message();
             errors.emplace_back(re);
             return false;
-        }, pe);
+        }, pe, 0);
 
         check(p, max_epoch, errors, pe);
         EXPECT_EQ(boost::filesystem::file_size(p), data.size());  // no size change
@@ -99,13 +99,13 @@ static constexpr const char* location = "/tmp/dblog_scan_test";
         dblog_scan::parse_error pe;
         std::vector<log_entry::read_error> errors;
 
-        epoch_id_type max_epoch = ds.scan_one_pwal_file(p, 0x100, [](const log_entry& e){
+        epoch_id_type max_epoch = ds.scan_one_pwal_file(p, 0x100, [](void*, const log_entry& e){
             VLOG(30) << static_cast<int>(e.type());
         }, [&errors](const log_entry::read_error& re){
             VLOG(30) << re.message();
             errors.emplace_back(re);
             return false;
-        }, pe);
+        }, pe, 0);
 
         check(p, max_epoch, errors, pe);
         EXPECT_EQ(boost::filesystem::file_size(p), data.size());  // no size change
@@ -122,13 +122,13 @@ static constexpr const char* location = "/tmp/dblog_scan_test";
         dblog_scan::parse_error pe;
         std::vector<log_entry::read_error> errors;
 
-        epoch_id_type max_epoch = ds.scan_one_pwal_file(p, 0x100, [](const log_entry& e){
+        epoch_id_type max_epoch = ds.scan_one_pwal_file(p, 0x100, [](void*, const log_entry& e){
             VLOG(30) << static_cast<int>(e.type());
         }, [&errors](const log_entry::read_error& re){
             VLOG(30) << re.message();
             errors.emplace_back(re);
             return false;
-        }, pe);
+        }, pe, 0);
 
         check(p, max_epoch, errors, pe);
     }

test/limestone/log/durable_test.cpp

@@ -164,10 +164,10 @@ std::tuple<limestone::api::log_entry::read_error, limestone::api::epoch_id_type,
     std::vector<log_entry> entries;
     limestone::api::epoch_id_type durable_epoch = UINT64_MAX;
     try {
-        durable_epoch = ds.scan_one_pwal_file(p, ld_epoch, [&entries](log_entry& e){ entries.emplace_back(e); }, [&err](log_entry::read_error& e) -> bool {
+        durable_epoch = ds.scan_one_pwal_file(p, ld_epoch, [&entries](void*, log_entry& e){ entries.emplace_back(e); }, [&err](log_entry::read_error& e) -> bool {
             err = e;
             throw std::runtime_error("pwal file read error");
-        }, ec);
+        }, ec, 0);
     } catch (std::runtime_error& re) {
     }
     return {err, durable_epoch, std::move(entries)};

test/limestone/log/log_dir_test.cpp

@@ -52,7 +52,7 @@ const boost::filesystem::path compaction_catalog_path = boost::filesystem::path(
 
     static bool starts_with(std::string a, std::string b) { return a.substr(0, b.length()) == b; }
     static bool is_pwal(const boost::filesystem::path& p) { return starts_with(p.filename().string(), "pwal"); }
-    static void ignore_entry(limestone::api::log_entry&) {}
+    static void ignore_entry(void*, limestone::api::log_entry&) {}
 
     void create_mainfest_file(int persistent_format_version = 1) {
         create_file(manifest_path, data_manifest(persistent_format_version));
@@ -281,7 +281,7 @@ TEST_F(log_dir_test, scan_pwal_files_in_dir_returns_max_epoch_nondurable) {
     // EXPECT_EQ(limestone::internal::scan_pwal_files_in_dir(location, 2, is_pwal, 0x100, ignore_entry), 0x101);
     limestone::internal::dblog_scan ds{boost::filesystem::path(location)};
     ds.set_thread_num(2);
-    EXPECT_EQ(ds.scan_pwal_files(0x100, ignore_entry, [](limestone::api::log_entry::read_error&){return false;}), 0x101);
+    EXPECT_EQ(ds.scan_pwal_files(0x100, ignore_entry, nullptr, [](limestone::api::log_entry::read_error&){return false;}), 0x101);
 }
 
 TEST_F(log_dir_test, scan_pwal_files_in_dir_rejects_unexpected_EOF) {