This is an automated email from the ASF dual-hosted git repository.
liaoxin01 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 48d9e57b5e5 [fix](cloud) Use bthread-aware shared mutex for tablet
header lock (#64574)
48d9e57b5e5 is described below
commit 48d9e57b5e5e3a914216a9582d82d6fc38563e0d
Author: Xin Liao <[email protected]>
AuthorDate: Thu Jun 25 15:49:48 2026 +0800
[fix](cloud) Use bthread-aware shared mutex for tablet header lock (#64574)
## Problem
The tablet header lock (`_meta_lock`) is a `std::shared_mutex`, which
under libstdc++ wraps `pthread_rwlock_t` and is **thread-affine**:
unlocking it from an OS thread other than the one that acquired it is
undefined behavior.
In cloud mode the header **write** lock can be held across a
**suspending** call. Concretely, when a query-driven rowset sync pulls
an overlapping (compacted) rowset, `CloudTablet::add_rowsets` warms up
the new remote file **while holding the write lock**; on a
cold-restarted BE, resolving the storage vault / building the S3 client
issues a meta-service RPC. The holding bthread suspends and may
**migrate to another worker pthread**, so the matching unlock runs on a
**different OS thread**. This corrupts the glibc rwlock (the
write-locked bit is left set with no owner), **permanently wedging** the
lock — all readers/writers on that tablet pile up and queries time out
(~90s).
Observed in graceful-restart tests: a tablet's header lock stuck with
`active_writer=[none]` yet every `try_lock`/`try_lock_shared` failing
for >70 min, and the last writer's acquire OS-tid differing from its
release OS-tid (proof of the cross-thread unlock).
## Fix
Replace `_meta_lock` with `BthreadSharedMutex`, a port of libc++'s
`std::shared_mutex` (the two-gate condition-variable algorithm) onto
`bthread::Mutex` / `bthread::ConditionVariable`:
- Ownership is an integer state guarded by a briefly held internal mutex
and carries **no OS-thread identity**, so locking on one worker and
unlocking on another after a bthread migration is well defined — the
permanent wedge can no longer happen.
- Waiting blocks on a bthread condition variable, suspending the bthread
instead of blocking the worker.
- Writer-preferring; satisfies the C++ SharedMutex requirements, so it
is a drop-in with `std::unique_lock` / `std::shared_lock`.
Only the tablet header lock is switched; unrelated `std::shared_mutex`
members (e.g. `TabletMeta::_meta_lock`) are left untouched. Call sites
that named the type explicitly are converted to class template argument
deduction or to `BthreadSharedMutex`.
---
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 | 51 ++--
be/src/cloud/cloud_tablet.h | 16 +-
be/src/storage/compaction/binlog_compaction.cpp | 2 +-
be/src/storage/compaction/compaction.cpp | 4 +-
.../cumulative_compaction_time_series_policy.cpp | 5 +-
be/src/storage/compaction/full_compaction.cpp | 2 +-
be/src/storage/rowset_builder.cpp | 2 +-
be/src/storage/schema_change/schema_change.cpp | 4 +-
be/src/storage/tablet/base_tablet.cpp | 4 +
be/src/storage/tablet/base_tablet.h | 9 +-
be/src/storage/tablet/tablet.cpp | 65 +++--
be/src/storage/tablet/tablet.h | 9 +-
be/src/storage/tablet/tablet_manager.cpp | 2 +-
be/src/storage/task/engine_clone_task.cpp | 2 +-
be/src/storage/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/cloud/cloud_tablet_test.cpp | 2 +-
be/test/util/bthread_shared_mutex_test.cpp | 264 +++++++++++++++++++++
25 files changed, 519 insertions(+), 95 deletions(-)
diff --git a/be/src/cloud/cloud_base_compaction.cpp
b/be/src/cloud/cloud_base_compaction.cpp
index f6e98e8fb31..33fca48daed 100644
--- a/be/src/cloud/cloud_base_compaction.cpp
+++ b/be/src/cloud/cloud_base_compaction.cpp
@@ -189,7 +189,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 e7229e6f12d..d136944ae4b 100644
--- a/be/src/cloud/cloud_cumulative_compaction.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction.cpp
@@ -503,7 +503,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 56d81e98a9f..12ea9c87c99 100644
--- a/be/src/cloud/cloud_full_compaction.cpp
+++ b/be/src/cloud/cloud_full_compaction.cpp
@@ -128,7 +128,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 4eb56aa4237..479e9fb153d 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -618,7 +618,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);
}
@@ -2367,7 +2367,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 eb2a0f1ab9c..657265594bc 100644
--- a/be/src/cloud/cloud_meta_mgr.h
+++ b/be/src/cloud/cloud_meta_mgr.h
@@ -160,7 +160,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 3522871f557..7827c438bf9 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -187,7 +187,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); });
@@ -239,7 +242,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.
@@ -265,7 +271,8 @@ Result<std::vector<Version>>
CloudTablet::capture_versions_with_freshness_tolera
std::ranges::any_of(std::views::values(_tablet_meta->all_rs_metas()), check_fn)
||
std::ranges::any_of(std::views::values(_tablet_meta->all_stale_rs_metas()),
check_fn);
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
@@ -411,7 +418,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;
@@ -493,7 +500,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;
}
@@ -514,7 +521,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;
@@ -546,7 +553,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) {
@@ -895,7 +902,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;
@@ -1156,30 +1163,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);
@@ -2051,7 +2052,7 @@ void CloudTablet::apply_visible_pending_rowsets() {
Defer defer {[&] { clear_unused_visible_pending_rowsets(); }};
std::unique_lock lock(_sync_meta_lock);
- std::unique_lock<std::shared_mutex> meta_wlock(_meta_lock);
+ std::unique_lock meta_wlock(_meta_lock);
int64_t next_version = _max_version + 1;
std::vector<RowsetSharedPtr> to_add;
std::lock_guard<std::mutex> pending_lock(_visible_pending_rs_lock);
diff --git a/be/src/cloud/cloud_tablet.h b/be/src/cloud/cloud_tablet.h
index 93ffc40488d..7fc3dec1451 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/storage/compaction/binlog_compaction.cpp
b/be/src/storage/compaction/binlog_compaction.cpp
index b41782e4df9..7ee72cf7615 100644
--- a/be/src/storage/compaction/binlog_compaction.cpp
+++ b/be/src/storage/compaction/binlog_compaction.cpp
@@ -149,7 +149,7 @@ Status BinlogCompaction::modify_rowsets() {
_output_rowset);
{
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);
RETURN_IF_ERROR(tablet()->modify_row_binlog_rowsets(output_rowsets,
_input_rowsets));
}
diff --git a/be/src/storage/compaction/compaction.cpp
b/be/src/storage/compaction/compaction.cpp
index ff87dbd6764..df2fee8b114 100644
--- a/be/src/storage/compaction/compaction.cpp
+++ b/be/src/storage/compaction/compaction.cpp
@@ -1504,7 +1504,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
@@ -1559,7 +1559,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/storage/compaction/cumulative_compaction_time_series_policy.cpp
b/be/src/storage/compaction/cumulative_compaction_time_series_policy.cpp
index ab23c43904e..af023f6beca 100644
--- a/be/src/storage/compaction/cumulative_compaction_time_series_policy.cpp
+++ b/be/src/storage/compaction/cumulative_compaction_time_series_policy.cpp
@@ -154,9 +154,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/storage/compaction/full_compaction.cpp
b/be/src/storage/compaction/full_compaction.cpp
index 9d730e0856c..260badefe4c 100644
--- a/be/src/storage/compaction/full_compaction.cpp
+++ b/be/src/storage/compaction/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/storage/rowset_builder.cpp
b/be/src/storage/rowset_builder.cpp
index 1be1d0873dc..bab3016dd02 100644
--- a/be/src/storage/rowset_builder.cpp
+++ b/be/src/storage/rowset_builder.cpp
@@ -147,7 +147,7 @@ void RowsetBuilder::_garbage_collection(bool cancel_txn) {
Status BaseRowsetBuilder::init_mow_context(std::shared_ptr<MowContext>&
mow_context) {
DCHECK(is_data_builder());
- 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/storage/schema_change/schema_change.cpp
b/be/src/storage/schema_change/schema_change.cpp
index 13082f89401..8b51e539089 100644
--- a/be/src/storage/schema_change/schema_change.cpp
+++ b/be/src/storage/schema_change/schema_change.cpp
@@ -978,7 +978,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;
@@ -1174,7 +1174,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/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index 78088e394d1..09f73ec43d2 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -188,6 +188,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/storage/tablet/base_tablet.h
b/be/src/storage/tablet/base_tablet.h
index cf8af9a5a27..a835e2465ee 100644
--- a/be/src/storage/tablet/base_tablet.h
+++ b/be/src/storage/tablet/base_tablet.h
@@ -35,6 +35,7 @@
#include "storage/tablet/tablet_meta.h"
#include "storage/tablet/tablet_schema.h"
#include "storage/version_graph.h"
+#include "util/bthread_shared_mutex.h"
namespace doris {
struct RowSetSplits;
@@ -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);
@@ -127,6 +128,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,
@@ -372,7 +377,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;
TimestampedVersionTracker _row_binlog_version_tracker;
diff --git a/be/src/storage/tablet/tablet.cpp b/be/src/storage/tablet/tablet.cpp
index d782574638b..84b083d112b 100644
--- a/be/src/storage/tablet/tablet.cpp
+++ b/be/src/storage/tablet/tablet.cpp
@@ -529,7 +529,7 @@ Status Tablet::revise_tablet_meta(const
std::vector<RowsetSharedPtr>& to_add,
Status Tablet::add_rowset(RowsetSharedPtr rowset, RowsetSharedPtr
row_binlog_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.
@@ -845,7 +845,7 @@ RowsetSharedPtr Tablet::_rowset_with_largest_size() {
Status Tablet::add_inc_rowset(const RowsetSharedPtr& rowset,
const RowsetSharedPtr& row_binlog_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())) {
// Ensure binlog<row> is also added on retry.
@@ -882,7 +882,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 =
@@ -1064,7 +1064,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
@@ -1191,7 +1191,10 @@ uint32_t Tablet::calc_compaction_score(CompactionType
compaction_type,
{
// 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
@@ -1208,7 +1211,7 @@ bool Tablet::suitable_for_compaction(
#ifndef BE_TEST
if (compaction_type == CompactionType::CUMULATIVE_COMPACTION &&
cumulative_compaction_policy != nullptr) {
- std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
+ std::lock_guard wrlock(_meta_lock);
if (_cumulative_compaction_policy == nullptr ||
_cumulative_compaction_policy->name() !=
cumulative_compaction_policy->name()) {
_cumulative_compaction_policy = cumulative_compaction_policy;
@@ -1338,7 +1341,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(
@@ -1396,12 +1399,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() {
@@ -1411,6 +1419,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;
@@ -1418,25 +1432,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/storage/tablet/tablet.h b/be/src/storage/tablet/tablet.h
index e4f2b32e964..918d892fdeb 100644
--- a/be/src/storage/tablet/tablet.h
+++ b/be/src/storage/tablet/tablet.h
@@ -214,7 +214,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; }
@@ -310,6 +310,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_binlog_compaction();
@@ -577,6 +581,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/storage/tablet/tablet_manager.cpp
b/be/src/storage/tablet/tablet_manager.cpp
index 4371864ce49..f53ff7a9017 100644
--- a/be/src/storage/tablet/tablet_manager.cpp
+++ b/be/src/storage/tablet/tablet_manager.cpp
@@ -570,7 +570,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/storage/task/engine_clone_task.cpp
b/be/src/storage/task/engine_clone_task.cpp
index f8cac18b04a..e6fe3d3031d 100644
--- a/be/src/storage/task/engine_clone_task.cpp
+++ b/be/src/storage/task/engine_clone_task.cpp
@@ -951,7 +951,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, copy_row_binlog);
diff --git a/be/src/storage/task/index_builder.cpp
b/be/src/storage/task/index_builder.cpp
index b7626d553ed..ef49626e143 100644
--- a/be/src/storage/task/index_builder.cpp
+++ b/be/src/storage/task/index_builder.cpp
@@ -931,7 +931,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) {
@@ -953,7 +953,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 63794bf5ec0..8fb67d04a06 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 ff4539f5969..a6149cae9c5 100644
--- a/be/test/cloud/cloud_meta_mgr_test.cpp
+++ b/be/test/cloud/cloud_meta_mgr_test.cpp
@@ -204,12 +204,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());
@@ -256,7 +256,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);
}
@@ -264,7 +264,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());
@@ -372,7 +372,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());
@@ -390,7 +390,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());
@@ -432,7 +432,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);
}
@@ -440,7 +440,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());
@@ -507,7 +507,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);
}
@@ -515,7 +515,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());
@@ -580,7 +580,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);
}
@@ -588,7 +588,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());
@@ -655,7 +655,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);
}
@@ -663,7 +663,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/cloud/cloud_tablet_test.cpp
b/be/test/cloud/cloud_tablet_test.cpp
index 23045f20d5e..228e7421ee3 100644
--- a/be/test/cloud/cloud_tablet_test.cpp
+++ b/be/test/cloud/cloud_tablet_test.cpp
@@ -1006,7 +1006,7 @@ public:
}
void add_initial_rowsets(const std::vector<RowsetSharedPtr>& rowsets) {
- std::unique_lock<std::shared_mutex>
meta_wlock(_tablet->get_header_lock());
+ std::unique_lock meta_wlock(_tablet->get_header_lock());
_tablet->add_rowsets(std::vector<RowsetSharedPtr>(rowsets), false,
meta_wlock, false);
}
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]
