logstore: reschedule flush_if_necessary() on lost try_lock instead of dropping it

conanfile.py

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

src/include/homestore/btree/detail/btree_internal.hpp

@@ -315,33 +315,29 @@ class BtreeMetrics : public sisl::MetricsGroup {
         REGISTER_COUNTER(btree_num_pc_gen_mismatch, "Number of gen mismatches to recover");
 
         REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_int_node_occupancy, "Interior node occupancy",
-                                                              "btree_node_occupancy", {"node_type", "interior"},
-                                                              HistogramBucketsType(PercentileBuckets));
+                                                      "btree_node_occupancy", {"node_type", "interior"},
+                                                      HistogramBucketsType(PercentileBuckets));
         REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_leaf_node_occupancy, "Leaf node occupancy",
-                                                              "btree_node_occupancy", {"node_type", "leaf"},
-                                                              HistogramBucketsType(PercentileBuckets));
+                                                      "btree_node_occupancy", {"node_type", "leaf"},
+                                                      HistogramBucketsType(PercentileBuckets));
         REGISTER_COUNTER(btree_retry_count, "number of retries");
         REGISTER_COUNTER(write_err_cnt, "number of errors in write");
         REGISTER_COUNTER(query_err_cnt, "number of errors in query");
         REGISTER_COUNTER(btree_write_ops_count, "number of btree operations");
         REGISTER_COUNTER(btree_query_ops_count, "number of btree operations");
         REGISTER_COUNTER(btree_remove_ops_count, "number of btree operations");
-        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_exclusive_time_in_int_node,
-                                                              "Exclusive time spent (Write locked) on interior node (ns)",
-                                                              "btree_exclusive_time_in_node", {"node_type", "interior"},
-                                                              HistogramBucketsType(OpLatecyBuckets));
-        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_exclusive_time_in_leaf_node,
-                                                              "Exclusive time spent (Write locked) on leaf node (ns)",
-                                                              "btree_exclusive_time_in_node", {"node_type", "leaf"},
-                                                              HistogramBucketsType(OpLatecyBuckets));
-        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_inclusive_time_in_int_node,
-                                                              "Inclusive time spent (Read locked) on interior node (ns)",
-                                                              "btree_inclusive_time_in_node", {"node_type", "interior"},
-                                                              HistogramBucketsType(OpLatecyBuckets));
-        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(btree_inclusive_time_in_leaf_node,
-                                                              "Inclusive time spent (Read locked) on leaf node (ns)",
-                                                              "btree_inclusive_time_in_node", {"node_type", "leaf"},
-                                                              HistogramBucketsType(OpLatecyBuckets));
+        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(
+            btree_exclusive_time_in_int_node, "Exclusive time spent (Write locked) on interior node (ns)",
+            "btree_exclusive_time_in_node", {"node_type", "interior"}, HistogramBucketsType(OpLatecyBuckets));
+        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(
+            btree_exclusive_time_in_leaf_node, "Exclusive time spent (Write locked) on leaf node (ns)",
+            "btree_exclusive_time_in_node", {"node_type", "leaf"}, HistogramBucketsType(OpLatecyBuckets));
+        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(
+            btree_inclusive_time_in_int_node, "Inclusive time spent (Read locked) on interior node (ns)",
+            "btree_inclusive_time_in_node", {"node_type", "interior"}, HistogramBucketsType(OpLatecyBuckets));
+        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(
+            btree_inclusive_time_in_leaf_node, "Inclusive time spent (Read locked) on leaf node (ns)",
+            "btree_inclusive_time_in_node", {"node_type", "leaf"}, HistogramBucketsType(OpLatecyBuckets));
 
         register_me_to_farm();
     }

src/include/homestore/meta_service.hpp

@@ -63,7 +63,7 @@ class MetablkMetrics : public sisl::MetricsGroupWrapper {
         REGISTER_COUNTER(compress_backoff_ratio_cnt, "compression back-off cnt because of exceeding ratio limit");
 
         REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(compress_ratio_percent, "compression ration percentage",
-                                                              HistogramBucketsType(PercentileBuckets));
+                                                      HistogramBucketsType(PercentileBuckets));
         register_me_to_farm();
     }
 

src/lib/blkalloc/varsize_blk_allocator.h

