This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 88db7a48f86 branch-4.0: [fix](cloud) Use bthread-aware shared mutex
for tablet header lock (#66019)
88db7a48f86 is described below
commit 88db7a48f862c37fb6e3517b08e1d7b48a768d00
Author: Xin Liao <[email protected]>
AuthorDate: Tue Jul 28 17:11:54 2026 +0800
branch-4.0: [fix](cloud) Use bthread-aware shared mutex for tablet header
lock (#66019)
Pick apache/doris#64574
---
be/src/cloud/cloud_base_compaction.cpp | 2 +-
be/src/cloud/cloud_cumulative_compaction.cpp | 2 +-
be/src/cloud/cloud_full_compaction.cpp | 2 +-
be/src/cloud/cloud_meta_mgr.cpp | 4 +-
be/src/cloud/cloud_meta_mgr.h | 2 +-
be/src/cloud/cloud_tablet.cpp | 49 ++--
be/src/cloud/cloud_tablet.h | 16 +-
be/src/olap/base_tablet.cpp | 4 +
be/src/olap/base_tablet.h | 9 +-
be/src/olap/compaction.cpp | 4 +-
.../cumulative_compaction_time_series_policy.cpp | 5 +-
be/src/olap/full_compaction.cpp | 2 +-
be/src/olap/rowset_builder.cpp | 2 +-
be/src/olap/schema_change.cpp | 4 +-
be/src/olap/tablet.cpp | 63 +++--
be/src/olap/tablet.h | 9 +-
be/src/olap/tablet_manager.cpp | 2 +-
be/src/olap/task/engine_clone_task.cpp | 2 +-
be/src/olap/task/index_builder.cpp | 4 +-
be/src/util/bthread_shared_mutex.h | 125 ++++++++++
.../cloud/cloud_empty_rowset_compaction_test.cpp | 2 +-
be/test/cloud/cloud_meta_mgr_test.cpp | 28 +--
be/test/util/bthread_shared_mutex_test.cpp | 264 +++++++++++++++++++++
23 files changed, 515 insertions(+), 91 deletions(-)
diff --git a/be/src/cloud/cloud_base_compaction.cpp
b/be/src/cloud/cloud_base_compaction.cpp
index e0d424c413c..4030f6f0390 100644
--- a/be/src/cloud/cloud_base_compaction.cpp
+++ b/be/src/cloud/cloud_base_compaction.cpp
@@ -188,7 +188,7 @@ Status CloudBaseCompaction::pick_rowsets_to_compact() {
std::shared_lock rlock(_tablet->get_header_lock());
_base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
_cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
- _input_rowsets =
cloud_tablet()->pick_candidate_rowsets_to_base_compaction();
+ _input_rowsets =
cloud_tablet()->pick_candidate_rowsets_to_base_compaction_unlocked();
}
if (auto st = check_version_continuity(_input_rowsets); !st.ok()) {
DCHECK(false) << st;
diff --git a/be/src/cloud/cloud_cumulative_compaction.cpp
b/be/src/cloud/cloud_cumulative_compaction.cpp
index 2fd3a8fc0a3..2a867812fb7 100644
--- a/be/src/cloud/cloud_cumulative_compaction.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction.cpp
@@ -502,7 +502,7 @@ Status CloudCumulativeCompaction::pick_rowsets_to_compact()
{
std::max(cloud_tablet()->cumulative_layer_point(),
_max_conflict_version + 1),
cloud_tablet()->alter_version() + 1);
// Get all rowsets whose version >= `candidate_version` as candidate
rowsets
- cloud_tablet()->traverse_rowsets(
+ cloud_tablet()->traverse_rowsets_unlocked(
[&candidate_rowsets, candidate_version](const RowsetSharedPtr&
rs) {
if (rs->start_version() >= candidate_version) {
candidate_rowsets.push_back(rs);
diff --git a/be/src/cloud/cloud_full_compaction.cpp
b/be/src/cloud/cloud_full_compaction.cpp
index 15160cfabd4..f2fbdd88993 100644
--- a/be/src/cloud/cloud_full_compaction.cpp
+++ b/be/src/cloud/cloud_full_compaction.cpp
@@ -127,7 +127,7 @@ Status CloudFullCompaction::pick_rowsets_to_compact() {
std::shared_lock rlock(_tablet->get_header_lock());
_base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
_cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
- _input_rowsets =
cloud_tablet()->pick_candidate_rowsets_to_full_compaction();
+ _input_rowsets =
cloud_tablet()->pick_candidate_rowsets_to_full_compaction_unlocked();
}
if (auto st = check_version_continuity(_input_rowsets); !st.ok()) {
DCHECK(false) << st;
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index 7bfc5705b81..c302a4184f9 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -514,7 +514,7 @@ Status CloudMetaMgr::_log_mow_delete_bitmap(CloudTablet*
tablet, GetRowsetRespon
std::vector<RowsetSharedPtr> old_rowsets;
RowsetIdUnorderedSet old_rowset_ids;
{
- std::lock_guard<std::shared_mutex>
rlock(tablet->get_header_lock());
+ std::lock_guard rlock(tablet->get_header_lock());
RETURN_IF_ERROR(tablet->get_all_rs_id_unlocked(old_max_version,
&old_rowset_ids));
old_rowsets = tablet->get_rowset_by_ids(&old_rowset_ids);
}
@@ -2102,7 +2102,7 @@ int64_t
CloudMetaMgr::get_inverted_index_file_size(RowsetMeta& rs_meta) {
}
Status CloudMetaMgr::fill_version_holes(CloudTablet* tablet, int64_t
max_version,
- std::unique_lock<std::shared_mutex>&
wlock) {
+ std::unique_lock<BthreadSharedMutex>&
wlock) {
if (max_version <= 0) {
return Status::OK();
}
diff --git a/be/src/cloud/cloud_meta_mgr.h b/be/src/cloud/cloud_meta_mgr.h
index ed2696192a4..1555f53713f 100644
--- a/be/src/cloud/cloud_meta_mgr.h
+++ b/be/src/cloud/cloud_meta_mgr.h
@@ -157,7 +157,7 @@ public:
// Fill version holes by creating empty rowsets for missing versions
Status fill_version_holes(CloudTablet* tablet, int64_t max_version,
- std::unique_lock<std::shared_mutex>& wlock);
+ std::unique_lock<BthreadSharedMutex>& wlock);
// Create an empty rowset to fill a version hole
Status create_empty_rowset_for_hole(CloudTablet* tablet, int64_t version,
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index 8a30cb75b5a..7898102bedb 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -189,7 +189,10 @@ Result<std::vector<Version>>
CloudTablet::capture_versions_prefer_cache(
const Version& spec_version) const {
g_capture_prefer_cache_count << 1;
Versions version_path;
- std::shared_lock rlock(_meta_lock);
+ // Caller (capture_consistent_versions_unlocked) already holds shared
+ // `_meta_lock`; do NOT re-acquire it here. The lock is writer-preferring,
+ // so a recursive shared acquisition self-deadlocks if a writer queues in
+ // between the outer and inner lock.
auto st =
_timestamped_version_tracker.capture_consistent_versions_prefer_cache(
spec_version, version_path,
[&](int64_t start, int64_t end) { return
rowset_is_warmed_up_unlocked(start, end); });
@@ -241,7 +244,10 @@ Result<std::vector<Version>>
CloudTablet::capture_versions_with_freshness_tolera
auto freshness_limit_tp = system_clock::now() -
milliseconds(query_freshness_tolerance_ms);
// find a version path where every edge(rowset) has been warmuped
Versions version_path;
- std::shared_lock rlock(_meta_lock);
+ // Caller (capture_consistent_versions_unlocked) already holds shared
+ // `_meta_lock`; do NOT re-acquire it here. The lock is writer-preferring,
+ // so a recursive shared acquisition self-deadlocks if a writer queues in
+ // between the outer and inner lock.
if (enable_unique_key_merge_on_write()) {
// For merge-on-write table, newly generated delete bitmap marks will
be on the rowsets which are in newest layout.
// So we can ony capture rowsets which are in newest data layout.
Otherwise there may be data correctness issue.
@@ -293,7 +299,8 @@ Result<std::vector<Version>>
CloudTablet::capture_versions_with_freshness_tolera
std::ranges::any_of(std::views::values(_tablet_meta->all_stale_rs_metas()),
should_be_visible_but_not_warmed_up);
if (should_fallback) {
- rlock.unlock();
+ // The outer caller still holds the shared `_meta_lock`; the base
+ // unlocked fallback below runs under that lock.
g_capture_with_freshness_tolerance_fallback_count << 1;
// if there exists a rowset which satisfies freshness tolerance and
its start version is larger than the path max version
// but has not been warmuped up yet, fallback to capture rowsets as
usual
@@ -439,7 +446,7 @@ Status CloudTablet::sync_if_not_running(SyncRowsetStats*
stats) {
}
void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool
version_overlap,
- std::unique_lock<std::shared_mutex>& meta_lock,
+ std::unique_lock<BthreadSharedMutex>& meta_lock,
bool warmup_delta_data) {
if (to_add.empty()) {
return;
@@ -690,7 +697,7 @@ void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr>
to_add, bool version_
}
void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
- std::unique_lock<std::shared_mutex>&) {
+ std::unique_lock<BthreadSharedMutex>&) {
if (to_delete.empty()) {
return;
}
@@ -711,7 +718,7 @@ void CloudTablet::delete_rowsets(const
std::vector<RowsetSharedPtr>& to_delete,
}
void CloudTablet::delete_rowsets_for_schema_change(const
std::vector<RowsetSharedPtr>& to_delete,
-
std::unique_lock<std::shared_mutex>&,
+
std::unique_lock<BthreadSharedMutex>&,
bool
recycle_deleted_rowsets) {
if (to_delete.empty()) {
return;
@@ -743,7 +750,7 @@ void CloudTablet::delete_rowsets_for_schema_change(const
std::vector<RowsetShare
void CloudTablet::replace_rowsets_with_schema_change_output(
const std::vector<RowsetSharedPtr>& output_rowsets, int64_t
alter_version,
- std::unique_lock<std::shared_mutex>& meta_lock, const char* stage,
+ std::unique_lock<BthreadSharedMutex>& meta_lock, const char* stage,
bool recycle_deleted_rowsets) {
std::vector<RowsetSharedPtr> to_delete;
for (auto& [v, rs] : _rs_version_map) {
@@ -1094,7 +1101,7 @@ int64_t CloudTablet::get_cloud_base_compaction_score()
const {
if (_tablet_meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {
bool has_delete = false;
int64_t point = cumulative_layer_point();
- std::shared_lock<std::shared_mutex> rlock(_meta_lock);
+ std::shared_lock rlock(_meta_lock);
for (const auto& [_, rs_meta] : _tablet_meta->all_rs_metas()) {
if (rs_meta->start_version() >= point) {
continue;
@@ -1355,30 +1362,24 @@ Result<RowsetSharedPtr>
CloudTablet::pick_a_rowset_for_index_change(int schema_v
return ret_rowset;
}
-std::vector<RowsetSharedPtr>
CloudTablet::pick_candidate_rowsets_to_base_compaction() {
+std::vector<RowsetSharedPtr>
CloudTablet::pick_candidate_rowsets_to_base_compaction_unlocked() {
std::vector<RowsetSharedPtr> candidate_rowsets;
- {
- std::shared_lock rlock(_meta_lock);
- for (const auto& [version, rs] : _rs_version_map) {
- if (version.first != 0 && version.first < _cumulative_point &&
- (_alter_version == -1 || version.second <= _alter_version)) {
- candidate_rowsets.push_back(rs);
- }
+ for (const auto& [version, rs] : _rs_version_map) {
+ if (version.first != 0 && version.first < _cumulative_point &&
+ (_alter_version == -1 || version.second <= _alter_version)) {
+ candidate_rowsets.push_back(rs);
}
}
std::sort(candidate_rowsets.begin(), candidate_rowsets.end(),
Rowset::comparator);
return candidate_rowsets;
}
-std::vector<RowsetSharedPtr>
CloudTablet::pick_candidate_rowsets_to_full_compaction() {
+std::vector<RowsetSharedPtr>
CloudTablet::pick_candidate_rowsets_to_full_compaction_unlocked() {
std::vector<RowsetSharedPtr> candidate_rowsets;
- {
- std::shared_lock rlock(_meta_lock);
- for (auto& [v, rs] : _rs_version_map) {
- // MUST NOT compact rowset [0-1] for some historical reasons (see
cloud_schema_change)
- if (v.first != 0) {
- candidate_rowsets.push_back(rs);
- }
+ for (auto& [v, rs] : _rs_version_map) {
+ // MUST NOT compact rowset [0-1] for some historical reasons (see
cloud_schema_change)
+ if (v.first != 0) {
+ candidate_rowsets.push_back(rs);
}
}
std::sort(candidate_rowsets.begin(), candidate_rowsets.end(),
Rowset::comparator);
diff --git a/be/src/cloud/cloud_tablet.h b/be/src/cloud/cloud_tablet.h
index c279c2bdf32..84cfc325700 100644
--- a/be/src/cloud/cloud_tablet.h
+++ b/be/src/cloud/cloud_tablet.h
@@ -55,7 +55,7 @@ struct SyncRowsetStats {
int64_t tablet_meta_cache_miss {0};
int64_t bthread_schedule_delay_ns {0};
- int64_t meta_lock_wait_ns {0}; // _meta_lock (std::shared_mutex) wait
across all acquisitions
+ int64_t meta_lock_wait_ns {0}; // _meta_lock (BthreadSharedMutex) wait
across all acquisitions
int64_t sync_meta_lock_wait_ns {
0}; // _sync_meta_lock (bthread::Mutex) wait across all
acquisitions
};
@@ -152,19 +152,19 @@ public:
// If 'need_download_data_async' is true, it means that we need to
download the new version
// rowsets datum async.
void add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap,
- std::unique_lock<std::shared_mutex>& meta_lock,
+ std::unique_lock<BthreadSharedMutex>& meta_lock,
bool warmup_delta_data = false);
// MUST hold EXCLUSIVE `_meta_lock`.
void delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
- std::unique_lock<std::shared_mutex>& meta_lock);
+ std::unique_lock<BthreadSharedMutex>& meta_lock);
// Like delete_rowsets, but also removes edges from the version graph.
// Used by schema change to prevent the greedy capture algorithm from
// preferring stale compaction rowsets over individual SC output rowsets.
// MUST hold EXCLUSIVE `_meta_lock`.
void delete_rowsets_for_schema_change(const std::vector<RowsetSharedPtr>&
to_delete,
- std::unique_lock<std::shared_mutex>&
meta_lock,
+
std::unique_lock<BthreadSharedMutex>& meta_lock,
bool recycle_deleted_rowsets = true);
// Replace local rowsets in [2, alter_version] with schema change output
rowsets.
@@ -175,7 +175,7 @@ public:
// MUST hold EXCLUSIVE `_meta_lock`.
void replace_rowsets_with_schema_change_output(
const std::vector<RowsetSharedPtr>& output_rowsets, int64_t
alter_version,
- std::unique_lock<std::shared_mutex>& meta_lock, const char* stage,
+ std::unique_lock<BthreadSharedMutex>& meta_lock, const char* stage,
bool recycle_deleted_rowsets);
// When the tablet is dropped, we need to recycle cached data:
@@ -285,7 +285,8 @@ public:
_last_active_time_ms = time_ms;
}
- std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction();
+ // MUST hold SHARED `_meta_lock`.
+ std::vector<RowsetSharedPtr>
pick_candidate_rowsets_to_base_compaction_unlocked();
inline Version max_version() const {
std::shared_lock rdlock(_meta_lock);
@@ -294,7 +295,8 @@ public:
int64_t base_size() const { return _base_size; }
- std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_full_compaction();
+ // MUST hold SHARED `_meta_lock`.
+ std::vector<RowsetSharedPtr>
pick_candidate_rowsets_to_full_compaction_unlocked();
Result<RowsetSharedPtr> pick_a_rowset_for_index_change(int schema_version,
bool&
is_base_rowset);
Status check_rowset_schema_for_build_index(std::vector<TColumn>& columns,
int schema_version);
diff --git a/be/src/olap/base_tablet.cpp b/be/src/olap/base_tablet.cpp
index 523144734dd..33bf5dcc132 100644
--- a/be/src/olap/base_tablet.cpp
+++ b/be/src/olap/base_tablet.cpp
@@ -183,6 +183,10 @@ void BaseTablet::update_max_version_schema(const
TabletSchemaSPtr& tablet_schema
uint32_t BaseTablet::get_real_compaction_score() const {
std::shared_lock l(_meta_lock);
+ return get_real_compaction_score_unlocked();
+}
+
+uint32_t BaseTablet::get_real_compaction_score_unlocked() const {
const auto& rs_metas = _tablet_meta->all_rs_metas();
return std::accumulate(rs_metas.begin(), rs_metas.end(), 0, [](uint32_t
score, const auto& it) {
return score + it.second->get_compaction_score();
diff --git a/be/src/olap/base_tablet.h b/be/src/olap/base_tablet.h
index 994cdbe8bea..ff45ef4163a 100644
--- a/be/src/olap/base_tablet.h
+++ b/be/src/olap/base_tablet.h
@@ -34,6 +34,7 @@
#include "olap/tablet_meta.h"
#include "olap/tablet_schema.h"
#include "olap/version_graph.h"
+#include "util/bthread_shared_mutex.h"
#include "util/metrics.h"
namespace doris {
@@ -94,7 +95,7 @@ public:
int32_t max_version_config();
// FIXME(plat1ko): It is not appropriate to expose this lock
- std::shared_mutex& get_header_lock() { return _meta_lock; }
+ BthreadSharedMutex& get_header_lock() { return _meta_lock; }
void update_max_version_schema(const TabletSchemaSPtr& tablet_schema);
@@ -122,6 +123,10 @@ public:
// this method just return the compaction sum on each rowset
// note(tsy): we should unify the compaction score calculation finally
uint32_t get_real_compaction_score() const;
+ // MUST hold shared `_meta_lock`. Use this variant when the caller already
+ // holds the header lock to avoid recursively re-acquiring the (now
+ // writer-preferring) `_meta_lock`, which would self-deadlock.
+ uint32_t get_real_compaction_score_unlocked() const;
// MUST hold shared meta lock
Status capture_rs_readers_unlocked(const Versions& version_path,
@@ -366,7 +371,7 @@ protected:
Result<CaptureRowsetResult> _remote_capture_rowsets(const Version&
version_range) const;
- mutable std::shared_mutex _meta_lock;
+ mutable BthreadSharedMutex _meta_lock;
TimestampedVersionTracker _timestamped_version_tracker;
// After version 0.13, all newly created rowsets are saved in
_rs_version_map.
// And if rowset being compacted, the old rowsets will be saved in
_stale_rs_version_map;
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index cbc6ec0d67c..7b59858ec22 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -1391,7 +1391,7 @@ Status CompactionMixin::modify_rowsets() {
{
std::lock_guard<std::mutex>
wrlock_(tablet()->get_rowset_update_lock());
- std::lock_guard<std::shared_mutex>
wrlock(_tablet->get_header_lock());
+ std::lock_guard wrlock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
// Here we will calculate all the rowsets delete bitmaps which are
committed but not published to reduce the calculation pressure
@@ -1446,7 +1446,7 @@ Status CompactionMixin::modify_rowsets() {
RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets,
_input_rowsets, true));
}
} else {
- std::lock_guard<std::shared_mutex> wrlock(_tablet->get_header_lock());
+ std::lock_guard wrlock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets,
_input_rowsets, true));
}
diff --git a/be/src/olap/cumulative_compaction_time_series_policy.cpp
b/be/src/olap/cumulative_compaction_time_series_policy.cpp
index 9ef4db87c6c..8ade1d7e45e 100644
--- a/be/src/olap/cumulative_compaction_time_series_policy.cpp
+++ b/be/src/olap/cumulative_compaction_time_series_policy.cpp
@@ -156,9 +156,12 @@ uint32_t
TimeSeriesCumulativeCompactionPolicy::calc_cumulative_compaction_score(
}
// Condition 6: If there is a continuous set of empty rowsets, prioritize
merging.
+ // This is reached via Tablet::suitable_for_compaction() ->
_calc_cumulative_compaction_score(),
+ // which already holds a shared `_meta_lock`. Use the unlocked variant to
avoid recursively
+ // re-acquiring the (writer-preferring) header lock, which would
self-deadlock.
std::vector<RowsetSharedPtr> input_rowsets;
std::vector<RowsetSharedPtr> candidate_rowsets =
- tablet->pick_candidate_rowsets_to_cumulative_compaction();
+ tablet->pick_candidate_rowsets_to_cumulative_compaction_unlocked();
tablet->calc_consecutive_empty_rowsets(
&input_rowsets, candidate_rowsets,
tablet->tablet_meta()->time_series_compaction_empty_rowsets_threshold());
diff --git a/be/src/olap/full_compaction.cpp b/be/src/olap/full_compaction.cpp
index 38d2a3fd97a..98f318ebf9d 100644
--- a/be/src/olap/full_compaction.cpp
+++ b/be/src/olap/full_compaction.cpp
@@ -169,7 +169,7 @@ Status FullCompaction::modify_rowsets() {
} else {
DBUG_EXECUTE_IF("FullCompaction.modify_rowsets.before.block",
DBUG_BLOCK);
std::lock_guard<std::mutex>
rowset_update_wlock(tablet()->get_rowset_update_lock());
- std::lock_guard<std::shared_mutex>
meta_wlock(_tablet->get_header_lock());
+ std::lock_guard meta_wlock(_tablet->get_header_lock());
RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets,
_input_rowsets, true));
DBUG_EXECUTE_IF("FullCompaction.modify_rowsets.sleep", { sleep(5); })
tablet()->save_meta();
diff --git a/be/src/olap/rowset_builder.cpp b/be/src/olap/rowset_builder.cpp
index 185ba95e858..fd14ddf9ab1 100644
--- a/be/src/olap/rowset_builder.cpp
+++ b/be/src/olap/rowset_builder.cpp
@@ -125,7 +125,7 @@ void RowsetBuilder::_garbage_collection() {
}
Status BaseRowsetBuilder::init_mow_context(std::shared_ptr<MowContext>&
mow_context) {
- std::lock_guard<std::shared_mutex> lck(tablet()->get_header_lock());
+ std::lock_guard lck(tablet()->get_header_lock());
_max_version_in_flush_phase = tablet()->max_version_unlocked();
std::vector<RowsetSharedPtr> rowset_ptrs;
// tablet is under alter process. The delete bitmap will be calculated
after conversion.
diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp
index db1880e000a..78f4b6a6f54 100644
--- a/be/src/olap/schema_change.cpp
+++ b/be/src/olap/schema_change.cpp
@@ -961,7 +961,7 @@ Status SchemaChangeJob::_do_process_alter_tablet(const
TAlterTabletReqV2& reques
std::lock_guard new_tablet_lock(_new_tablet->get_push_lock());
std::lock_guard base_tablet_wlock(_base_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
- std::lock_guard<std::shared_mutex>
new_tablet_wlock(_new_tablet->get_header_lock());
+ std::lock_guard new_tablet_wlock(_new_tablet->get_header_lock());
do {
RowsetSharedPtr max_rowset;
@@ -1149,7 +1149,7 @@ Status SchemaChangeJob::_do_process_alter_tablet(const
TAlterTabletReqV2& reques
}
} else {
// set state to ready
- std::lock_guard<std::shared_mutex>
new_wlock(_new_tablet->get_header_lock());
+ std::lock_guard new_wlock(_new_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
res = _new_tablet->set_tablet_state(TabletState::TABLET_RUNNING);
if (!res) {
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index 6ed2a188193..a01d78e3605 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -485,7 +485,7 @@ Status Tablet::revise_tablet_meta(const
std::vector<RowsetSharedPtr>& to_add,
Status Tablet::add_rowset(RowsetSharedPtr rowset) {
DCHECK(rowset != nullptr);
- std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
+ std::lock_guard wrlock(_meta_lock);
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
// If the rowset already exist, just return directly. The rowset_id is an
unique-id,
// we can use it to check this situation.
@@ -695,7 +695,7 @@ RowsetSharedPtr Tablet::_rowset_with_largest_size() {
// add inc rowset should not persist tablet meta, because it will be persisted
when publish txn.
Status Tablet::add_inc_rowset(const RowsetSharedPtr& rowset) {
DCHECK(rowset != nullptr);
- std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
+ std::lock_guard wrlock(_meta_lock);
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
if (_contains_rowset(rowset->rowset_id())) {
return Status::OK();
@@ -731,7 +731,7 @@ void Tablet::delete_expired_stale_rowset() {
std::vector<std::pair<Version, std::vector<RowsetId>>>
deleted_stale_rowsets;
// hold write lock while processing stable rowset
{
- std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
+ std::lock_guard wrlock(_meta_lock);
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
// Compute the end time to delete rowsets, when a expired rowset
createtime less then this time, it will be deleted.
int64_t expired_stale_sweep_endtime =
@@ -913,7 +913,7 @@ void Tablet::delete_expired_stale_rowset() {
}
#ifndef BE_TEST
{
- std::shared_lock<std::shared_mutex> rlock(_meta_lock);
+ std::shared_lock rlock(_meta_lock);
save_meta();
}
#endif
@@ -1029,7 +1029,10 @@ uint32_t Tablet::calc_compaction_score() {
{
// Need meta lock, because it will iterator "all_rs_metas" of tablet
meta.
std::shared_lock rdlock(_meta_lock);
- int32_t score = get_real_compaction_score();
+ // Use the unlocked variant: we already hold a shared `_meta_lock`
here,
+ // and the lock is writer-preferring, so recursively re-acquiring it
+ // would self-deadlock if a writer queues in between.
+ int32_t score = get_real_compaction_score_unlocked();
if (_compaction_score > 0 && _compaction_score != score) {
LOG(WARNING) << "cumu cache score not equal real score, cache
score; "
<< _compaction_score << ", real score: " << score
@@ -1167,7 +1170,7 @@ void
Tablet::_max_continuous_version_from_beginning_unlocked(Version* version, V
}
void Tablet::calculate_cumulative_point() {
- std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
+ std::lock_guard wrlock(_meta_lock);
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
int64_t ret_cumulative_point;
_cumulative_compaction_policy->calculate_cumulative_point(
@@ -1225,12 +1228,17 @@ Status Tablet::_contains_version(const Version&
version) {
}
std::vector<RowsetSharedPtr>
Tablet::pick_candidate_rowsets_to_cumulative_compaction() {
+ std::shared_lock rlock(_meta_lock);
+ return pick_candidate_rowsets_to_cumulative_compaction_unlocked();
+}
+
+std::vector<RowsetSharedPtr>
Tablet::pick_candidate_rowsets_to_cumulative_compaction_unlocked() {
std::vector<RowsetSharedPtr> candidate_rowsets;
if (_cumulative_point == K_INVALID_CUMULATIVE_POINT) {
return candidate_rowsets;
}
- return _pick_visible_rowsets_to_compaction(_cumulative_point,
-
std::numeric_limits<int64_t>::max());
+ return _pick_visible_rowsets_to_compaction_unlocked(_cumulative_point,
+
std::numeric_limits<int64_t>::max());
}
std::vector<RowsetSharedPtr>
Tablet::pick_candidate_rowsets_to_base_compaction() {
@@ -1240,6 +1248,12 @@ std::vector<RowsetSharedPtr>
Tablet::pick_candidate_rowsets_to_base_compaction()
std::vector<RowsetSharedPtr> Tablet::_pick_visible_rowsets_to_compaction(
int64_t min_start_version, int64_t max_start_version) {
+ std::shared_lock rlock(_meta_lock);
+ return _pick_visible_rowsets_to_compaction_unlocked(min_start_version,
max_start_version);
+}
+
+std::vector<RowsetSharedPtr>
Tablet::_pick_visible_rowsets_to_compaction_unlocked(
+ int64_t min_start_version, int64_t max_start_version) {
auto [visible_version, update_ts] = get_visible_version_and_time();
bool update_time_long = MonotonicMillis() - update_ts >
config::compaction_keep_invisible_version_timeout_sec * 1000L;
@@ -1247,25 +1261,24 @@ std::vector<RowsetSharedPtr>
Tablet::_pick_visible_rowsets_to_compaction(
update_time_long ?
config::compaction_keep_invisible_version_min_count
:
config::compaction_keep_invisible_version_max_count;
+ // Caller MUST hold shared `_meta_lock`; do not re-acquire it here (the
lock
+ // is writer-preferring, recursive shared acquisition self-deadlocks).
std::vector<RowsetSharedPtr> candidate_rowsets;
- {
- std::shared_lock rlock(_meta_lock);
- for (const auto& [version, rs] : _rs_version_map) {
- int64_t version_start = version.first;
- // rowset is remote or rowset is not in given range
- if (!rs->is_local() || version_start < min_start_version ||
- version_start > max_start_version) {
- continue;
- }
+ for (const auto& [version, rs] : _rs_version_map) {
+ int64_t version_start = version.first;
+ // rowset is remote or rowset is not in given range
+ if (!rs->is_local() || version_start < min_start_version ||
+ version_start > max_start_version) {
+ continue;
+ }
- // can compact, met one of the conditions:
- // 1. had been visible;
- // 2. exceeds the limit of keep invisible versions.
- int64_t version_end = version.second;
- if (version_end <= visible_version ||
- version_end > visible_version + keep_invisible_version_limit) {
- candidate_rowsets.push_back(rs);
- }
+ // can compact, met one of the conditions:
+ // 1. had been visible;
+ // 2. exceeds the limit of keep invisible versions.
+ int64_t version_end = version.second;
+ if (version_end <= visible_version ||
+ version_end > visible_version + keep_invisible_version_limit) {
+ candidate_rowsets.push_back(rs);
}
}
std::sort(candidate_rowsets.begin(), candidate_rowsets.end(),
Rowset::comparator);
diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
index 372c141cc5c..15653877e81 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -200,7 +200,7 @@ public:
Versions calc_missed_versions(int64_t spec_version, Versions
existing_versions) const override;
// meta lock
- std::shared_mutex& get_header_lock() { return _meta_lock; }
+ BthreadSharedMutex& get_header_lock() { return _meta_lock; }
std::mutex& get_rowset_update_lock() { return _rowset_update_lock; }
std::mutex& get_push_lock() { return _ingest_lock; }
std::mutex& get_base_compaction_lock() { return _base_compaction_lock; }
@@ -282,6 +282,10 @@ public:
void check_tablet_path_exists();
std::vector<RowsetSharedPtr>
pick_candidate_rowsets_to_cumulative_compaction();
+ // MUST hold shared `_meta_lock`. Use this when the caller already holds
the
+ // header lock (e.g. the time-series cumulative score path under
+ // suitable_for_compaction) to avoid recursive shared acquisition.
+ std::vector<RowsetSharedPtr>
pick_candidate_rowsets_to_cumulative_compaction_unlocked();
std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction();
std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_full_compaction();
std::vector<RowsetSharedPtr>
pick_candidate_rowsets_to_build_inverted_index(
@@ -526,6 +530,9 @@ private:
std::vector<RowsetSharedPtr> _pick_visible_rowsets_to_compaction(int64_t
min_start_version,
int64_t
max_start_version);
+ // MUST hold shared `_meta_lock`.
+ std::vector<RowsetSharedPtr> _pick_visible_rowsets_to_compaction_unlocked(
+ int64_t min_start_version, int64_t max_start_version);
void _init_context_common_fields(RowsetWriterContext& context);
diff --git a/be/src/olap/tablet_manager.cpp b/be/src/olap/tablet_manager.cpp
index 38a6a4c06dd..a1e05882f36 100644
--- a/be/src/olap/tablet_manager.cpp
+++ b/be/src/olap/tablet_manager.cpp
@@ -559,7 +559,7 @@ Status TabletManager::_drop_tablet(TTabletId tablet_id,
TReplicaId replica_id, b
{
// drop tablet will update tablet meta, should lock
- std::lock_guard<std::shared_mutex>
wrlock(to_drop_tablet->get_header_lock());
+ std::lock_guard wrlock(to_drop_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
// NOTE: has to update tablet here, but must not update tablet meta
directly.
// because other thread may hold the tablet object, they may save meta
too.
diff --git a/be/src/olap/task/engine_clone_task.cpp
b/be/src/olap/task/engine_clone_task.cpp
index a3d6c5acb66..9f66146b95a 100644
--- a/be/src/olap/task/engine_clone_task.cpp
+++ b/be/src/olap/task/engine_clone_task.cpp
@@ -863,7 +863,7 @@ Status EngineCloneTask::_finish_clone(Tablet* tablet, const
std::string& clone_d
std::lock_guard
build_inverted_index_lock(tablet->get_build_inverted_index_lock());
std::lock_guard<std::mutex> push_lock(tablet->get_push_lock());
std::lock_guard<std::mutex> rwlock(tablet->get_rowset_update_lock());
- std::lock_guard<std::shared_mutex> wrlock(tablet->get_header_lock());
+ std::lock_guard wrlock(tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
if (is_incremental_clone) {
status = _finish_incremental_clone(tablet, cloned_tablet_meta,
version);
diff --git a/be/src/olap/task/index_builder.cpp
b/be/src/olap/task/index_builder.cpp
index 2744f953ec9..7551f26aab8 100644
--- a/be/src/olap/task/index_builder.cpp
+++ b/be/src/olap/task/index_builder.cpp
@@ -934,7 +934,7 @@ Status IndexBuilder::modify_rowsets(const
Merger::Statistics* stats) {
if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
_tablet->enable_unique_key_merge_on_write()) {
std::lock_guard<std::mutex>
rowset_update_wlock(_tablet->get_rowset_update_lock());
- std::lock_guard<std::shared_mutex>
meta_wlock(_tablet->get_header_lock());
+ std::lock_guard meta_wlock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
DeleteBitmapPtr delete_bitmap =
std::make_shared<DeleteBitmap>(_tablet->tablet_id());
for (auto i = 0; i < _input_rowsets.size(); ++i) {
@@ -956,7 +956,7 @@ Status IndexBuilder::modify_rowsets(const
Merger::Statistics* stats) {
// should call it after merge delete_bitmap
RETURN_IF_ERROR(_tablet->modify_rowsets(_output_rowsets,
_input_rowsets, true));
} else {
- std::lock_guard<std::shared_mutex> wrlock(_tablet->get_header_lock());
+ std::lock_guard wrlock(_tablet->get_header_lock());
RETURN_IF_ERROR(_tablet->modify_rowsets(_output_rowsets,
_input_rowsets, true));
}
diff --git a/be/src/util/bthread_shared_mutex.h
b/be/src/util/bthread_shared_mutex.h
new file mode 100644
index 00000000000..a993b5b7867
--- /dev/null
+++ b/be/src/util/bthread_shared_mutex.h
@@ -0,0 +1,125 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <bthread/condition_variable.h>
+#include <bthread/mutex.h>
+
+#include <climits>
+#include <mutex>
+
+namespace doris {
+
+// A reader-writer lock for bthread contexts. It is a port of libc++'s
+// std::shared_mutex (the two-gate condition-variable algorithm) onto
+// bthread::Mutex/bthread::ConditionVariable. Unlike std::shared_mutex
+// (pthread_rwlock_t), ownership carries no OS-thread identity, so it is safe
to
+// lock on one bthread worker and unlock on another after a bthread migrates.
+// Satisfies the C++ SharedMutex requirements (usable with std::unique_lock /
+// std::shared_lock). Writer-preferring.
+class BthreadSharedMutex {
+public:
+ BthreadSharedMutex() = default;
+ ~BthreadSharedMutex() = default;
+
+ BthreadSharedMutex(const BthreadSharedMutex&) = delete;
+ BthreadSharedMutex& operator=(const BthreadSharedMutex&) = delete;
+
+ void lock() {
+ std::unique_lock<bthread::Mutex> lk(_mutex);
+ while (_state & _write_entered) {
+ _gate1.wait(lk);
+ }
+ _state |= _write_entered;
+ while (_state & _n_readers) {
+ _gate2.wait(lk);
+ }
+ }
+
+ bool try_lock() {
+ std::unique_lock<bthread::Mutex> lk(_mutex);
+ if (_state == 0) {
+ _state = _write_entered;
+ return true;
+ }
+ return false;
+ }
+
+ void unlock() {
+ std::lock_guard<bthread::Mutex> lk(_mutex);
+ _state = 0;
+ // Notify while still holding `_mutex`. Releasing the mutex before
+ // notifying can lose a wakeup with bthread::ConditionVariable when the
+ // waiter is a pthread and the notifier is a bthread (or vice versa).
+ _gate1.notify_all();
+ }
+
+ void lock_shared() {
+ std::unique_lock<bthread::Mutex> lk(_mutex);
+ while ((_state & _write_entered) || (_state & _n_readers) ==
_n_readers) {
+ _gate1.wait(lk);
+ }
+ unsigned num_readers = (_state & _n_readers) + 1;
+ _state &= ~_n_readers;
+ _state |= num_readers;
+ }
+
+ bool try_lock_shared() {
+ std::unique_lock<bthread::Mutex> lk(_mutex);
+ unsigned num_readers = _state & _n_readers;
+ if (!(_state & _write_entered) && num_readers != _n_readers) {
+ ++num_readers;
+ _state &= ~_n_readers;
+ _state |= num_readers;
+ return true;
+ }
+ return false;
+ }
+
+ void unlock_shared() {
+ std::unique_lock<bthread::Mutex> lk(_mutex);
+ unsigned num_readers = (_state & _n_readers) - 1;
+ _state &= ~_n_readers;
+ _state |= num_readers;
+ // Notify while still holding `_mutex` (do not unlock first): with
+ // bthread::ConditionVariable a notify issued after releasing the mutex
+ // can be lost across pthread/bthread contexts, leaving a writer parked
+ // forever on a reader count that is already zero.
+ if (_state & _write_entered) {
+ if (num_readers == 0) {
+ _gate2.notify_one();
+ }
+ } else {
+ if (num_readers == _n_readers - 1) {
+ _gate1.notify_one();
+ }
+ }
+ }
+
+private:
+ bthread::Mutex _mutex;
+ bthread::ConditionVariable _gate1;
+ bthread::ConditionVariable _gate2;
+ unsigned _state {0};
+
+ // High bit: a writer has entered. Remaining bits: active reader count.
+ static constexpr unsigned _write_entered = 1U << (sizeof(unsigned) *
CHAR_BIT - 1);
+ static constexpr unsigned _n_readers = ~_write_entered;
+};
+
+} // namespace doris
diff --git a/be/test/cloud/cloud_empty_rowset_compaction_test.cpp
b/be/test/cloud/cloud_empty_rowset_compaction_test.cpp
index 5a3bb13ff76..b42465d7f3c 100644
--- a/be/test/cloud/cloud_empty_rowset_compaction_test.cpp
+++ b/be/test/cloud/cloud_empty_rowset_compaction_test.cpp
@@ -123,7 +123,7 @@ public:
version2_meta,
&version2_rowset);
EXPECT_TRUE(status.ok());
{
- std::unique_lock<std::shared_mutex>
lock(_tablet->get_header_lock());
+ std::unique_lock lock(_tablet->get_header_lock());
_tablet->add_rowsets({version2_rowset}, false, lock, false);
}
diff --git a/be/test/cloud/cloud_meta_mgr_test.cpp
b/be/test/cloud/cloud_meta_mgr_test.cpp
index aec1a5f21d8..593af5ba98b 100644
--- a/be/test/cloud/cloud_meta_mgr_test.cpp
+++ b/be/test/cloud/cloud_meta_mgr_test.cpp
@@ -266,12 +266,12 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_no_holes) {
// Add all rowsets to tablet
{
- std::unique_lock<std::shared_mutex> lock(tablet->get_header_lock());
+ std::unique_lock lock(tablet->get_header_lock());
tablet->add_rowsets(std::move(rowsets), false, lock, false);
}
// Test fill_version_holes directly - should not add any rowsets since
there are no holes
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 4, wlock);
EXPECT_TRUE(status.ok());
@@ -318,7 +318,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_with_holes) {
// Add all rowsets to tablet
{
- std::unique_lock<std::shared_mutex> lock(tablet->get_header_lock());
+ std::unique_lock lock(tablet->get_header_lock());
tablet->add_rowsets(std::move(rowsets), false, lock, false);
}
@@ -326,7 +326,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_with_holes) {
EXPECT_EQ(tablet->tablet_meta()->all_rs_metas().size(), 3);
// Test fill_version_holes directly to fill missing versions 1 and 3
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 4, wlock);
EXPECT_TRUE(status.ok());
@@ -434,7 +434,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_edge_cases) {
auto tablet =
std::make_shared<CloudTablet>(engine,
std::make_shared<TabletMeta>(*tablet_meta));
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 0, wlock);
EXPECT_TRUE(status.ok());
@@ -452,7 +452,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_edge_cases) {
auto tablet =
std::make_shared<CloudTablet>(engine,
std::make_shared<TabletMeta>(*tablet_meta));
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 5, wlock);
EXPECT_TRUE(status.ok());
@@ -494,7 +494,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_trailing_holes) {
// Add all rowsets to tablet
{
- std::unique_lock<std::shared_mutex> lock(tablet->get_header_lock());
+ std::unique_lock lock(tablet->get_header_lock());
tablet->add_rowsets(std::move(rowsets), false, lock, false);
}
@@ -502,7 +502,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_trailing_holes) {
EXPECT_EQ(tablet->tablet_meta()->all_rs_metas().size(), 3);
// Test fill_version_holes to fill trailing holes (versions 3, 4, 5)
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 5, wlock);
EXPECT_TRUE(status.ok());
@@ -569,7 +569,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_single_hole) {
// Add all rowsets to tablet
{
- std::unique_lock<std::shared_mutex> lock(tablet->get_header_lock());
+ std::unique_lock lock(tablet->get_header_lock());
tablet->add_rowsets(std::move(rowsets), false, lock, false);
}
@@ -577,7 +577,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_single_hole) {
EXPECT_EQ(tablet->tablet_meta()->all_rs_metas().size(), 2);
// Test fill_version_holes to fill single hole (version 1)
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 2, wlock);
EXPECT_TRUE(status.ok());
@@ -642,7 +642,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_multiple_consecutive_holes) {
// Add all rowsets to tablet
{
- std::unique_lock<std::shared_mutex> lock(tablet->get_header_lock());
+ std::unique_lock lock(tablet->get_header_lock());
tablet->add_rowsets(std::move(rowsets), false, lock, false);
}
@@ -650,7 +650,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_multiple_consecutive_holes) {
EXPECT_EQ(tablet->tablet_meta()->all_rs_metas().size(), 2);
// Test fill_version_holes to fill multiple consecutive holes (versions 1,
2, 3, 4)
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 5, wlock);
EXPECT_TRUE(status.ok());
@@ -717,7 +717,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_mixed_holes) {
// Add all rowsets to tablet
{
- std::unique_lock<std::shared_mutex> lock(tablet->get_header_lock());
+ std::unique_lock lock(tablet->get_header_lock());
tablet->add_rowsets(std::move(rowsets), false, lock, false);
}
@@ -725,7 +725,7 @@ TEST_F(CloudMetaMgrTest,
test_fill_version_holes_mixed_holes) {
EXPECT_EQ(tablet->tablet_meta()->all_rs_metas().size(), 4);
// Test fill_version_holes with max_version = 8 (should fill 1, 3, 4, 7, 8)
- std::unique_lock<std::shared_mutex> wlock(tablet->get_header_lock());
+ std::unique_lock wlock(tablet->get_header_lock());
Status status = meta_mgr.fill_version_holes(tablet.get(), 8, wlock);
EXPECT_TRUE(status.ok());
diff --git a/be/test/util/bthread_shared_mutex_test.cpp
b/be/test/util/bthread_shared_mutex_test.cpp
new file mode 100644
index 00000000000..8382ecdd006
--- /dev/null
+++ b/be/test/util/bthread_shared_mutex_test.cpp
@@ -0,0 +1,264 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "util/bthread_shared_mutex.h"
+
+#include <bthread/bthread.h>
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <shared_mutex>
+#include <thread>
+#include <vector>
+
+namespace doris {
+
+// 1) Basic SharedMutex contract. try_lock/try_lock_shared are non-blocking, so
+// this can be checked single-threaded.
+TEST(BthreadSharedMutexTest, BasicContract) {
+ BthreadSharedMutex m;
+
+ // Multiple shared holders can coexist.
+ ASSERT_TRUE(m.try_lock_shared());
+ ASSERT_TRUE(m.try_lock_shared());
+ // A writer cannot acquire while any reader is held.
+ ASSERT_FALSE(m.try_lock());
+ m.unlock_shared();
+ ASSERT_FALSE(m.try_lock()); // still one reader
+ m.unlock_shared();
+
+ // Exclusive ownership.
+ ASSERT_TRUE(m.try_lock());
+ ASSERT_FALSE(m.try_lock()); // exclusive: a second writer fails
+ ASSERT_FALSE(m.try_lock_shared()); // a reader is blocked while the writer
holds
+ m.unlock();
+
+ // Free again.
+ ASSERT_TRUE(m.try_lock_shared());
+ m.unlock_shared();
+}
+
+// 2) Writer-pending behavior of the two-gate algorithm: once a writer has set
+// the write-entered bit and is waiting for existing readers to drain, new
+// readers must block until the writer acquires and releases the lock.
+namespace {
+struct WriterPendingCtx {
+ BthreadSharedMutex* m;
+ std::atomic<bool> acquired {false};
+ std::atomic<bool> released {false};
+};
+
+void* writer_pending_fn(void* arg) {
+ auto* c = static_cast<WriterPendingCtx*>(arg);
+ c->m->lock(); // sets write-entered, then blocks until the reader drains
+ c->acquired.store(true);
+ bthread_usleep(20 * 1000);
+ c->m->unlock();
+ c->released.store(true);
+ return nullptr;
+}
+} // namespace
+
+TEST(BthreadSharedMutexTest, WriterPendingBlocksNewReaders) {
+ bthread_setconcurrency(8);
+ BthreadSharedMutex m;
+ m.lock_shared(); // reader R1 holds a shared lock
+
+ WriterPendingCtx ctx;
+ ctx.m = &m;
+ bthread_t w;
+ ASSERT_EQ(0, bthread_start_background(&w, nullptr, writer_pending_fn,
&ctx));
+
+ // Wait until the writer has reached lock() and set the write-entered bit,
+ // observed via try_lock_shared() beginning to fail (writer-preferring: a
new
+ // reader must not barge ahead of a pending writer). Probe instead of a
fixed
+ // sleep so a slow CI worker doesn't flake; bounded so a stuck writer fails
+ // the test rather than hanging. Successful probes are released
immediately so
+ // they don't add a lingering reader.
+ bool writer_pending = false;
+ for (int i = 0; i < 1000; ++i) { // up to ~10s
+ if (!m.try_lock_shared()) {
+ writer_pending = true;
+ break;
+ }
+ m.unlock_shared();
+ bthread_usleep(10 * 1000);
+ }
+ ASSERT_TRUE(writer_pending); // writer set _write_entered, blocking
new readers
+ ASSERT_FALSE(ctx.acquired.load()); // still waiting on R1 to drain
+
+ m.unlock_shared(); // drain R1 -> writer can now acquire
+ ASSERT_EQ(0, bthread_join(w, nullptr));
+ ASSERT_TRUE(ctx.acquired.load());
+ ASSERT_TRUE(ctx.released.load());
+
+ // The lock is usable afterwards.
+ ASSERT_TRUE(m.try_lock_shared());
+ m.unlock_shared();
+}
+
+// 3) Cross-thread unlock regression. Ownership is not tied to the OS thread
that
+// acquired the lock, so unlocking from a different thread is well defined
+// (with std::shared_mutex / pthread_rwlock_t this is undefined behavior and
+// can permanently wedge the lock).
+TEST(BthreadSharedMutexTest, CrossThreadUnlockExclusive) {
+ BthreadSharedMutex m;
+ m.lock(); // acquired on this thread
+ std::thread t([&] { m.unlock(); }); // released on a different OS thread
+ t.join();
+ // Not wedged: the lock can be acquired again.
+ ASSERT_TRUE(m.try_lock());
+ m.unlock();
+}
+
+TEST(BthreadSharedMutexTest, CrossThreadUnlockShared) {
+ BthreadSharedMutex m;
+ m.lock_shared();
+ std::thread t([&] { m.unlock_shared(); });
+ t.join();
+ // The reader was fully released, so a writer can acquire.
+ ASSERT_TRUE(m.try_lock());
+ m.unlock();
+}
+
+// 3b) bthread migration variant: hold the lock across a suspension point so
the
+// bthread may migrate to another worker pthread, and unlock from the
resumed
+// context (potentially a different OS thread). Many bthreads run
concurrently
+// to force migration; the test passing means the lock never wedges.
+namespace {
+struct MigrateCtx {
+ BthreadSharedMutex* m;
+ std::atomic<int>* done;
+ int iters;
+};
+
+void* migrate_fn(void* arg) {
+ auto* c = static_cast<MigrateCtx*>(arg);
+ for (int i = 0; i < c->iters; ++i) {
+ c->m->lock();
+ bthread_usleep(100); // suspension point while holding the write lock
+ c->m->unlock(); // may run on a different worker than lock()
+ c->m->lock_shared();
+ bthread_usleep(100);
+ c->m->unlock_shared();
+ }
+ c->done->fetch_add(1);
+ return nullptr;
+}
+} // namespace
+
+TEST(BthreadSharedMutexTest, BthreadMigrationNoWedge) {
+ bthread_setconcurrency(8);
+ BthreadSharedMutex m;
+ std::atomic<int> done {0};
+ MigrateCtx ctx {&m, &done, 100};
+
+ constexpr int kBthreads = 16;
+ std::vector<bthread_t> tids(kBthreads);
+ for (auto& t : tids) {
+ ASSERT_EQ(0, bthread_start_background(&t, nullptr, migrate_fn, &ctx));
+ }
+ for (auto t : tids) {
+ ASSERT_EQ(0, bthread_join(t, nullptr));
+ }
+ ASSERT_EQ(kBthreads, done.load()); // all finished -> no permanent wedge
+ ASSERT_TRUE(m.try_lock());
+ m.unlock();
+}
+
+// 4) Mixed reader/writer stress with invariant counters, also exercising the
+// RAII wrappers std::shared_lock / std::unique_lock (point 5).
+namespace {
+struct StressCtx {
+ BthreadSharedMutex* m;
+ std::atomic<int>* active_readers;
+ std::atomic<int>* active_writers;
+ std::atomic<bool>* failed;
+ int iters;
+ int id;
+};
+
+void* stress_fn(void* arg) {
+ auto* c = static_cast<StressCtx*>(arg);
+ for (int i = 0; i < c->iters; ++i) {
+ if ((c->id + i) % 4 == 0) { // ~25% writers
+ std::unique_lock<BthreadSharedMutex> wlock(*c->m);
+ int aw = c->active_writers->fetch_add(1) + 1;
+ // Invariant: at most one writer, and no readers while writing.
+ if (aw != 1 || c->active_readers->load() != 0) {
+ c->failed->store(true);
+ }
+ bthread_usleep(10);
+ c->active_writers->fetch_sub(1);
+ } else {
+ std::shared_lock<BthreadSharedMutex> rlock(*c->m);
+ c->active_readers->fetch_add(1);
+ // Invariant: no writer while readers are active.
+ if (c->active_writers->load() != 0) {
+ c->failed->store(true);
+ }
+ bthread_usleep(10);
+ c->active_readers->fetch_sub(1);
+ }
+ }
+ return nullptr;
+}
+} // namespace
+
+TEST(BthreadSharedMutexTest, MixedReaderWriterInvariants) {
+ bthread_setconcurrency(8);
+ BthreadSharedMutex m;
+ std::atomic<int> active_readers {0};
+ std::atomic<int> active_writers {0};
+ std::atomic<bool> failed {false};
+
+ constexpr int kBthreads = 16;
+ constexpr int kIters = 500;
+ std::vector<bthread_t> tids(kBthreads);
+ std::vector<StressCtx> ctxs(kBthreads);
+ for (int i = 0; i < kBthreads; ++i) {
+ ctxs[i] = {&m, &active_readers, &active_writers, &failed, kIters, i};
+ ASSERT_EQ(0, bthread_start_background(&tids[i], nullptr, stress_fn,
&ctxs[i]));
+ }
+ for (auto t : tids) {
+ ASSERT_EQ(0, bthread_join(t, nullptr));
+ }
+ ASSERT_FALSE(failed.load());
+ ASSERT_EQ(0, active_readers.load());
+ ASSERT_EQ(0, active_writers.load());
+}
+
+// 5) RAII drop-in coverage: BthreadSharedMutex must work with std::shared_lock
+// and std::unique_lock, which is what the call sites rely on.
+TEST(BthreadSharedMutexTest, RaiiWrappers) {
+ BthreadSharedMutex m;
+ {
+ std::shared_lock<BthreadSharedMutex> rlock(m);
+ ASSERT_FALSE(m.try_lock()); // writer blocked while shared lock held
+ }
+ {
+ std::unique_lock<BthreadSharedMutex> wlock(m);
+ ASSERT_FALSE(m.try_lock_shared()); // reader blocked while exclusive
lock held
+ }
+ // Adopt an already-held lock, mirroring the call-site pattern.
+ ASSERT_TRUE(m.try_lock());
+ { std::unique_lock<BthreadSharedMutex> wlock(m, std::adopt_lock); }
+ ASSERT_TRUE(m.try_lock_shared()); // released by the adopting guard
+ m.unlock_shared();
+}
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
