Merge pull request #842 from szmyd/cherry-pick-align

Align chunk start offset to 4k and logstore inline flush

Author: szmyd

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "6.20.32"
+    version = "6.20.33"
 
     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;