(doris) branch master updated: [fix](auto-partition) Replace std::mutex with bthread::Mutex in VTabletWriter (#57508)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

liaoxin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git


The following commit(s) were added to refs/heads/master by this push:
     new a82699ba087 [fix](auto-partition) Replace std::mutex with bthread::Mutex in VTabletWriter (#57508)
a82699ba087 is described below

commit a82699ba08757c1556f096e7be7fdf1f3625d3a0
Author: Xin Liao <li...@selectdb.com>
AuthorDate: Tue Nov 4 09:43:51 2025 +0800

    [fix](auto-partition) Replace std::mutex with bthread::Mutex in VTabletWriter (#57508)
    
    ### What problem does this PR solve?
    
    Fix potential performance issue where std::mutex was used in bthread
    context. When bthread blocks on std::mutex, it blocks the entire pthread
    instead of just yielding the current bthread, which can cause other
    bthreads on the same pthread to be blocked unnecessarily.
    
    This ensures proper bthread scheduling and improves performance in high
    concurrency scenarios during data writing operations.
---
 be/src/vec/sink/writer/vtablet_writer.cpp | 4 ++--
 be/src/vec/sink/writer/vtablet_writer.h   | 4 +++-
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/be/src/vec/sink/writer/vtablet_writer.cpp b/be/src/vec/sink/writer/vtablet_writer.cpp
index f349ac5f1db..c7abe1baa14 100644
--- a/be/src/vec/sink/writer/vtablet_writer.cpp
+++ b/be/src/vec/sink/writer/vtablet_writer.cpp
@@ -1333,7 +1333,7 @@ void VTabletWriter::_send_batch_process() {
 
     while (true) {
         // incremental open will temporarily make channels into abnormal state. stop checking when this.
-        std::unique_lock<std::mutex> l(_stop_check_channel);
+        std::unique_lock<bthread::Mutex> l(_stop_check_channel);
 
         int running_channels_num = 0;
         int opened_nodes = 0;
@@ -1637,7 +1637,7 @@ Status VTabletWriter::_init(RuntimeState* state, RuntimeProfile* profile) {
 Status VTabletWriter::_incremental_open_node_channel(
         const std::vector<TOlapTablePartition>& partitions) {
     // do what we did in prepare() for partitions. indexes which don't change when we create new partition is orthogonal to partitions.
-    std::unique_lock<std::mutex> _l(_stop_check_channel);
+    std::unique_lock<bthread::Mutex> _l(_stop_check_channel);
     for (int i = 0; i < _schema->indexes().size(); ++i) {
         const OlapTableIndexSchema* index = _schema->indexes()[i];
         std::vector<TTabletWithPartition> tablets;
diff --git a/be/src/vec/sink/writer/vtablet_writer.h b/be/src/vec/sink/writer/vtablet_writer.h
index 64dcbb09f82..ba1c0c505d6 100644
--- a/be/src/vec/sink/writer/vtablet_writer.h
+++ b/be/src/vec/sink/writer/vtablet_writer.h
@@ -31,6 +31,8 @@
 #include <google/protobuf/stubs/callback.h>
 
 // IWYU pragma: no_include <bits/chrono.h>
+#include <bthread/mutex.h>
+
 #include <atomic>
 #include <chrono> // IWYU pragma: keep
 #include <cstddef>
@@ -702,7 +704,7 @@ private:
     std::unique_ptr<OlapTabletFinder> _tablet_finder;
 
     // index_channel
-    std::mutex _stop_check_channel;
+    bthread::Mutex _stop_check_channel;
     std::vector<std::shared_ptr<IndexChannel>> _channels;
     std::unordered_map<int64_t, std::shared_ptr<IndexChannel>> _index_id_to_channel;
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org