@@ -179,9 +179,8 @@ class BlkAllocMetrics : public sisl::MetricsGroup {
         REGISTER_COUNTER(num_retries, "Number of times it retried because of empty cache");
         REGISTER_COUNTER(num_blks_alloc_direct, "Number of blks alloc attempt directly because of empty cache");
 
-        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(frag_pct_distribution,
-                                                              "Distribution of fragmentation percentage",
-                                                              HistogramBucketsType(PercentileBuckets));
+        REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(frag_pct_distribution, "Distribution of fragmentation percentage",
+                                                      HistogramBucketsType(PercentileBuckets));
 
         register_me_to_farm();
     }

src/lib/checkpoint/cp_mgr.cpp

@@ -97,13 +97,11 @@ void CPManager::start_timer() {
     auto usecs = cp_timer_us();
     LOGINFO("cp timer is set to {} usec", usecs);
     iomanager.run_on_wait(m_timer_fiber, [this, usecs]() {
-        m_cp_timer_hdl = iomanager.schedule_thread_timer(usecs * 1000, true /* recurring */, nullptr /* cookie */,
-                                                         [this](void*, uint64_t exp_count) {
-                                                             if (exp_count > 1) {
-                                                                 LOGINFO("cp timer expired {} times, running once", exp_count);
-                                                             }
-                                                             trigger_cp_flush(false /* false */);
-                                                         });
+        m_cp_timer_hdl = iomanager.schedule_thread_timer(
+            usecs * 1000, true /* recurring */, nullptr /* cookie */, [this](void*, uint64_t exp_count) {
+                if (exp_count > 1) { LOGINFO("cp timer expired {} times, running once", exp_count); }
+                trigger_cp_flush(false /* false */);
+            });
     });
 }
 

src/lib/common/homestore_assert.hpp

@@ -144,15 +144,15 @@
                                                               fmt::make_format_args(detail_name, detail_val))))        \
                 ();                                                                                                    \
                 fmt::vformat_to(fmt::appender{buf}, fmt::string_view{msgcb}, fmt::make_format_args(args...));          \
-                return check_and_format_log(buf, freq, 0);                                                                                          \
+                return check_and_format_log(buf, freq, 0);                                                             \
             }),                                                                                                        \
             msg, ##__VA_ARGS__);                                                                                       \
     }
 
 #define HS_LOG_EVERY_N(level, mod, freq, msg, ...) HS_DETAILED_LOG_EVERY_N(level, mod, freq, , , , , msg, ##__VA_ARGS__)
 
 #define HS_DETAILED_LOG_EVERY_N_SEC(level, mod, interval_sec, submod_name, submod_val, detail_name, detail_val, msg,   \
-                                     ...)                                                                              \
+                                    ...)                                                                               \
     {                                                                                                                  \
         LOG##level##MOD_FMT(                                                                                           \
             BOOST_PP_IF(BOOST_VMD_IS_EMPTY(mod), base, mod),                                                           \
@@ -168,16 +168,16 @@
                                                               fmt::make_format_args(detail_name, detail_val))))        \
                 ();                                                                                                    \
                 fmt::vformat_to(fmt::appender{buf}, fmt::string_view{msgcb}, fmt::make_format_args(args...));          \
-                return check_and_format_log(buf, 0, interval_sec);                                                                                          \
+                return check_and_format_log(buf, 0, interval_sec);                                                     \
             }),                                                                                                        \
             msg, ##__VA_ARGS__);                                                                                       \
     }
 
 #define HS_LOG_EVERY_N_SEC(level, mod, interval_sec, msg, ...)                                                         \
     HS_DETAILED_LOG_EVERY_N_SEC(level, mod, interval_sec, , , , , msg, ##__VA_ARGS__)
 
-#define HS_DETAILED_LOG_EVERY_N_OR_SEC(level, mod, freq, interval_sec, submod_name, submod_val, detail_name,          \
-                                        detail_val, msg, ...)                                                          \
+#define HS_DETAILED_LOG_EVERY_N_OR_SEC(level, mod, freq, interval_sec, submod_name, submod_val, detail_name,           \
+                                       detail_val, msg, ...)                                                           \
     {                                                                                                                  \
         LOG##level##MOD_FMT(                                                                                           \
             BOOST_PP_IF(BOOST_VMD_IS_EMPTY(mod), base, mod),                                                           \
@@ -193,7 +193,7 @@
                                                               fmt::make_format_args(detail_name, detail_val))))        \
                 ();                                                                                                    \
                 fmt::vformat_to(fmt::appender{buf}, fmt::string_view{msgcb}, fmt::make_format_args(args...));          \
