Skip to content

[fix](cloud) Avoid conflicts when recycling transactions with the same label #65746

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wyxxxcat wants to merge 1 commit into
base: master
Choose a base branch
from
+121 −54
Open

[fix](cloud) Avoid conflicts when recycling transactions with the same label #65746

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 66 additions & 40 deletions cloud/src/recycler/recycler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6382,7 +6382,7 @@ int InstanceRecycler::recycle_expired_txn_label() {
std::string end_recycle_txn_key;
recycle_txn_key(recycle_txn_key_info0, &begin_recycle_txn_key);
recycle_txn_key(recycle_txn_key_info1, &end_recycle_txn_key);
std::vector<std::string> recycle_txn_info_keys;
std::unordered_map<std::string, std::vector<std::string>> recycle_txn_keys_by_label;

LOG_WARNING("begin to recycle expired txn").tag("instance_id", instance_id_);

Expand Down Expand Up @@ -6423,7 +6423,17 @@ int InstanceRecycler::recycle_expired_txn_label() {
current_time_ms)) {
VLOG_DEBUG << "found recycle txn, key=" << hex(k);
num_expired++;
recycle_txn_info_keys.emplace_back(k);

std::string_view k1 = k;
k1.remove_prefix(1); // Remove key space
std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> out;
if (decode_key(&k1, &out) != 0) {
LOG_ERROR("failed to decode key").tag("key", hex(k));
return -1;
}
int64_t db_id = std::get<int64_t>(std::get<0>(out[3]));
auto label_key = txn_label_key({instance_id_, db_id, recycle_txn_pb.label()});
recycle_txn_keys_by_label[label_key].emplace_back(k);
}
return 0;
};
Expand Down Expand Up @@ -6532,35 +6542,48 @@ int InstanceRecycler::recycle_expired_txn_label() {

auto loop_done = [&]() -> int {
DORIS_CLOUD_DEFER {
recycle_txn_info_keys.clear();
recycle_txn_keys_by_label.clear();
};
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_info_keys",
&recycle_txn_info_keys);
for (const auto& k : recycle_txn_info_keys) {
concurrent_delete_executor.add([&]() {
int ret = delete_recycle_txn_kv(k);
if (ret == 1) {
const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times);
for (int i = 1; i <= max_retry; ++i) {
LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k);
ret = delete_recycle_txn_kv(k);
// clang-format off
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error", &ret);
// clang-format off
if (ret != 1) {
break;
}
// random sleep 0-100 ms to retry
std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100));
"InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_keys_by_label",
&recycle_txn_keys_by_label);
auto delete_recycle_txn_kv_with_retry = [&](const std::string& k) -> int {
int ret = delete_recycle_txn_kv(k);
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error",
&ret);
if (ret == 1) {
const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times);
for (int i = 1; i <= max_retry; ++i) {
LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k);
ret = delete_recycle_txn_kv(k);
TEST_SYNC_POINT_CALLBACK(
"InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_"
"error",
&ret);
if (ret != 1) {
break;
}
// random sleep 0-100 ms to retry
std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100));
}
if (ret != 0) {
LOG_WARNING("failed to delete recycle txn kv")
.tag("instance id", instance_id_)
.tag("key", hex(k));
return -1;
}
return ret;
};

for (auto& [label_key, txn_keys] : recycle_txn_keys_by_label) {
concurrent_delete_executor.add([&, txn_keys = std::move(txn_keys), label_key]() {
VLOG_DEBUG << "recycle txn label group, key=" << hex(label_key)
<< " txn_count=" << txn_keys.size();
for (const auto& k : txn_keys) {
int ret = delete_recycle_txn_kv_with_retry(k);
if (ret != 0) {
LOG_WARNING("failed to delete recycle txn kv")
.tag("instance id", instance_id_)
.tag("key", hex(k))
.tag("label_key", hex(label_key));
return -1;
}
}
return 0;
});
Expand Down Expand Up @@ -7322,8 +7345,8 @@ int InstanceRecycler::scan_and_statistics_rowsets() {
std::string recyc_rs_key0;
std::string recyc_rs_key1;
recycle_rowset_key(recyc_rs_key_info0, &recyc_rs_key0);
recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1);
int64_t earlest_ts = std::numeric_limits<int64_t>::max();
recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1);
int64_t earlest_ts = std::numeric_limits<int64_t>::max();

