Skip to content

fix found issues #806

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 3 commits into from
Sep 24, 2025
+49 −26
Merged

fix found issues #806

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
12284d9
fix found issues
JacksonYao287 Sep 22, 2025
f6f593d
fix comments
JacksonYao287 Sep 24, 2025
5a4af41
fix
JacksonYao287 Sep 24, 2025
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 = "6.20.23"
version = "6.20.24"

homepage = "https://github.com/eBay/Homestore"
description = "HomeStore Storage Engine"
Expand Down
55 changes: 35 additions & 20 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 @@ -820,7 +820,8 @@ void RaftReplDev::async_alloc_write(sisl::blob const& header, sisl::blob const&
COUNTER_DECREMENT(m_metrics, outstanding_data_write_cnt, 1);

if (err) {
HS_DBG_ASSERT(false, "Error in writing data, err_code={}", err.value());
HS_DBG_ASSERT(false, "Error in writing data, err_code={}, category={}, err_message={}", err.value(),
err.category().name(), err.message());
handle_error(rreq, ReplServiceError::DRIVE_WRITE_ERROR);
} else {
// update metrics for originated rreq;
Expand Down Expand Up @@ -955,7 +956,8 @@ void RaftReplDev::on_push_data_received(intrusive< sisl::GenericRpcData >& rpc_d

if (err) {
COUNTER_INCREMENT(m_metrics, write_err_cnt, 1);
RD_DBG_ASSERT(false, "Error in writing data, error_code={}", err.value());
RD_DBG_ASSERT(false, "Error in writing data, error_code={},category={}, err_message={}", err.value(),
err.category().name(), err.message());
handle_error(rreq, ReplServiceError::DRIVE_WRITE_ERROR);
} else {
rreq->release_data();
Expand Down Expand Up @@ -1273,7 +1275,7 @@ void RaftReplDev::on_fetch_data_received(intrusive< sisl::GenericRpcData >& rpc_
fetch_req->request()->entries()->size());

std::vector< sisl::sg_list > sgs_vec;
std::vector< folly::Future< bool > > futs;
std::vector< folly::Future< std::error_code > > futs;
sgs_vec.reserve(fetch_req->request()->entries()->size());
futs.reserve(fetch_req->request()->entries()->size());

Expand Down Expand Up @@ -1309,12 +1311,21 @@ void RaftReplDev::on_fetch_data_received(intrusive< sisl::GenericRpcData >& rpc_
folly::collectAllUnsafe(futs).thenValue(
[this, rpc_data = std::move(rpc_data), sgs_vec = std::move(sgs_vec)](auto&& vf) {
for (auto const& err_c : vf) {
if (sisl_unlikely(err_c.value())) {
COUNTER_INCREMENT(m_metrics, read_err_cnt, 1);
RD_REL_ASSERT(false, "Error in reading data");
// TODO: Find a way to return error to the Listener
// TODO: actually will never arrive here as iomgr will assert
// (should not assert but to raise alert and leave the raft group);
const auto& err = err_c.value();
if (err) {
// if read data failed, we should ignore the rpc_data and let the follower retry the fetch
RD_LOGD(NO_TRACE_ID,
"Data Channel: Error happens when fetching data. value={}, category={}, err_message={}, "
"ignoring this call",
err.value(), err.category().name(), err.message());

for (auto const& sgs : sgs_vec) {
for (auto const& iov : sgs.iovs) {
iomanager.iobuf_free(reinterpret_cast< uint8_t* >(iov.iov_base));
}
}
rpc_data->send_response();
return;
}
}

Expand Down Expand Up @@ -1345,10 +1356,12 @@ void RaftReplDev::handle_fetch_data_response(sisl::GenericClientResponse respons
auto raw_data = resp_blob.cbytes();
auto total_size = resp_blob.size();

COUNTER_INCREMENT(m_metrics, fetch_total_blk_size, total_size);
if (!total_size || !raw_data) {
RD_LOGW(NO_TRACE_ID, "Empty response from remote!");
return;
}

RD_DBG_ASSERT_GT(total_size, 0, "Empty response from remote");
RD_DBG_ASSERT(raw_data, "Empty response from remote");
COUNTER_INCREMENT(m_metrics, fetch_total_blk_size, total_size);

RD_LOGD(NO_TRACE_ID, "Data Channel: FetchData completed for {} requests", rreqs.size());

Expand Down Expand Up @@ -1381,8 +1394,9 @@ void RaftReplDev::handle_fetch_data_response(sisl::GenericClientResponse respons
HISTOGRAM_OBSERVE(m_metrics, rreq_data_write_latency_us, data_write_latency);
HISTOGRAM_OBSERVE(m_metrics, rreq_total_data_write_latency_us, total_data_write_latency);

RD_REL_ASSERT(!err,
"Error in writing data"); // TODO: Find a way to return error to the Listener
RD_REL_ASSERT(!err, "Error in writing data, error_code={}, category={}, err_message={}",
err.value(), err.category().name(), err.message());
// TODO: Find a way to return error to the Listener
rreq->release_data();
rreq->add_state(repl_req_state_t::DATA_WRITTEN);
rreq->m_data_written_promise.setValue();
Expand Down Expand Up @@ -2511,13 +2525,14 @@ ReplServiceError RaftReplDev::init_req_ctx(repl_req_ptr_t rreq, repl_key rkey, j
}

void RaftReplDev::become_leader_cb() {
auto current_gate = m_traffic_ready_lsn.load();
auto new_gate = raft_server()->get_last_log_idx();
repl_lsn_t existing_gate = 0;

@Besroy Besroy Sep 23, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we add a assert here to confirm the new gate is larger or equal to the current one?

@JacksonYao287 JacksonYao287 Sep 23, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure

if (!m_traffic_ready_lsn.compare_exchange_strong(existing_gate, new_gate)) {
// was a follower, m_traffic_ready_lsn should be zero on follower.
RD_REL_ASSERT(!existing_gate, "existing gate should be zero");
}
RD_LOGD(NO_TRACE_ID, "become_leader_cb: setting traffic_ready_lsn from {} to {}", existing_gate, new_gate);
m_traffic_ready_lsn.store(new_gate);

// in nuraft, a leader might become a new leader directly, which means it is unnecessary to become a follower before
// becoming a leader.

RD_LOGD(NO_TRACE_ID, "become_leader_cb: setting traffic_ready_lsn from {} to {}", current_gate, new_gate);
}

bool RaftReplDev::is_ready_for_traffic() const {
Expand Down
6 changes: 1 addition & 5 deletions src/lib/replication/repl_dev/raft_repl_dev.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -357,11 +357,7 @@ class RaftReplDev : public ReplDev,

void become_leader_cb();

void become_follower_cb() {
// m_traffic_ready_lsn should be zero on follower.
m_traffic_ready_lsn.store(0);
RD_LOGD(NO_TRACE_ID, "become_follower_cb setting traffic_ready_lsn to 0");
}
void become_follower_cb() { RD_LOGD(NO_TRACE_ID, "become_follower_cb called!"); }

@Besroy Besroy Sep 23, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we add m_traffic_ready_lsn.store(0) to make sure the follower can process the coming traffic

@JacksonYao287 JacksonYao287 Sep 23, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don`t get this. follower can not process any traffic since we have a check like this

    if (!repl_dev->is_leader()) {
        LOGW("failed to create shard for pg={}, not leader", pg_owner);
        decr_pending_request_num();
        return folly::makeUnexpected(ShardError(ShardErrorCode::NOT_LEADER, repl_dev->get_leader_id()));
    }

    if (!repl_dev->is_ready_for_traffic()) {
        LOGW("failed to create shard for pg={}, not ready for traffic", pg_owner);
        decr_pending_request_num();
        return folly::makeUnexpected(ShardError(ShardErrorCode::RETRY_REQUEST));
    }

m_traffic_ready_lsn is only used by leader , no?

@Besroy Besroy Sep 23, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is_ready_for_traffic is not limited to the put path; resetting it ensures all operations, such as the get path, proceed as expected. For example, the get blob path: https://github.com/eBay/HomeObject/blob/main/src/lib/homestore_backend/hs_blob_manager.cpp#L297-L302

@JacksonYao287 JacksonYao287 Sep 23, 2025 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the right solution is adding is_leader() check in get_blob path. we should always read_blob from leader, since follower is probably slow than leader , and return a wrong result. changing m_traffic_ready_lsn at follower side give no help? for example, if a put_blob has a lsn 100, and it is committed at leader, then when we get_blob at leader , we can find this blob. however, if we get_blob at follower , even if we have already set m_traffic_ready_lsn.store(0) at follower, which makes sure we pass this check (if (!repl_dev->is_ready_for_traffic())), lsn 100 might not be committed at follower and thus return a not_found for this get_blob at follower. so , I think follower should return not_leader here and let the client retry leader.

@Besroy Besroy Sep 24, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIRC, the gateway can read blobs from all members, with a policy to choose the member in the same DC to reduce latency. Reading blobs only from the leader increases the leader's load, so I believe allowing followers to support reads is the right strategy. If a follower cannot find the blob (e.g., due to uncommitted), the gateway will retry with other members

@JacksonYao287 JacksonYao287 Sep 24, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I got your point, locality for reading is necessary.
I am thinking if the blob we are trying to read has been deleted and gc(we can not find the blob in pg_index_table), then every member will return UNKNOWN_BLOB error to gateway. what should gateway do?

retry for ever ? or there is any specific retry policy

there should be a guy to tell gateway that the blob does not exist in all members.


/// @brief This method is called when the data journal is compacted
///
Expand Down
10 changes: 8 additions & 2 deletions src/lib/replication/service/raft_repl_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,13 @@ void RaftReplService::start() {
}

void RaftReplService::stop() {
// we stop reaper thread here before destorying repl_dev to prevent data from being fetched after repl_dev is
// stopped.

// FIXME: there is still a case that before we stop_reaper_thread, some fetch_data requests have already been sent
// out. we need use a counter to make sure all the fetch_data request has completed.
stop_reaper_thread();

start_stopping();
while (true) {
auto pending_request_num = get_pending_request_num();
Expand Down Expand Up @@ -236,8 +243,7 @@ void RaftReplService::stop() {
}

RaftReplService::~RaftReplService() {
stop_reaper_thread();

LOGINFO("raft_repl_service has been destructed!");
// the base class destructor will clear the m_rd_map
}

Expand Down
Loading