Skip to content

SDSTOR-24056: hold lock when updating superblk of raft_repl_dev #898

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

Merged
JacksonYao287 merged 1 commit into from
Jul 20, 2026
Merged

SDSTOR-24056: hold lock when updating superblk of raft_repl_dev #898

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
2 changes: 1 addition & 1 deletion conanfile.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

class HomestoreConan(ConanFile):
name = "homestore"
version = "7.5.12"
version = "7.5.13"

homepage = "https://github.com/eBay/Homestore"
description = "HomeStore Storage Engine"
Expand Down
2 changes: 1 addition & 1 deletion src/include/homestore/replication/repl_decls.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ VENUM(ReplServiceError, int32_t,
NO_SPACE_LEFT = -20000,
DRIVE_WRITE_ERROR = -20001,
DATA_DUPLICATED = -20002,
QUIENCE_STATE = -20003,
QUIESCENCE_STATE = -20003,
QUORUM_NOT_MET = -20004,
REPLACE_MEMBER_TASK_MISMATCH = -20005,
UNREADY_STATE = -20006,
Expand Down
10 changes: 5 additions & 5 deletions src/lib/index/wb_cache.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -627,23 +627,23 @@ void IndexWBCache::recover(sisl::byte_view sb) {
// (Up) buffer is not committed, node need to be kept and (potentially) repaired later
if (buf->m_created_cp_id != icp_ctx->id()) {
LOGTRACEMOD(wbcache,
"NOT FREE committing buffer {} node deleted is false reason: node commited?= {} "
"NOT FREE committing buffer {} node deleted is false reason: node committed?= {} "
"up committed? {}",
buf->to_string(), was_node_committed(buf), was_node_committed(buf->m_up_buffer));
buf->m_node_freed = false;
r_cast< persistent_hdr_t* >(buf->m_bytes)->node_deleted = false;
m_vdev->commit_blk(buf->m_blkid);
// it can happen when children moved to one of right parent sibling and then the previous node is
// deleted but not commited during crash (upbuffer is not committed). but its children already
// deleted but not committed during crash (upbuffer is not committed). but its children already
// committed. and freed (or changed)
if (buf->m_node_level) { potential_parent_recovered_bufs.insert(buf); }
} else {
LOGINFO("deleting and creating new buf {}", buf->to_string());
deleted_bufs.push_back(buf);
}
// 1- upbuffer was dirtied by the same cp, so it is not commited, so we don't need to repair it.
// 1- upbuffer was dirtied by the same cp, so it is not committed, so we don't need to repair it.
// remove it from down_waiting list (probably recursively going up) 2- upbuffer was created and
// freed at the same cp, so it is not commited, so we don't need to repair it.
// freed at the same cp, so it is not committed, so we don't need to repair it.
if (buf->m_up_buffer) {
LOGTRACEMOD(wbcache, "remove_down_buffer {} from up buffer {}", buf->to_string(),
buf->m_up_buffer->to_string());
Expand All @@ -656,7 +656,7 @@ void IndexWBCache::recover(sisl::byte_view sb) {
LOGTRACEMOD(wbcache, "recovering new buf {}", buf->to_string());
// New node
if (was_node_committed(buf) && was_node_committed(buf->m_up_buffer)) {
// Both current and up buffer is commited, we can safely commit the current block
// Both current and up buffer is committed, we can safely commit the current block
LOGTRACEMOD(wbcache, "New buffer {} and the up buffer {} are committed", buf->to_string(),
buf->m_up_buffer->to_string());
m_vdev->commit_blk(buf->m_blkid);
Expand Down
68 changes: 36 additions & 32 deletions src/lib/replication/repl_dev/raft_repl_dev.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ RaftReplDev::RaftReplDev(RaftReplService& svc, superblk< raft_repl_dev_superblk
m_identify_str = m_rdev_name + ":" + group_id_str();

RD_LOGI(NO_TRACE_ID,
"Started {} RaftReplDev group_id={}, replica_id={}, raft_server_id={} commited_lsn={}, "
"Started {} RaftReplDev group_id={}, replica_id={}, raft_server_id={} committed_lsn={}, "
"compact_lsn={}, checkpoint_lsn:{}, next_dsn={} "
"log_dev={} log_store={}",
(load_existing ? "Existing" : "New"), group_id_str(), my_replica_id_str(), m_raft_server_id,
Expand Down Expand Up @@ -226,7 +226,8 @@ AsyncReplResult<> RaftReplDev::start_replace_member(std::string& task_id, const
// until the successor finishes the catch-up of the latest log, and then resign. Return NOT_LEADER and let
// client retry.
if (is_leader) {
RD_LOGI(trace_id, "Step1. Replace member, leader is the member_out so yield leadership, task_id={}", task_id);
RD_LOGI(trace_id, "Step1. Replace member, leader is the member_out so yield leadership, task_id={}",
task_id);
raft_server()->yield_leadership(false /* immediate */, -1 /* successor */);
}
RD_LOGE(trace_id, "Step1. Replace member, I am not leader, can not handle the request, task_id={}", task_id);
Expand Down Expand Up @@ -544,7 +545,7 @@ ReplServiceError RaftReplDev::do_add_member(const replica_member_info& member, u
RD_LOGW(trace_id, "Ignoring error returned from nuraft add_member, member={}, err={}",
boost::uuids::to_string(member.id), ret);
} else if (ret == nuraft::cmd_result_code::CANCELLED) {
// nuraft mesg will return cancelled if the change is not commited after waiting for
// nuraft mesg will return cancelled if the change is not committed after waiting for
// raft_leader_change_timeout_ms(default 3200).
RD_LOGE(trace_id, "Add member failed, member={}, err={}", boost::uuids::to_string(member.id), ret);
return ReplServiceError::CANCELLED;
Expand Down Expand Up @@ -821,7 +822,7 @@ ReplServiceError RaftReplDev::set_priority(const replica_id_t& member, int32_t p
auto priority_ret = raft_server()->set_priority(nuraft_mesg::to_server_id(member), priority);
// Set_priority should be handled by leader, but if the intent is to set the leader's priority to 0, it returns
// BROADCAST. In this case return NOT_LEADER to let client retry new leader.
// If there is an uncommited_config, nuraft set_priority will honor this uncommited config and generate new
// If there is an uncommitted_config, nuraft set_priority will honor this uncommitted config and generate new
// config based on it and won't have config_changing error.
if (priority_ret != nuraft::raft_server::PrioritySetResult::SET) {
RD_LOGE(trace_id, "Propose to raft to set priority failed, result: {}",
Expand Down Expand Up @@ -851,7 +852,7 @@ folly::SemiFuture< ReplServiceError > RaftReplDev::destroy_group() {
// will become a garbage rreq in this follower. now if we trigger a destroy_group, a new rreq {originator=1, term=1,
// dsn=0} will created in the follower since the default dsn of a repl_key is 0.after the log of this rreq is
// appended to log store and get a new lsn, if we link the new lsn to the old rreq (rreq is identified by
// {originator, term, dsn}) which has alread have a lsn, then a assert will be throw out. pls refer to
// {originator, term, dsn}) which has already have a lsn, then a assert will be throw out. pls refer to
// repl_req_ctx::set_lsn

// here, we set the dsn to a new one , which is definitely unique in the follower, so that the new rreq will not
Expand All @@ -865,7 +866,7 @@ folly::SemiFuture< ReplServiceError > RaftReplDev::destroy_group() {

if (err != ReplServiceError::OK) {
// Failed to initialize the repl_req_ctx for replace member.
LOGERROR("Failed to initialize repl_req_ctx for destorying group, error={}", err);
LOGERROR("Failed to initialize repl_req_ctx for destroying group, error={}", err);
return folly::makeSemiFuture< ReplServiceError >(std::move(err));
}

Expand Down Expand Up @@ -1276,7 +1277,7 @@ repl_req_ptr_t RaftReplDev::applier_create_req(repl_key const& rkey, journal_typ
} else {
RD_LOGD(
rkey.traceID,
"For Repl_key=[{}] alloc hints returned error={}, failing this req, data_channl: {}, is_proposer: {} ",
"For Repl_key=[{}] alloc hints returned error={}, failing this req, data_channel: {}, is_proposer: {} ",
rkey.to_string(), status, is_data_channel, rreq->is_proposer());
}
// Do not call handle_error here, because handle_error is for rreq which needs to be terminated. This one can be
Expand Down Expand Up @@ -1761,7 +1762,7 @@ void RaftReplDev::handle_config_commit(const repl_lsn_t lsn, raft_cluster_config
(void)new_conf;
auto prev_lsn = m_commit_upto_lsn.load(std::memory_order_relaxed);
if (prev_lsn >= lsn || !m_commit_upto_lsn.compare_exchange_strong(prev_lsn, lsn)) {
RD_LOGE(NO_TRACE_ID, "Raft Channel: unexpected log {} commited before config {} committed", prev_lsn, lsn);
RD_LOGE(NO_TRACE_ID, "Raft Channel: unexpected log {} committed before config {} committed", prev_lsn, lsn);
}
}

Expand Down Expand Up @@ -2105,7 +2106,7 @@ std::set< replica_id_t > RaftReplDev::get_active_peers() const {
? my_committed_idx - HS_DYNAMIC_CONFIG(consensus.laggy_threshold)
: 0;
// peer's last log idx should also >= leader's start_index-1(ensure existence), otherwise leader can't append log
// entries to it and baseline resync will be triggerred. Try to avoid conflict between baseline resync and normal
// entries to it and baseline resync will be triggered. Try to avoid conflict between baseline resync and normal
// replication.
least_active_repl_idx = std::max(least_active_repl_idx, m_data_journal->start_index() - 1);
for (auto p : repl_status) {
Expand Down Expand Up @@ -2334,7 +2335,7 @@ void RaftReplDev::leave() {
// 2. it is removed from the cluster and the new config(excluding this node) is being committed on this node
// 3. it is removed from the cluster , but the node is down and new config log(excluding this node) is not
// replicated to this removed node. when the node restart, leader will not send any append entry to this node,
// since it is not a member of the raft group. it will become a condidate and send request-vote request to other
// since it is not a member of the raft group. it will become a candidate and send request-vote request to other
// members of this raft group. a member will send RemovedFromCluster to the node if this member finds the node
// is no longer a member of the raft group.

Expand All @@ -2348,8 +2349,11 @@ void RaftReplDev::leave() {

// Persist that destroy pending in superblk, so that in case of crash before cleanup of resources, it can be done
// post restart.
m_rd_sb->destroy_pending = 0x1;
m_rd_sb.write();
{
std::unique_lock lg{m_sb_mtx};
m_rd_sb->destroy_pending = 0x1;
m_rd_sb.write();
}

RD_LOGI(NO_TRACE_ID, "RaftReplDev leave group_id={}", group_id_str());
m_destroy_promise.setValue(ReplServiceError::OK); // In case proposer is waiting for the destroy to complete
Expand All @@ -2374,10 +2378,10 @@ nuraft::cb_func::ReturnCode RaftReplDev::raft_event(nuraft::cb_func::Type type,
// T1: L1 is leader, push data to follower F1, F1 generate rreq1 and allocate blk for this data. let`s say the
// lsn of this data is lsn-100

// T2: leader switchs to L2, L2 send log to F1, F1 generate rreq2 and try to allocate blk but got
// T2: leader switches to L2, L2 send log to F1, F1 generate rreq2 and try to allocate blk but got
// no_space_left. then it will set the quiesce flag and waiting for the commit of lsn-99 (100 - 1).

// T3: leader switchs back to L1, L1 send log (lsn-100) to F1, when F1 tries to create rreq for lsn-100, rreq1
// T3: leader switches back to L1, L1 send log (lsn-100) to F1, when F1 tries to create rreq for lsn-100, rreq1
// (which is created at T1) will be found since the rkey{server,term,dsn} is same as rreq1. so F1 will skip
// creating a new rreq. since the data for lsn-100 is already in written at T1, F1 start appending log entries
// to its log store, which will call logstore::append_log_entries and thus call localize_journal_entry_finish.
Expand Down Expand Up @@ -2432,7 +2436,7 @@ nuraft::cb_func::ReturnCode RaftReplDev::raft_event(nuraft::cb_func::Type type,
// always go with -1 from NuRraft code.
//
// We are rejecting this log entry, meaning we can accept previous log entries.
// If there is nothing we can accept(i==0), that maens we are waiting for commit
// If there is nothing we can accept(i==0), that means we are waiting for commit
// of previous lsn, set it to 1 in this case.
m_state_machine->reset_next_batch_size_hint(std::max(1ul, i));
return nuraft::cb_func::ReturnCode::ReturnNull;
Expand Down Expand Up @@ -2564,7 +2568,7 @@ void RaftReplDev::monitor_replace_member_replication_status() {
boost::uuids::to_string(replica_in), boost::uuids::to_string(replica_out))
}

/////////////////////////////////// Private metohds ////////////////////////////////////
/////////////////////////////////// Private methods ////////////////////////////////////
void RaftReplDev::cp_flush(CP* cp, cshared< ReplDevCPContext > ctx) {
if (is_destroyed()) {
RD_LOGI(NO_TRACE_ID, "Raft repl dev is destroyed, ignore cp flush");
Expand Down Expand Up @@ -2614,7 +2618,7 @@ void RaftReplDev::cp_cleanup(CP*) {}
void RaftReplDev::gc_repl_reqs() {
auto cur_dsn = m_next_dsn.load();
if (cur_dsn != 0) cur_dsn = cur_dsn - 1;
// On follower, DSN below cur_dsn should very likely be commited.
// On follower, DSN below cur_dsn should very likely be committed.
// It is not guaranteed because DSN and LSN are generated separately,
// DSN in async_alloc_write before pushing data, LSN later when
// proposing to raft. Two simultaneous write requests on leader can have
Expand All @@ -2636,7 +2640,7 @@ void RaftReplDev::gc_repl_reqs() {
// The DSN can be out of order, wait till rreq expired.
RD_LOGD_EVERY_N(
unmove(gc_log_freq), rreq->traceID(),
"legacy req with commited DSN, rreq=[{}] , dsn = {}, next_dsn = {}, gap= {}, elapsed_hours {}",
"legacy req with committed DSN, rreq=[{}] , dsn = {}, next_dsn = {}, gap= {}, elapsed_hours {}",
rreq->to_string(), rreq->dsn(), cur_dsn, cur_dsn - rreq->dsn(),
get_elapsed_time_sec(rreq->created_time()) / 3600);
expired_rreqs.push_back(rreq);
Expand Down Expand Up @@ -2762,7 +2766,7 @@ void RaftReplDev::on_log_found(logstore_seq_num_t lsn, log_buffer buf, void* ctx
}

rreq->set_lsn(repl_lsn);
// keep lentry in scope for the lyfe cycle of the rreq
// keep lentry in scope for the life cycle of the rreq
rreq->set_lentry(lentry);
auto status = init_req_ctx(rreq, rkey, jentry->code, false /* is_proposer */, entry_to_hdr(jentry),
entry_to_key(jentry), data_size, m_listener);
Expand Down Expand Up @@ -2847,13 +2851,13 @@ bool RaftReplDev::save_snp_resync_data(nuraft::buffer& data, nuraft::snapshot& s
void RaftReplDev::on_restart() { m_listener->on_restart(); }

bool RaftReplDev::is_resync_mode() {
int64_t const leader_commited_lsn = raft_server()->get_leader_committed_log_idx();
int64_t const leader_committed_lsn = raft_server()->get_leader_committed_log_idx();
int64_t const my_log_idx = raft_server()->get_last_log_idx();
auto diff = leader_commited_lsn - my_log_idx;
auto diff = leader_committed_lsn - my_log_idx;
bool resync_mode = (diff > HS_DYNAMIC_CONFIG(consensus.resync_log_idx_threshold));
if (resync_mode) {
RD_LOGD(NO_TRACE_ID, "Raft Channel: Resync mode, leader_commited_lsn={}, my_log_idx={}, diff={}",
leader_commited_lsn, my_log_idx, diff);
RD_LOGD(NO_TRACE_ID, "Raft Channel: Resync mode, leader_committed_lsn={}, my_log_idx={}, diff={}",
leader_committed_lsn, my_log_idx, diff);
}
return resync_mode;
}
Expand All @@ -2876,8 +2880,8 @@ void RaftReplDev::quiesce_reqs() {
// can make sure
// 1 all the pending reqs has allocated their blocks
// 2 no new pending reqs will be initialized again.
m_in_quience.store(true, std::memory_order_release);
RD_LOGD(NO_TRACE_ID, "enter quience state, waiting for all the pending req to be initialized");
m_in_quiescence.store(true, std::memory_order_release);
RD_LOGD(NO_TRACE_ID, "enter quiescence state, waiting for all the pending req to be initialized");
while (true) {
uint64_t pending_req_num = get_pending_init_req_num();
if (pending_req_num) {
Expand All @@ -2891,9 +2895,9 @@ void RaftReplDev::quiesce_reqs() {
void RaftReplDev::reset_latch_lsn() { m_latch_lsn.store(INT64_MAX); }

void RaftReplDev::resume_accepting_reqs() {
m_in_quience.store(false, std::memory_order_release);
m_in_quiescence.store(false, std::memory_order_release);
reset_latch_lsn();
RD_LOGD(NO_TRACE_ID, "exit quience state, resume accepting new requests");
RD_LOGD(NO_TRACE_ID, "exit quiescence state, resume accepting new requests");
}

void RaftReplDev::clear_chunk_req(chunk_num_t chunk_id) {
Expand Down Expand Up @@ -2922,7 +2926,7 @@ void RaftReplDev::clear_chunk_req(chunk_num_t chunk_id) {
folly::collectAllUnsafe(futs)
.thenValue([this](auto&& vf) {
// TODO:: handle the error in freeing blk if necessary in the future.
// for nuobject case, error for freeing blk in the emergent chunk can be ingored
// for nuobject case, error for freeing blk in the emergent chunk can be ignored
RD_LOGD(
NO_TRACE_ID,
"all the necessary in-memory rreqs which has allocated blks on the emergent chunk have been cleaned up "
Expand All @@ -2941,10 +2945,10 @@ ReplServiceError RaftReplDev::init_req_ctx(repl_req_ptr_t rreq, repl_key rkey, j
}

init_req_counter counter(m_pending_init_req_num);
if (is_in_quience()) {
// In quience state, reject any new requests.
RD_LOGD(rkey.traceID, "Rejecting new request in quience state, rkey=[{}]", rkey.to_string());
return ReplServiceError::QUIENCE_STATE;
if (is_in_quiescence()) {
// In quiescence state, reject any new requests.
RD_LOGD(rkey.traceID, "Rejecting new request in quiescence state, rkey=[{}]", rkey.to_string());
return ReplServiceError::QUIESCENCE_STATE;
}

return rreq->init(rkey, op_code, is_proposer, user_header, key, data_size, m_listener);
Expand Down
Loading
Loading