Perf: eliminate spurious worker wakeups on the log write path

- siso/miso ring buffers computed the low-space flag from a read cursor
  that was only refreshed when the buffer was nearly full, so after 50%
  cumulative usage every log call kept waking the worker for no reason;
  refresh the cursor before raising the flag
- let the worker keep draining while the previous round made progress
  instead of always sleeping the full 66 ms interval

Author: pippocao

src/bq_log/log/log_imp.cpp

@@ -487,14 +487,16 @@ namespace bq {
         merged_log_level_bitmap_ = tmp;
     }
 
-    void log_imp::process(bool is_force_flush)
+    bool log_imp::process(bool is_force_flush)
     {
         constexpr uint64_t flush_io_min_interval_ms = 100;
         uint64_t current_epoch_ms = 0;
+        bool did_work = false;
         while (true) {
             auto read_chunk = buffer_->read_chunk();
             scoped_log_buffer_handle<log_buffer> scoped_read_chunk(*buffer_, read_chunk);
             if (read_chunk.result == enum_buffer_result_code::success) {
+                did_work = true;
                 bq::log_entry_handle log_item(read_chunk.data_addr, read_chunk.data_size);
                 current_epoch_ms = log_item.get_log_head().timestamp_epoch;
                 process_log_chunk(log_item);
@@ -514,6 +516,7 @@ namespace bq {
                 last_flush_io_epoch_ms_ = current_epoch_ms;
             }
         }
+        return did_work;
     }
 
     void log_imp::sync_process(bool is_force_flush)

src/bq_log/log/log_imp.h

@@ -78,7 +78,7 @@ namespace bq {
 
         bq::string& get_config();
 
-        void process(bool is_force_flush);
+        bool process(bool is_force_flush);
 
         void sync_process(bool is_force_flush);
 

src/bq_log/log/log_manager.cpp

@@ -160,22 +160,24 @@ namespace bq {
         return false;
     }
 
-    void log_manager::process_by_worker(log_imp* target_log, bool is_force_flush)
+    bool log_manager::process_by_worker(log_imp* target_log, bool is_force_flush)
     {
         bq::platform::scoped_spin_lock_read_crazy scoped_lock(logs_lock_);
         if (phase::working != phase_.load(bq::platform::memory_order::relaxed)) {
-            return;
+            return false;
         }
+        bool did_work = false;
         if (target_log) {
-            target_log->process(is_force_flush);
+            did_work = target_log->process(is_force_flush);
         } else {
             for (decltype(log_imp_list_)::size_type i = 0; i < log_imp_list_.size(); ++i) {
                 auto& log_impl = log_imp_list_[i];
                 if (log_impl->get_thread_mode() == log_thread_mode::async) {
-                    log_imp_list_[i]->process(is_force_flush);
+                    did_work = log_imp_list_[i]->process(is_force_flush) || did_work;
                 }
             }
         }
+        return did_work;
     }
 
     void log_manager::force_flush_all()

src/bq_log/log/log_manager.h

@@ -36,7 +36,7 @@ namespace bq {
 
         bool reset_config(const bq::string& log_name, const bq::string& config_content);
 
-        void process_by_worker(log_imp* target_log, bool is_force_flush);
+        bool process_by_worker(log_imp* target_log, bool is_force_flush);
 
         uint32_t get_logs_count() const;
 

src/bq_log/log/log_worker.cpp

@@ -99,7 +99,13 @@ namespace bq {
             }
             // normal process
             else {
-                manager_->process_by_worker((thread_mode_ == log_thread_mode::async) ? nullptr : log_target_, false);
+                bool did_work = manager_->process_by_worker((thread_mode_ == log_thread_mode::async) ? nullptr : log_target_, false);
+                if (did_work) {
+                    if (is_cancelled()) {
+                        break;
+                    }
+                    continue;
+                }
             }
             if (is_cancelled()) {
                 break;

src/bq_log/types/buffer/miso_ring_buffer.cpp

@@ -211,6 +211,11 @@ namespace bq {
             handle.data_addr = new_block.chunk_head.data;
             used_blocks_count += need_block_count;
             handle.low_space_flag = ((used_blocks_count << 1) >= aligned_blocks_count_);
+            if (handle.low_space_flag) {
+                read_cursor_ref = cursors_.read_cursor_.load_acquire();
+                used_blocks_count = static_cast<uint32_t>(current_write_cursor + need_block_count - read_cursor_ref);
+                handle.low_space_flag = ((used_blocks_count << 1) >= aligned_blocks_count_);
+            }
 #if defined(BQ_LOG_BUFFER_DEBUG)
             ++result_code_statistics_[(int32_t)enum_buffer_result_code::success];
 #endif

src/bq_log/types/buffer/siso_ring_buffer.cpp

@@ -119,6 +119,11 @@ namespace bq {
         new_block.to_chunk_head().data_size = size;
 
         handle.low_space_flag = ((aligned_blocks_count_ - left_space) << 1) >= aligned_blocks_count_;
+        if (handle.low_space_flag) {
+            head_->wt_reading_cursor_cache_ = head_->reading_cursor().load_acquire();
+            left_space = static_cast<uint32_t>(head_->wt_reading_cursor_cache_ + aligned_blocks_count_ - head_->wt_writing_cursor_cache_);
+            handle.low_space_flag = ((aligned_blocks_count_ - left_space) << 1) >= aligned_blocks_count_;
+        }
         handle.result = enum_buffer_result_code::success;
 #if defined(BQ_LOG_BUFFER_DEBUG)
         ++result_code_statistics_[(int32_t)enum_buffer_result_code::success];