auto handle_rowset_kv = [&, this](std::string_view k, std::string_view v) -> int {
RecycleRowsetPB rowset;
Expand All @@ -7347,11 +7370,12 @@ int InstanceRecycler::scan_and_statistics_rowsets() {
metrics_context.total_need_recycle_num++;
metrics_context.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size();
segment_metrics_context_.total_need_recycle_num += rowset.rowset_meta().num_segments();
segment_metrics_context_.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size();
segment_metrics_context_.total_need_recycle_data_size +=
rowset.rowset_meta().total_disk_size();
return 0;
}

if(!rowset_meta->has_is_recycled() || !rowset_meta->is_recycled()) {
if (!rowset_meta->has_is_recycled() || !rowset_meta->is_recycled()) {
return 0;
}

Expand Down Expand Up @@ -7398,7 +7422,7 @@ int InstanceRecycler::scan_and_statistics_tmp_rowsets() {
DCHECK_GT(rowset.txn_id(), 0)
<< "txn_id=" << rowset.txn_id() << " rowset=" << rowset.ShortDebugString();

if(!rowset.has_is_recycled() || !rowset.is_recycled()) {
if (!rowset.has_is_recycled() || !rowset.is_recycled()) {
return 0;
}

Expand Down Expand Up @@ -7475,7 +7499,8 @@ int InstanceRecycler::scan_and_statistics_abort_timeout_txn() {
return 0;
};

int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key, std::move(handle_abort_timeout_txn_kv));
int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key,
std::move(handle_abort_timeout_txn_kv));
metrics_context.report(true);
return ret;
}
Expand Down Expand Up @@ -7509,7 +7534,8 @@ int InstanceRecycler::scan_and_statistics_expired_txn_label() {
return 0;
};

int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, std::move(handle_expired_txn_label_kv));
int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key,
std::move(handle_expired_txn_label_kv));
metrics_context.report(true);
return ret;
}
Expand Down Expand Up @@ -7740,7 +7766,7 @@ int InstanceRecycler::scan_and_statistics_restore_jobs() {
return 0;
}
metrics_context.total_need_recycle_num++;
if(restore_job_pb.need_recycle_data()) {
if (restore_job_pb.need_recycle_data()) {
scan_tablet_and_statistics(restore_job_pb.tablet_id(), metrics_context);
}
return 0;
Expand Down Expand Up @@ -7782,7 +7808,7 @@ void InstanceRecycler::scan_and_statistics_operation_logs() {

OperationLogReferenceInfo ref_info;
if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(),
&ref_info)) {
&ref_info)) {
metrics_context.total_need_recycle_num++;
metrics_context.total_need_recycle_data_size += operation_log.ByteSizeLong();
}
Expand Down Expand Up @@ -7938,8 +7964,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector<RowsetDeleteTask
std::string dbm_start_key =
meta_delete_bitmap_key({reference_instance_id, tablet_id, rowset_id, 0, 0});
std::string dbm_end_key = meta_delete_bitmap_key(
{reference_instance_id, tablet_id, rowset_id,
std::numeric_limits<int64_t>::max(), std::numeric_limits<int64_t>::max()});
{reference_instance_id, tablet_id, rowset_id, std::numeric_limits<int64_t>::max(),
std::numeric_limits<int64_t>::max()});
txn->remove(dbm_start_key, dbm_end_key);
LOG_INFO("remove delete bitmap kv in cleanup phase")
.tag("instance_id", instance_id_)
Expand All @@ -7962,8 +7988,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector<RowsetDeleteTask

