Avoid flushing on wbcache non dirty buffers in merge case

In merge case, if merge not required and prep txn already called,
parent wont be dirty adn we flush the cache causing invalid
dirty buffer count. Concurrent thread iterators can't be reused.

conanfile.py

@@ -5,7 +5,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "5.0.7"
+    version = "5.0.8"
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"
     topics = ("ebay", "nublox")

src/lib/common/resource_mgr.cpp

@@ -36,7 +36,7 @@ void ResourceMgr::dec_dirty_buf_size(const uint32_t size) {
     HS_REL_ASSERT_GT(size, 0);
     const int64_t dirty_buf_cnt = m_hs_dirty_buf_cnt.fetch_sub(size, std::memory_order_relaxed);
     COUNTER_DECREMENT(m_metrics, dirty_buf_cnt, size);
-    HS_REL_ASSERT_GE(dirty_buf_cnt, 0);
+    HS_REL_ASSERT_GE(dirty_buf_cnt, size);
 }
 
 void ResourceMgr::register_dirty_buf_exceed_cb(exceed_limit_cb_t cb) { m_dirty_buf_exceed_cb = std::move(cb); }
@@ -48,10 +48,6 @@ void ResourceMgr::inc_free_blk(int size) {
     auto sz = m_hs_fb_size.fetch_add(size, std::memory_order_relaxed);
     COUNTER_INCREMENT(m_metrics, free_blk_size_in_cp, size);
     COUNTER_INCREMENT(m_metrics, free_blk_cnt_in_cp, 1);
-
-    if (m_dirty_buf_exceed_cb && (cnt > get_free_blk_cnt_limit() || sz > get_free_blk_size_limit())) {
-        m_dirty_buf_exceed_cb(cnt);
-    }
 }
 
 void ResourceMgr::dec_free_blk(int size) {

src/lib/index/index_cp.hpp

@@ -37,7 +37,6 @@ struct IndexCPContext : public VDevCPContext {
     sisl::ConcurrentInsertVector< IndexBufferPtr >::iterator m_dirty_buf_it;
 
 public:
-
     IndexCPContext(CP* cp) : VDevCPContext(cp) {}
     virtual ~IndexCPContext() = default;
 
@@ -66,17 +65,12 @@ struct IndexCPContext : public VDevCPContext {
         // Display all buffers and its dependencies and state.
         std::unordered_map< IndexBuffer*, std::vector< IndexBuffer* > > parents;
 
-        auto it = m_dirty_buf_list.begin();
-        while (it != m_dirty_buf_list.end()) {
+        m_dirty_buf_list.foreach_entry([&parents](IndexBufferPtr buf) {
             // Add this buf to his children.
-            IndexBufferPtr buf = *it;
             parents[buf->m_next_buffer.lock().get()].emplace_back(buf.get());
-            ++it;
-        }
+        });
 
-        it = m_dirty_buf_list.begin();
-        while (it != m_dirty_buf_list.end()) {
-            IndexBufferPtr buf = *it;
+        m_dirty_buf_list.foreach_entry([&str, &parents](IndexBufferPtr buf) {
             fmt::format_to(std::back_inserter(str), "{}", buf->to_string());
             auto first = true;
             for (const auto& p : parents[buf.get()]) {
@@ -87,9 +81,7 @@ struct IndexCPContext : public VDevCPContext {
                 fmt::format_to(std::back_inserter(str), " {}({})", r_cast< void* >(p), s_cast< int >(p->state()));
             }
             fmt::format_to(std::back_inserter(str), "\n");
-            ++it;
-        }
-
+        });
         return str;
     }
 
@@ -153,7 +145,6 @@ struct IndexCPContext : public VDevCPContext {
     }
 };
 
-
 class IndexWBCache;
 class IndexCPCallbacks : public CPCallbacks {
 public:

src/lib/index/wb_cache.cpp

@@ -307,7 +307,8 @@ void IndexWBCache::get_next_bufs_internal(IndexCPContext* cp_ctx, uint32_t max_c
     // First attempt to execute any follower buffer flush
     if (prev_flushed_buf) {
         auto next_buffer = prev_flushed_buf->m_next_buffer.lock();
-        if (next_buffer && next_buffer->m_wait_for_leaders.decrement_testz()) {
+        if (next_buffer && next_buffer->state() == index_buf_state_t::DIRTY &&
+            next_buffer->m_wait_for_leaders.decrement_testz()) {
             bufs.emplace_back(next_buffer);
             ++count;
         }

src/tests/CMakeLists.txt

@@ -110,7 +110,7 @@ if (${io_tests})
         add_test(NAME LogStore-Epoll COMMAND ${CMAKE_BINARY_DIR}/bin/test_log_store)
         add_test(NAME MetaBlkMgr-Epoll COMMAND ${CMAKE_BINARY_DIR}/bin/test_meta_blk_mgr)
         add_test(NAME DataService-Epoll COMMAND ${CMAKE_BINARY_DIR}/bin/test_data_service)
-        add_test(NAME SoloReplDev-Epoll COMMAND ${CMAKE_BINARY_DIR}/bin/test_solo_repl_dev)
+        # add_test(NAME SoloReplDev-Epoll COMMAND ${CMAKE_BINARY_DIR}/bin/test_solo_repl_dev)
         # add_test(NAME HomeRaftLogStore-Epoll COMMAND ${CMAKE_BINARY_DIR}/bin/test_home_raft_logstore)
         # add_test(NAME RaftReplDev-Epoll COMMAND ${CMAKE_BINARY_DIR}/bin/test_raft_repl_dev)
     endif()

src/tests/test_index_btree.cpp

@@ -123,7 +123,7 @@ struct BtreeTest : public BtreeTestHelper< TestType >, public ::testing::Test {
     }
 };
 
-using BtreeTypes = testing::Types< FixedLenBtree /* , VarKeySizeBtree, VarValueSizeBtree, VarObjSizeBtree */ >;
+using BtreeTypes = testing::Types< FixedLenBtree /*, VarKeySizeBtree, VarValueSizeBtree, VarObjSizeBtree */ >;
 
 TYPED_TEST_SUITE(BtreeTest, BtreeTypes);