-                return check_and_format_log(buf, freq, interval_sec);                                                                                          \
+                return check_and_format_log(buf, freq, interval_sec);                                                  \
             }),                                                                                                        \
             msg, ##__VA_ARGS__);                                                                                       \
     }
@@ -361,7 +361,7 @@
  * If interval_sec >= 300, the behavior effectively becomes "log first occurrence after each 5min reset."
  */
 [[maybe_unused]] static bool check_and_format_log(fmt::memory_buffer& buf, uint64_t freq = 0,
-                                                    uint64_t interval_sec = 0) {
+                                                  uint64_t interval_sec = 0) {
     static constexpr uint64_t COUNTER_RESET_SEC{300}; // Reset every 5 minutes
     static thread_local Clock::time_point last_cleanup{Clock::now()};
     static thread_local std::unordered_map< size_t, std::pair< uint32_t, uint64_t > > log_map{};
@@ -388,8 +388,7 @@
     const size_t msg_hash = std::hash< std::string_view >{}(msg);
 
     // Milliseconds since last cleanup (max ~49 days with uint32_t)
-    const uint32_t now_ms =
-        std::chrono::duration_cast< std::chrono::milliseconds >(now - last_cleanup).count();
+    const uint32_t now_ms = std::chrono::duration_cast< std::chrono::milliseconds >(now - last_cleanup).count();
 
     auto [it, happened] = log_map.emplace(msg_hash, std::make_pair(now_ms, 0));
     uint32_t elapsed_ms = 0;
@@ -429,8 +428,8 @@
 
         // Update state after logging (so next log shows "since this log")
         if (!happened) {
-            it->second.first = now_ms;  // Update timestamp
-            it->second.second = 0;      // Reset count (next occurrence will be "1 since this log")
+            it->second.first = now_ms; // Update timestamp
+            it->second.second = 0;     // Reset count (next occurrence will be "1 since this log")
         }
     }
 

src/lib/device/virtual_dev.hpp

@@ -55,7 +55,7 @@ class VirtualDevMetrics : public sisl::MetricsGroupWrapper {
         REGISTER_COUNTER(random_chunk_allocation_cnt,
                          "random chunk allocation count"); // ideally it should be zero for hdd
         REGISTER_HISTOGRAM_WITH_CARDINALITY_REDUCTION(blk_alloc_latency, "Blk allocation latency", "blk_alloc_latency",
-                                                              {}, HistogramBucketsType(OpLatecyBuckets));
+                                                      {}, HistogramBucketsType(OpLatecyBuckets));
         register_me_to_farm();
     }
 

src/lib/logstore/log_dev.cpp

@@ -437,12 +437,12 @@ bool LogDev::can_flush_in_this_thread() {
     return (!HS_DYNAMIC_CONFIG(logstore.flush_only_in_dedicated_thread) && iomanager.am_i_worker_reactor());
 }
 
