Align chunk start offset to 4k and logstore inline flush (eBay#838)

Align the chunk start offset to 4k to fix performance degradation
of 4kb writes from 150us to 20us on p5gx. Because physical block
size is 4kb on nvme on p5gx. Earlier the chunk start offset was aligned
to 512. This alignment will only affect during first time boot.
Existing nodes during upgrade wont have issues as they use old format.
Add logstore inline flush mode for solo repl dev.
Run the completion callback in a different thread only for raft
repl devi as there is 100us difference in latency. For nublocks
callback can be run inline.

Author: sanebay

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "7.0.5"
+    version = "7.0.6"
 
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/lib/checkpoint/cp_mgr.cpp

@@ -39,8 +39,10 @@ CPManager::CPManager() :
         [this](meta_blk* mblk, sisl::byte_view buf, size_t size) { on_meta_blk_found(std::move(buf), (void*)mblk); },
         nullptr);
 
-    resource_mgr().register_dirty_buf_exceed_cb(
-        [this]([[maybe_unused]] int64_t dirty_buf_count, bool critical) { this->trigger_cp_flush(false /* force */); });
+    resource_mgr().register_dirty_buf_exceed_cb([this]([[maybe_unused]] int64_t dirty_buf_count, bool critical) {
+        LOGINFO("Dirty buffer exceeded count {} critical {}", dirty_buf_count, critical);
+        this->trigger_cp_flush(false /* force */);
+    });
 
     start_timer_thread();
     start_cp_thread();

src/lib/common/homestore_config.fbs

