Skip to content

[Fix]: bug fixes of simm client's fail-fast mechanism ; optimize stable test & kvio bench tools #29

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
Sebastian-Keith merged 2 commits into from
Apr 13, 2026
Merged

[Fix]: bug fixes of simm client's fail-fast mechanism ; optimize stable test & kvio bench tools #29

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1dca03c
[Fix]: bug fixes of simm client's fail-fast mechanism ; optimize stab…
Sebastian-Keith Apr 13, 2026
340ac46
[Fix]: update some code comments
Sebastian-Keith Apr 13, 2026
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
14 changes: 7 additions & 7 deletions k8s/simm/templates/cluster-manager.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ subjects:
apiVersion: v1
kind: Service
metadata:
##定义headless service名，用于关联到每个pod资源创建DNS资源记录
## Define headless service name, used to create DNS records for each pod
name: simm-clustermanager-svc
{{ if .Values.namespace }}
namespace: {{ .Values.namespace }}
Expand All @@ -60,9 +60,9 @@ spec:
{{ end }}
---
apiVersion: apps/v1
kind: Deployment #定义资源类型
metadata: #定义资源的元数据信息, 比如资源的名称、namespace、标签等信息
name: simm-clustermanager #定义资源的名字，在同一个namespace空间中必须是唯一的
kind: Deployment # Resource type
metadata: # Resource metadata (name, namespace, labels, etc.)
name: simm-clustermanager # Resource name, must be unique within the namespace
{{ if .Values.namespace }}
namespace: {{ .Values.namespace }}
{{ end }}
Expand All @@ -72,15 +72,15 @@ metadata: #定义资源的元数据信息, 比如资源的名称、namespace、
{{ else }}
app: app-simm-clustermanager-svc
{{ end }}
spec: #定义资源需要的参数属性
selector: #定义标签选择器
spec: # Resource spec
selector: # Label selector
matchLabels:
{{ if .Values.namespace }}
app: app-{{ .Values.namespace }}-simm-clustermanager-svc
{{ else }}
app: app-simm-clustermanager-svc
{{ end }}
template: #定义pod templete
template: # Pod template
metadata:
{{- with .Values.podAnnotations }}
annotations:
Expand Down
16 changes: 8 additions & 8 deletions k8s/simm/templates/data-server.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
apiVersion: v1
kind: Service
metadata:
##定义headless service名，用于关联到每个pod资源创建DNS资源记录
## Define headless service name, used to create DNS records for each pod
name: simm-data-svc
{{ if .Values.namespace }}
namespace: {{ .Values.namespace }}
Expand All @@ -25,25 +25,25 @@ spec:
{{ end }}
---
apiVersion: apps/v1
kind: StatefulSet #定义资源类型
metadata: #定义资源的元数据信息, 比如资源的名称、namespace、标签等信息
name: simm-data-svc #定义资源的名字，在同一个namespace空间中必须是唯一的
kind: StatefulSet # Resource type
metadata: # Resource metadata (name, namespace, labels, etc.)
name: simm-data-svc # Resource name, must be unique within the namespace
{{ if .Values.namespace }}
namespace: {{ .Values.namespace }}
{{ end }}
spec: #定义资源需要的参数属性
# 该serviceName就是上面headless service的name，指定此StatefulSet属于哪个headless service
spec: # Resource spec
# serviceName references the headless service above, binding this StatefulSet to it
serviceName: simm-data-svc
replicas: {{ .Values.replicaCount }}
selector: #定义标签选择器
selector: # Label selector
matchLabels:
{{ if .Values.namespace }}
app: app-{{ .Values.namespace }}-simm-data-svc
{{ else }}
app: app-simm-data-svc
{{ end }}
podManagementPolicy: Parallel
template: #定义pod templete
template: # Pod template
metadata:
{{- with .Values.podAnnotations }}
annotations:
Expand Down
19 changes: 18 additions & 1 deletion src/client/clnt_kv.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -309,7 +309,24 @@ std::vector<int16_t> KVStore::MPut(const std::vector<std::string> &keys, std::ve
}
auto rets = simm::clnt::ClientMessenger::Instance().MultiPut(keys, metadatas);
for (size_t i = 0; i < key_cnt; ++i) {
if (rets[i] < 0) {
if (rets[i] == DsErr::DataAlreadyExists) {
// FIXME(szzhao): remove workaround after it's implemented in data server
MLOG_WARN("MPut kv {} will override existing entry", keys[i]);
auto del_res = Delete(keys[i]);
if (del_res != CommonErr::OK) {
MLOG_ERROR("MPut kv {} by deleting existing one failed: {}", keys[i], del_res);
continue;
}
auto ctx = std::make_shared<simm::common::SimmContext>();
auto retry_res = simm::clnt::ClientMessenger::Instance().Put(keys[i], metadatas[i], ctx);
// NOTE(zxliao): assume same value already put if entry still exist after deleting
if (retry_res == CommonErr::OK || retry_res == DsErr::DataAlreadyExists) {
rets[i] = CommonErr::OK;
} else {
rets[i] = retry_res;
MLOG_ERROR("MPut kv {} retry after delete failed: {}", keys[i], retry_res);
}
} else if (rets[i] < 0) {
MLOG_ERROR("MPut kv {} failed: {}", keys[i], rets[i]);
}
}
Expand Down
155 changes: 94 additions & 61 deletions src/client/clnt_messenger.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <future>
#include <memory>
#include <mutex>
#include <string>
Expand Down Expand Up @@ -144,60 +145,74 @@ error_code_t ClientMessenger::Init() {
break;
}

bool should_reinit = false;
if (get_cm_address() != cm_addr_) {
// cluster manager address changed, maybe it was restarted, so client should
// sync with it and get latest data servers address info
should_reinit = true;
} else {
for (auto [addr, ds_ctx] : ds_conn_ctxs_) {
if (!ds_ctx->active.load()) {
// track when DS was first seen as dead
{
std::lock_guard lg(ds_dead_since_mtx_);
if (!ds_dead_since_.count(addr)) {
ds_dead_since_[addr] = std::chrono::steady_clock::now();
try {
bool should_reinit = false;
if (get_cm_address() != cm_addr_) {
// cluster manager address changed, maybe it was restarted, so client should
// sync with it and get latest data servers address info
should_reinit = true;
} else {
// Collect inactive DS addresses and track dead_since
std::vector<std::string> inactive_addrs;
for (auto [addr, ds_ctx] : ds_conn_ctxs_) {
if (!ds_ctx->active.load()) {
{
std::lock_guard lg(ds_dead_since_mtx_);
if (!ds_dead_since_.count(addr)) {
ds_dead_since_[addr] = std::chrono::steady_clock::now();
}
}
inactive_addrs.push_back(addr);
}
}

// Try to reconnect — new DS may have come up with same port
if (CommonErr::OK == build_connection(addr)) {
std::lock_guard lg(ds_dead_since_mtx_);
ds_dead_since_.erase(addr);
continue;
}
// Parallel reconnect attempts for all inactive DS
std::vector<std::future<std::pair<std::string, error_code_t>>> futures;
for (const auto &addr : inactive_addrs) {
futures.push_back(std::async(std::launch::async, [this, addr]() {
return std::make_pair(addr, build_connection(addr));
}));
}

// Reconnect failed — check if deferred reshard wait window exceeded.
// NOTE: CM updates its routing table immediately upon DS handshake (IP update or
// replacement), but the client has no way to learn about it promptly because
// CM-to-client routing push (RPC_ROUTING_TABLE_UPDATE) is not yet implemented
// (see cm_service.cc TODO). Until push is available, the client can only discover
// the new IP by polling CM via ReInit() after this wait window expires.
// If the DS restarts with a different IP, IO to the affected shards will fail for
// up to clnt_deferred_reshard_wait_inSecs seconds. This is a known limitation;
// implementing CM→client push will eliminate the gap.
std::chrono::duration<double> dur;
{
// Process reconnect results
for (auto &f : futures) {
auto [addr, ret] = f.get();
if (ret == CommonErr::OK) {
std::lock_guard lg(ds_dead_since_mtx_);
dur = std::chrono::steady_clock::now() - ds_dead_since_[addr];
}

if (dur > std::chrono::seconds(FLAGS_clnt_deferred_reshard_wait_inSecs)) {
// Window expired: CM may have done reshard or IP update, pull new routes
ds_dead_since_.erase(addr);
} else {
// Reconnect failed — check if deferred reshard wait window exceeded.
// NOTE: CM updates its routing table immediately upon DS handshake (IP update or
// replacement), but the client has no way to learn about it promptly because
// CM-to-client routing push (RPC_ROUTING_TABLE_UPDATE) is not yet implemented
// (see cm_service.cc TODO). Until push is available, the client can only discover
// the new IP by polling CM via ReInit() after this wait window expires.
std::chrono::duration<double> dur;
{
std::lock_guard lg(ds_dead_since_mtx_);
ds_dead_since_.erase(addr);
dur = std::chrono::steady_clock::now() - ds_dead_since_[addr];
}

if (dur > std::chrono::seconds(FLAGS_clnt_deferred_reshard_wait_inSecs)) {
{
std::lock_guard lg(ds_dead_since_mtx_);
ds_dead_since_.erase(addr);
}
should_reinit = true;
MLOG_ERROR("Failover thread: DS {} unreachable for {}s (> {}s window), triggering reinit",
addr, static_cast<int>(dur.count()), FLAGS_clnt_deferred_reshard_wait_inSecs);
}
should_reinit = true;
MLOG_ERROR("Failover thread: DS {} unreachable for {}s (> {}s window), triggering reinit",
addr, static_cast<int>(dur.count()), FLAGS_clnt_deferred_reshard_wait_inSecs);
// else: still within wait window, keep retrying next cycle
}
// else: still within wait window, keep retrying next cycle
}
}
}
if (should_reinit) {
ReInit();
if (should_reinit) {
ReInit();
}
} catch (const std::exception &e) {
MLOG_ERROR("Failover thread caught exception: {}", e.what());
} catch (...) {
MLOG_ERROR("Failover thread caught unknown exception");
}
}
});
Expand Down Expand Up @@ -245,7 +260,13 @@ error_code_t ClientMessenger::build_connection(const std::string &addr, BuildCon
return ClntErr::BuildConnectionFailed;
}
std::unique_lock lock(ds_ctx->connect_wait_mutex_);
ds_ctx->connect_cv_.wait(lock, [&]() { return !ds_ctx->connecting_.load(std::memory_order_acquire); });
bool wait_done = ds_ctx->connect_cv_.wait_for(
lock, std::chrono::seconds(30),
[&]() { return !ds_ctx->connecting_.load(std::memory_order_acquire); });
if (!wait_done) {
MLOG_WARN("Timed out waiting for in-flight connection to {}", addr);
return ClntErr::BuildConnectionFailed;
}
if (ds_ctx->active.load() && ds_ctx->LoadConnection() != nullptr) {
return CommonErr::OK;
}
Expand Down Expand Up @@ -353,6 +374,11 @@ error_code_t ClientMessenger::call_sync(uint16_t shard_id,
return ClntErr::ClntLookupShardFailed;
}

//FIXME: Disable retry mechanism for modify reqeusts to avoid ABA issues
const bool retryable_req =
req_type == static_cast<sicl::rpc::ReqType>(simm::ds::KVServerRpcType::RPC_CLIENT_KV_GET) ||
req_type == static_cast<sicl::rpc::ReqType>(simm::ds::KVServerRpcType::RPC_CLIENT_KV_LOOKUP);

auto rpc_ctx = ctx->get_rpc_ctx();
auto retry_delay = std::chrono::milliseconds(100);
for (auto i = 0; i <= FLAGS_clnt_syncreq_retry_count; ++i) {
Expand Down Expand Up @@ -389,6 +415,13 @@ error_code_t ClientMessenger::call_sync(uint16_t shard_id,
ReconnectByErrors(rpc_ctx, ds_ctx, shard_id, tag);
} else {
MLOG_WARN("Transport connection is inactive for shard id {}, data server is {}", shard_id, ds_ctx->ip_port);
// Connection already inactive — no point retrying, fail fast
break;
}

if (!retryable_req) {
MLOG_WARN("Sync request type {} is non-retryable, fail fast after first error", static_cast<uint64_t>(req_type));
break;
}

if (!FLAGS_clnt_syncreq_enable_retry) {
Expand All @@ -400,8 +433,12 @@ error_code_t ClientMessenger::call_sync(uint16_t shard_id,
}
}

MLOG_ERROR("Failed to send request after {} retries (sync call)",
FLAGS_clnt_syncreq_enable_retry ? FLAGS_clnt_syncreq_retry_count : 0);
MLOG_ERROR("Failed to send sync request, type:{}(retryable:{}), enable retry:{}, retry count:{}",
static_cast<uint64_t>(req_type),
retryable_req ? "Y" : "N",
FLAGS_clnt_syncreq_enable_retry ? "Y" : "N",
FLAGS_clnt_syncreq_retry_count);

return ClntErr::ClntSendRPCFailed;
}

Expand Down Expand Up @@ -550,7 +587,6 @@ error_code_t ClientMessenger::ApplyRouteTableDiff(const QueryShardRoutingTableAl
std::unordered_set<std::string> live_servers;
std::vector<std::string> servers_to_connect;

shard_table_.clear();
for (const auto &entry : routing.shard_info()) {
std::string ip = entry.data_server_address().ip();
uint16_t port = static_cast<uint16_t>(entry.data_server_address().port());
Expand Down Expand Up @@ -585,7 +621,8 @@ void ClientMessenger::ReconnectByErrors(std::shared_ptr<sicl::rpc::RpcContext> r

if (rpc_ctx->ErrorCode() == sicl::transport::SICL_ERR_INVALID_STATE ||
rpc_ctx->ErrorCode() == sicl::transport::SICL_ERR_VERBS_WC_ERROR ||
rpc_ctx->ErrorCode() == sicl::transport::SICL_ERR_VERBS_POST_SEND) {
rpc_ctx->ErrorCode() == sicl::transport::SICL_ERR_VERBS_POST_SEND ||
rpc_ctx->ErrorCode() == sicl::transport::SICL_ERR_TIMEOUT) {
MLOG_WARN("Encountered transport error {} for shard id {}, will try to reconnect (sync call)",
rpc_ctx->ErrorCode(),
shard_id);
Expand Down Expand Up @@ -720,7 +757,7 @@ error_code_t ClientMessenger::AsyncPut(const std::string &key,
#ifdef SIMM_APIPERF
auto t1 = std::chrono::steady_clock::now();
#endif
call_async<KVPutRequestPB, KVPutResponsePB>(
auto ret = call_async<KVPutRequestPB, KVPutResponsePB>(
shard_id,
static_cast<sicl::rpc::ReqType>(simm::ds::KVServerRpcType::RPC_CLIENT_KV_PUT),
req,
Expand All @@ -734,7 +771,7 @@ error_code_t ClientMessenger::AsyncPut(const std::string &key,
std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count());
#endif

return CommonErr::OK;
return ret;
}

int32_t ClientMessenger::Get(const std::string &key,
Expand Down Expand Up @@ -849,7 +886,7 @@ error_code_t ClientMessenger::AsyncGet(const std::string &key,
#ifdef SIMM_APIPERF
auto t1 = std::chrono::steady_clock::now();
#endif
call_async<KVGetRequestPB, KVGetResponsePB>(
auto ret = call_async<KVGetRequestPB, KVGetResponsePB>(
shard_id,
static_cast<sicl::rpc::ReqType>(simm::ds::KVServerRpcType::RPC_CLIENT_KV_GET),
req,
Expand All @@ -863,7 +900,7 @@ error_code_t ClientMessenger::AsyncGet(const std::string &key,
std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count());
#endif

return CommonErr::OK;
return ret;
}

error_code_t ClientMessenger::Delete(const std::string &key, std::shared_ptr<simm::common::SimmContext> ctx) {
Expand Down Expand Up @@ -927,15 +964,13 @@ error_code_t ClientMessenger::AsyncDelete(const std::string &key,
cb(new_resp->ret_code());
}
};
call_async<KVDelRequestPB, KVDelResponsePB>(
return call_async<KVDelRequestPB, KVDelResponsePB>(
shard_id,
static_cast<sicl::rpc::ReqType>(simm::ds::KVServerRpcType::RPC_CLIENT_KV_DEL),
req,
resp,
ctx,
std::move(done));

return CommonErr::OK;
}

error_code_t ClientMessenger::Exists(const std::string &key, std::shared_ptr<simm::common::SimmContext> ctx) {
Expand Down Expand Up @@ -999,15 +1034,13 @@ error_code_t ClientMessenger::AsyncExists(const std::string &key,
cb(new_resp->ret_code());
}
};
call_async<KVLookupRequestPB, KVLookupResponsePB>(
return call_async<KVLookupRequestPB, KVLookupResponsePB>(
shard_id,
static_cast<sicl::rpc::ReqType>(simm::ds::KVServerRpcType::RPC_CLIENT_KV_LOOKUP),
req,
resp,
ctx,
std::move(done));

return CommonErr::OK;
}

std::vector<error_code_t> ClientMessenger::MultiPut(const std::vector<std::string> &keys,
Expand Down Expand Up @@ -1076,10 +1109,10 @@ std::vector<error_code_t> ClientMessenger::MultiExists(const std::vector<std::st
return multiExistsResults;
}

inline sicl::transport::TimerTick ClientMessenger::convert_timeout_setting_to_timer_tick(int32_t timeout_ms) {
sicl::transport::TimerTick ClientMessenger::convert_timeout_setting_to_timer_tick(int32_t timeout_ms) {
if (timeout_ms < 0) {
// if timeout is set to negative value, means wait forever
return sicl::transport::TimerTick::TIMER_END;
MLOG_WARN("Negative timeout_ms {} clamped to TIMER_60S to prevent infinite wait", timeout_ms);
return sicl::transport::TimerTick::TIMER_60S;
} else if (timeout_ms <= 1) {
return sicl::transport::TimerTick::TIMER_1MS;
} else if (timeout_ms <= 3) {
Expand Down
Loading
Loading