-bool LogDev::flush_if_necessary(int64_t threshold_size) {
+bool LogDev::flush_if_necessary(int64_t threshold_size, bool force) {
     if (is_stopping()) return false;
     incr_pending_request_num();
     if (!can_flush_in_this_thread()) {
         iomanager.run_on_forget(logstore_service().flush_thread(),
-                                [this, threshold_size]() { flush_if_necessary(threshold_size); });
+                                [this, threshold_size, force]() { flush_if_necessary(threshold_size, force); });
         decr_pending_request_num();
         return false;
     }
@@ -456,12 +456,37 @@ bool LogDev::flush_if_necessary(int64_t threshold_size) {
     bool const flush_by_size = (pending_sz >= threshold_size);
     bool const flush_by_time =
         !flush_by_size && pending_sz && (elapsed_time > HS_DYNAMIC_CONFIG(logstore.max_time_between_flush_us));
-    if (flush_by_size || flush_by_time) {
+    if (force || flush_by_size || flush_by_time) {
         std::unique_lock lck(m_flush_mtx, std::try_to_lock);
         if (lck.owns_lock()) {
             decr_pending_request_num();
             return flush();
         }
+        // Lost the race to a concurrent flush() (e.g. another write's own flush_if_necessary() call, or
+        // HomeLogStore::truncate()'s internal flush()). That concurrent flush's own snapshot of m_log_idx
+        // may not include the data that made this call decide to flush, so giving up here silently can
+        // leave that data unflushed indefinitely if nothing else ever calls flush_if_necessary() again
+        // for this logdev -- normally masked by the periodic flush timer eventually retrying, but with
+        // it disabled (flush_timer_frequency_us=0, e.g. in tests) this is a real, reproducible hang: the
+        // very last write issued in a run has no later trigger to fall back on.
+        //
+        // Reschedule with force=true. LogDev::flush() unconditionally resets m_last_flush_time at its
+        // start, so the flush that just won this race may have already reset that clock by the time the
+        // retry runs -- re-deriving flush_by_size/flush_by_time here could then read false again (this
+        // write's own size may still be under threshold_size) and silently abandon the retry for good.
+        // force=true skips that re-derivation: we already decided to flush, so the retry only needs to
+        // keep trying the lock, not re-litigate whether to.
+        //
+        // Target a random *worker* reactor, not flush_thread() directly, even though the retry ultimately
+        // needs to run there (can_flush_in_this_thread() bounces it back). We're already ON flush_thread
+        // here, and IOReactor::deliver_msg runs same-reactor targets inline instead of queuing them --
+        // so posting straight back to flush_thread would recurse synchronously on every failed try_lock.
+        // Under sustained contention this stack-overflows the process (confirmed with the lock held for
+        // ~200ms in testing; this predates force -- it's already present in the plain reschedule above).
+        // Routing through random_worker first forces a real queued hop, so the stack unwinds between
+        // attempts no matter how many times try_lock fails.
+        iomanager.run_on_forget(iomgr::reactor_regex::random_worker,
+                                [this, threshold_size]() { flush_if_necessary(threshold_size, /* force = */ true); });
     }
     decr_pending_request_num();
     return false;
@@ -481,6 +506,12 @@ bool LogDev::flush_under_guard() {
     return flush();
 }
 
+#ifdef _PRERELEASE
+void LogDev::test_touch_last_flush_time() {
+    if (iomgr_flip::instance()->test_flip("test_touch_last_flush_time")) { m_last_flush_time = Clock::now(); }
+}
+#endif
+
 bool LogDev::flush() {
     if (!is_ready()) {
         THIS_LOGDEV_LOG(INFO, "LogDev is not ready to flush, log_dev={}", m_logdev_id);

src/lib/logstore/log_dev.hpp

@@ -583,6 +583,18 @@ class LogDev : public std::enable_shared_from_this< LogDev > {
         return HS_DYNAMIC_CONFIG(logstore.flush_threshold_size) - sizeof(log_group_header);
     }
 
+#ifdef _PRERELEASE
+    // Test-only: acquires m_flush_mtx directly, bypassing flush() -- lets a UT hold the real lock
+    // deterministically without a real flush cycle, which would unavoidably flush any data appended
+    // while held (its m_log_idx snapshot is taken fresh at call time). See
+    // LogStoreTest.FlushIfNecessaryRetrySurvivesStaleClockReset.
+    std::unique_lock< iomgr::FiberManagerLib::mutex > test_acquire_flush_mtx() { return std::unique_lock(m_flush_mtx); }
+
+    // Test-only: simulates a concurrent flush's clock-reset side effect (resets m_last_flush_time)
+    // without actually flushing. Only active if the "test_touch_last_flush_time" flip is armed.
+    void test_touch_last_flush_time();
+#endif
+
     LogDev(logdev_id_t logdev_id,
            flush_mode_t flush_mode = static_cast< flush_mode_t >(HS_DYNAMIC_CONFIG(logstore.flush_mode)),
            uuid_t pid = boost::uuids::nil_uuid());
@@ -653,9 +665,12 @@ class LogDev : public std::enable_shared_from_this< LogDev > {
     /// redirect the flush to a flush thread and run there.
     ///
     /// @param threshold_size [Optional]: Size in bytes after which it will flush, if set to -1, will use default size
+    /// @param force [Optional]: Skip the size/time threshold check and go straight to the try_lock. Used internally
+    ///        when rescheduling a retry after losing the try_lock race, so the retry can't be silently talked out of
+    ///        trying again by a stale-clock re-derivation of the threshold check.
     ///
     /// @return bool : True if it has flushed the data, false otherwise
-    bool flush_if_necessary(int64_t threshold_size = -1);
+    bool flush_if_necessary(int64_t threshold_size = -1, bool force = false);
 
     /// @brief : Look at all logstore and find out the safest point upto which it can truncate and truncate them.
     ///