@@ -107,7 +107,7 @@ table LogStore {
     try_flush_iteration: uint64 = 10240(hotswap);
 
     // Logdev flushes in multiples of this size, setting to 0 will make it use default device optimal size
-    flush_size_multiple_logdev: uint64 = 512;
+    flush_size_multiple_logdev: uint64 = 0;
 
     // Logdev will flush the logs only in a dedicated thread. Turn this on, if flush IO doesn't want to
     // intervene with data IO path.
@@ -165,7 +165,7 @@ table Generic {
 
 table ResourceLimits {
     /* it is going to use 2 times of this space because of two concurrent cps */
-    dirty_buf_percent: uint32 = 1 (hotswap);
+    dirty_buf_percent: uint32 = 10 (hotswap);
 
     /* it is going to use 2 times of this space because of two concurrent cps */
     free_blk_cnt: uint32 = 10000000 (hotswap);
@@ -190,7 +190,7 @@ table ResourceLimits {
     /* 0 means HomeStore doesn't reserve anything and let nuraft controlls the truncation */
     /* default reserve 1 million logs */
     raft_logstore_reserve_threshold: uint32 = 1000000 (hotswap);
-    
+
     /* resource audit timer in ms */
     resource_audit_timer_ms: uint32 = 120000;
 
@@ -262,10 +262,10 @@ table Consensus {
     // Minimum log gap a replica has to be from leader before joining the replica set.
     // 0 indicates the new member will join in cluster immediately.
     min_log_gap_to_join: int32 = 0;
-    
+
     // amount of time in millis to wait on data write before fetch data from remote;
     wait_data_write_timer_ms: uint64 = 1500 (hotswap);
-    
+
     // Leadership expiry (=0 indicates 20 times heartbeat period), set -1 to never expire
     leadership_expiry_ms: int32 = 0;
 

src/lib/device/device_manager.cpp

@@ -115,7 +115,7 @@ void DeviceManager::format_devices() {
     // Get common iomgr_attributes
     for (auto& dinfo : m_dev_infos) {
         format_single_device(dinfo);
-    } 
+    }
 
     // Verify the first blocks to see if the devs are unique
     HS_REL_ASSERT(verify_unique_devs(), "Found duplicate physical devices in the system");
@@ -163,7 +163,9 @@ bool DeviceManager::verify_unique_devs() const {
     for (auto& pdev : m_all_pdevs) {
         if (!pdev) { continue; }
         auto buf = hs_utils::iobuf_alloc(hs_super_blk::first_block_size(), sisl::buftag::superblk, 512);
-        if (auto err = pdev->read_super_block(buf, hs_super_blk::first_block_size(), hs_super_blk::first_block_offset()); err) {
+        if (auto err =
+                pdev->read_super_block(buf, hs_super_blk::first_block_size(), hs_super_blk::first_block_offset());
+            err) {
             LOGERROR("Failed to read first block from device={}, error={}", pdev->get_devname(), err.message());
             ret = false;
             continue;
@@ -704,7 +706,10 @@ uint32_t DeviceManager::populate_pdev_info(const dev_info& dinfo, const iomgr::d
     pinfo.max_pdev_chunks = hs_super_blk::max_chunks_in_pdev(dinfo);
 
     auto sb_size = hs_super_blk::total_size(dinfo);
-    pinfo.data_offset = hs_super_blk::first_block_offset() + sb_size;
+    // Data offset is the data start offset of a pdev. First chunk's is created at data start
+    // offset and remainings chunks are created contiguously one after the other.
+    // Align the data start offset so that chunk start is also aligned to physical page size.
+    pinfo.data_offset = sisl::round_up(hs_super_blk::first_block_offset() + sb_size, attr.phys_page_size);
     pinfo.size = dinfo.dev_size - pinfo.data_offset - (hdd ? sb_size : 0);
     pinfo.dev_attr = attr;
     pinfo.system_uuid = uuid;

src/lib/logstore/log_dev.cpp

@@ -32,6 +32,7 @@
 #include "common/homestore_config.hpp"
 #include "common/homestore_utils.hpp"
 #include "common/crash_simulator.hpp"
+#include "replication/service/generic_repl_svc.h"
 
 namespace homestore {
 
@@ -582,13 +583,26 @@ void LogDev::on_flush_completion(LogGroup* lg) {
     // since we support out-of-order lsn write, so no need to guarantee the order of logstore write completion
     for (auto const& [idx, req] : req_map) {
         m_pending_callback++;
-        iomanager.run_on_forget(iomgr::reactor_regex::random_worker, /* iomgr::fiber_regex::syncio_only, */
-                                [this, dev_offset, idx, req]() {
-                                    auto ld_key = logdev_key{idx, dev_offset};
-                                    auto comp_cb = req->log_store->get_comp_cb();
-                                    (req->cb) ? req->cb(req, ld_key) : comp_cb(req, ld_key);
-                                    m_pending_callback--;
-                                });
+        auto callback_lambda = [this, dev_offset, idx, req]() {
+            auto ld_key = logdev_key{idx, dev_offset};
+            auto comp_cb = req->log_store->get_comp_cb();
+            (req->cb) ? req->cb(req, ld_key) : comp_cb(req, ld_key);
+            m_pending_callback--;
+        };
+
+        // Only server side replication which uses raft runs the callback on a random worker.
+        bool server_side_replication = true;
+        if (hs()->has_repl_data_service()) {
+            auto& repl_svc = dynamic_cast< GenericReplService& >(hs()->repl_service());
+            server_side_replication = repl_svc.get_impl_type() == repl_impl_type::server_side;
+        }
+
+        if (server_side_replication) {
+            iomanager.run_on_forget(iomgr::reactor_regex::random_worker, /* iomgr::fiber_regex::syncio_only, */
+                                    [this, callback_lambda]() { callback_lambda(); });
+        } else {
+            callback_lambda();
+        }
     }
 }
 

src/lib/replication/repl_dev/solo_repl_dev.cpp

@@ -18,7 +18,7 @@ SoloReplDev::SoloReplDev(superblk< solo_repl_dev_superblk >&& rd_sb, bool load_e
     auto const gid = m_rd_sb->group_id;
     if (load_existing) {
         m_logdev_id = m_rd_sb->logdev_id;
-        logstore_service().open_logdev(m_rd_sb->logdev_id, flush_mode_t::TIMER, gid);
+        logstore_service().open_logdev(m_rd_sb->logdev_id, flush_mode_t::TIMER | flush_mode_t::INLINE, gid);
         logstore_service()
             .open_log_store(m_rd_sb->logdev_id, m_rd_sb->logstore_id, true /* append_mode */)
             .thenValue([this](auto log_store) {
@@ -29,7 +29,7 @@ SoloReplDev::SoloReplDev(superblk< solo_repl_dev_superblk >&& rd_sb, bool load_e
             });
         m_commit_upto = m_rd_sb->durable_commit_lsn;
     } else {
-        m_logdev_id = logstore_service().create_new_logdev(flush_mode_t::TIMER, gid);
+        m_logdev_id = logstore_service().create_new_logdev(flush_mode_t::TIMER | flush_mode_t::INLINE, gid);
         m_data_journal = logstore_service().create_new_log_store(m_logdev_id, true /* append_mode */);
         m_rd_sb->logstore_id = m_data_journal->get_store_id();
         m_rd_sb->logdev_id = m_logdev_id;