// Remove versioned meta rowset key
if (!task.versioned_rowset_key.empty()) {
versioned::document_remove<RowsetMetaCloudPB>(
txn.get(), task.versioned_rowset_key, task.versionstamp);
versioned::document_remove<RowsetMetaCloudPB>(txn.get(), task.versioned_rowset_key,
task.versionstamp);
LOG_INFO("remove versioned meta rowset key in cleanup phase")
.tag("instance_id", instance_id_)
.tag("tablet_id", tablet_id)
Expand Down
69 changes: 55 additions & 14 deletions cloud/test/recycler_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8245,20 +8245,24 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_failure_test) {

auto txn_kv = mem_txn_kv;
ASSERT_TRUE(txn_kv.get()) << "exit get MemTxnKv error" << std::endl;
make_multiple_txn_info_kvs(txn_kv, 20000, 15000);
check_multiple_txn_info_kvs(txn_kv, 20000);
make_multiple_txn_info_kvs(txn_kv, 40000, 30000);
check_multiple_txn_info_kvs(txn_kv, 40000);

auto* sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
SyncPoint::get_instance()->clear_all_call_backs();
};
size_t recycle_txn_info_keys_cnt = 0;
sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_info_keys",
[&](auto&& args) {
auto* recycle_txn_info_keys =
try_any_cast<std::vector<std::string>*>(args[0]);
recycle_txn_info_keys_cnt += recycle_txn_info_keys->size();
});
size_t recycle_txn_keys_cnt = 0;
sp->set_call_back(
"InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_keys_by_label",
[&](auto&& args) {
auto* recycle_txn_keys_by_label =
try_any_cast<std::unordered_map<std::string, std::vector<std::string>>*>(
args[0]);
for (const auto& entry : *recycle_txn_keys_by_label) {
recycle_txn_keys_cnt += entry.second.size();
}
});
sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.failure", [](auto&& args) {
auto* ret = try_any_cast<int*>(args[0]);
*ret = -1;
Expand All @@ -8276,7 +8280,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_failure_test) {
std::cout << "recycle expired txn label cost="
<< std::chrono::duration_cast<std::chrono::milliseconds>(finish - start).count()
<< "ms" << std::endl;
check_multiple_txn_info_kvs(txn_kv, (20000 - recycle_txn_info_keys_cnt));
check_multiple_txn_info_kvs(txn_kv, (40000 - recycle_txn_keys_cnt));
}
TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) {
config::label_keep_max_second = 0;
Expand Down Expand Up @@ -8427,7 +8431,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) {
std::cout << "Update label after count: " << update_label_after_count << std::endl;
std::cout << "Transaction conflict count: " << txn_conflict_count << std::endl;

EXPECT_GT(txn_conflict_count, 0) << "txn_conflict sync point should be triggered";
EXPECT_EQ(txn_conflict_count, 0) << "txn conflicts should not occur within one label group";

std::unique_ptr<Transaction> verify_txn;
ASSERT_EQ(mem_txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK);
Expand Down Expand Up @@ -8455,7 +8459,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) {
}
}

TEST(RecyclerTest, recycle_txn_label_deal_with_conflict_error_test) {
TEST(RecyclerTest, recycle_txn_label_propagate_delete_error_test) {
config::label_keep_max_second = 0;
config::recycle_pool_parallelism = 20;

Expand Down Expand Up @@ -8601,10 +8605,47 @@ TEST(RecyclerTest, recycle_txn_label_deal_with_conflict_error_test) {
std::make_shared<TxnLazyCommitter>(mem_txn_kv));
ASSERT_EQ(recycler.init(), 0);

// deal with conflict but error during recycle
// Propagate a recycle error without relying on an internal label conflict.
ASSERT_EQ(recycler.recycle_expired_txn_label(), -1);
Comment thread
wyxxxcat marked this conversation as resolved.

EXPECT_GT(txn_conflict_count, 0) << "txn_conflict sync point should be triggered";
EXPECT_EQ(txn_conflict_count, 0) << "txn conflicts should not occur within one label group";
}

TEST(RecyclerTest, recycle_txn_label_retry_after_conflict_test) {
config::label_keep_max_second = 0;

auto mem_txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(mem_txn_kv->init(), 0);
make_single_txn_related_kvs(mem_txn_kv, 0, 1);

auto* sp = SyncPoint::get_instance();
DORIS_CLOUD_DEFER {
SyncPoint::get_instance()->clear_all_call_backs();
};

std::atomic<int> delete_attempt_count {0};
sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error",
[&](auto&& args) {
auto* ret = try_any_cast<int*>(args[0]);
const int attempt = delete_attempt_count.fetch_add(1);
*ret = attempt == 0 ? 1 : 0;
});
sp->enable_processing();

InstanceInfoPB instance;
instance.set_instance_id(instance_id);
InstanceRecycler recycler(mem_txn_kv, instance, thread_group,
std::make_shared<TxnLazyCommitter>(mem_txn_kv));
ASSERT_EQ(recycler.init(), 0);

ASSERT_EQ(recycler.recycle_expired_txn_label(), 0);
EXPECT_EQ(delete_attempt_count, 2);

std::unique_ptr<Transaction> verify_txn;
ASSERT_EQ(mem_txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK);
const std::string recycle_key = recycle_txn_key({instance_id, 0, 1000000});
std::string recycle_value;
EXPECT_EQ(verify_txn->get(recycle_key, &recycle_value), TxnErrorCode::TXN_KEY_NOT_FOUND);
}

TEST(RecyclerTest, recycle_restore_job_complete_state) {
Expand Down
Loading