From a2e42f838b651b2ed9875e225535bc9c59dd3f9c Mon Sep 17 00:00:00 2001 From: Sebastian-Keith Date: Wed, 8 Apr 2026 16:28:58 +0800 Subject: [PATCH] [Fix]: 1) add ds GetHandler check for case that req buf size less than kvmeta size ; 2) optimize stable test --- src/client/clnt_messenger.cc | 4 + src/data_server/kv_rpc_handler.cc | 159 ++- tests/CMakeLists.txt | 3 +- tests/client/test_clnt_messenger.cc | 21 + tests/data_server/test_ds_kv_service.cc | 114 ++ tests/tools/CMakeLists.txt | 16 + tests/tools/test_simm_stable_test_utils.cc | 31 + tools/simm_stable_test.cc | 1283 ++++++++++++++------ tools/simm_stable_test_upgrade.md | 141 +++ tools/simm_stable_test_utils.h | 45 + 10 files changed, 1397 insertions(+), 420 deletions(-) create mode 100644 tests/tools/CMakeLists.txt create mode 100644 tests/tools/test_simm_stable_test_utils.cc create mode 100644 tools/simm_stable_test_upgrade.md create mode 100644 tools/simm_stable_test_utils.h diff --git a/src/client/clnt_messenger.cc b/src/client/clnt_messenger.cc index 3f8ea1d..f912d73 100644 --- a/src/client/clnt_messenger.cc +++ b/src/client/clnt_messenger.cc @@ -41,6 +41,7 @@ DECLARE_uint32(clnt_thread_pool_size); DECLARE_uint32(clnt_cm_addr_check_interval_inSecs); DECLARE_uint32(clnt_syncreq_retry_count); DECLARE_bool(clnt_syncreq_enable_retry); +DECLARE_bool(clnt_use_k8s); DECLARE_bool(simm_enable_trace); DECLARE_LOG_MODULE("simm_client"); @@ -177,6 +178,9 @@ std::string ClientMessenger::get_cm_address() { } #endif const std::string default_cm_addr = FLAGS_cm_primary_node_ip + ":" + std::to_string(FLAGS_cm_rpc_inter_port); + if (!FLAGS_clnt_use_k8s) { + return default_cm_addr; + } // get cluster manager pod info from K8S api, get vector of [pod_name, pod_ip] // TODO: change to get cluster manager info from etcd auto [ret, cm_ips] = diff --git a/src/data_server/kv_rpc_handler.cc b/src/data_server/kv_rpc_handler.cc index 804ccbb..81d8c17 100644 --- a/src/data_server/kv_rpc_handler.cc +++ b/src/data_server/kv_rpc_handler.cc @@ -30,35 +30,59 @@ void KVGetHandler::Work(const std::shared_ptr ctx, auto ret = service_->KVGet(simm_ctx, req, kv_entry); if (ret == CommonErr::OK) { auto [meta, data] = KVCachePool::GetBufferPair(&kv_entry->slab_info); + if (req->buf_len() < meta->value_len) { + MLOG_ERROR("KVGetHandler::Work client buffer too small for key {}, buf_len:{}, value_len:{}", + req->key(), + req->buf_len(), + meta->value_len); + service_->KVGetCallback(kv_entry); + rsp->set_ret_code(CommonErr::InvalidArgument); + simm::common::Metrics::Instance("data_server") + .ObserveRequestDuration("Get", + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()) + .count())); + simm::common::Metrics::Instance("data_server").IncRequestsTotal("Get"); + simm::common::Metrics::Instance("data_server").IncErrorsTotal("Get"); + conn->SendResponse(*rsp, ctx, [rsp](std::shared_ptr ctx) { + if (ctx->Failed()) { + MLOG_ERROR("{} response failed: {}({})", rsp->GetTypeName(), ctx->ErrorText(), ctx->ErrorCode()); + } + }); + return; + } sicl::transport::RequestParam param{ .mem_desc = static_cast(kv_entry->slab_info.block_addr->descr)}; std::vector rkeys(req->buf_rkey().begin(), req->buf_rkey().end()); - sicl::transport::WriteCallback done = - [this, kv_entry, rsp, conn, ctx, meta, simm_ctx](sicl::transport::Status status) mutable { - error_code_t ret = CommonErr::OK; - if (status.isOk()) { - // return actual value length to client - rsp->set_val_len(meta->value_len); - // record bytes written on successful write - simm::common::Metrics::Instance("data_server").IncWrittenTotal(static_cast(meta->value_len)); - } else { - MLOG_ERROR("KVGetHandler::Work connection write failed, err_code:{}, err_msg:{}", - std::to_string(status.errCode()), - status.errMsg()); - ret = DsErr::DataRDMATransportFailed; - simm::common::Metrics::Instance("data_server").IncErrorsTotal("Get"); - } - service_->KVGetCallback(kv_entry); - rsp->set_ret_code(ret); - simm::common::Metrics::Instance("data_server").ObserveRequestDuration("Get", - static_cast(std::chrono::duration_cast(std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()).count())); - simm::common::Metrics::Instance("data_server").IncRequestsTotal("Get"); - conn->SendResponse(*rsp, ctx, [rsp](std::shared_ptr ctx) { - if (ctx->Failed()) { - MLOG_ERROR("{} response failed: {}({})", rsp->GetTypeName(), ctx->ErrorText(), ctx->ErrorCode()); - } - }); - }; + sicl::transport::WriteCallback done = [this, kv_entry, rsp, conn, ctx, meta, simm_ctx]( + sicl::transport::Status status) mutable { + error_code_t ret = CommonErr::OK; + if (status.isOk()) { + // return actual value length to client + rsp->set_val_len(meta->value_len); + // record bytes written on successful write + simm::common::Metrics::Instance("data_server").IncWrittenTotal(static_cast(meta->value_len)); + } else { + MLOG_ERROR("KVGetHandler::Work connection write failed, err_code:{}, err_msg:{}", + std::to_string(status.errCode()), + status.errMsg()); + ret = DsErr::DataRDMATransportFailed; + simm::common::Metrics::Instance("data_server").IncErrorsTotal("Get"); + } + service_->KVGetCallback(kv_entry); + rsp->set_ret_code(ret); + simm::common::Metrics::Instance("data_server") + .ObserveRequestDuration("Get", + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()) + .count())); + simm::common::Metrics::Instance("data_server").IncRequestsTotal("Get"); + conn->SendResponse(*rsp, ctx, [rsp](std::shared_ptr ctx) { + if (ctx->Failed()) { + MLOG_ERROR("{} response failed: {}({})", rsp->GetTypeName(), ctx->ErrorText(), ctx->ErrorCode()); + } + }); + }; auto res = conn->write(data, meta->value_len, req->buf_addr(), rkeys, done, param); if (res != sicl::transport::Result::SICL_SUCCESS) { done(sicl::transport::Status(res)); // synchronized return @@ -66,8 +90,11 @@ void KVGetHandler::Work(const std::shared_ptr ctx, return; } rsp->set_ret_code(ret); - simm::common::Metrics::Instance("data_server").ObserveRequestDuration("Get", - static_cast(std::chrono::duration_cast(std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()).count())); + simm::common::Metrics::Instance("data_server") + .ObserveRequestDuration("Get", + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()) + .count())); simm::common::Metrics::Instance("data_server").IncRequestsTotal("Get"); if (ret != CommonErr::OK) { simm::common::Metrics::Instance("data_server").IncErrorsTotal("Get"); @@ -112,32 +139,35 @@ void KVPutHandler::Work(const std::shared_ptr ctx, sicl::transport::RequestParam param{ .mem_desc = static_cast(kv_entry->slab_info.block_addr->descr)}; std::vector rkeys(req->buf_rkey().begin(), req->buf_rkey().end()); - sicl::transport::ReadCallback done = - [this, shard_id, kv_entry, rsp, conn, ctx, simm_ctx, meta](sicl::transport::Status status) mutable { - error_code_t ret; - if (status.isOk()) { - service_->KVPutSuccessHooks(kv_entry); - ret = CommonErr::OK; - // record bytes read when Put succeeded - simm::common::Metrics::Instance("data_server").IncReadTotal(static_cast(meta->value_len)); - } else { - service_->KVPutFailedRewind(shard_id, kv_entry); - MLOG_ERROR("KVPutHandler::Work connection read failed: err_code:{}, err_msg:{}", - std::to_string(status.errCode()), - status.errMsg()); - ret = DsErr::DataRDMATransportFailed; - simm::common::Metrics::Instance("data_server").IncErrorsTotal("Put"); - } - rsp->set_ret_code(ret); - simm::common::Metrics::Instance("data_server").ObserveRequestDuration("Put", - static_cast(std::chrono::duration_cast(std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()).count())); - simm::common::Metrics::Instance("data_server").IncRequestsTotal("Put"); - conn->SendResponse(*rsp, ctx, [rsp](std::shared_ptr ctx) { - if (ctx->Failed()) { - MLOG_ERROR("{} response failed: {}({})", rsp->GetTypeName(), ctx->ErrorText(), ctx->ErrorCode()); - } - }); - }; + sicl::transport::ReadCallback done = [this, shard_id, kv_entry, rsp, conn, ctx, simm_ctx, meta]( + sicl::transport::Status status) mutable { + error_code_t ret; + if (status.isOk()) { + service_->KVPutSuccessHooks(kv_entry); + ret = CommonErr::OK; + // record bytes read when Put succeeded + simm::common::Metrics::Instance("data_server").IncReadTotal(static_cast(meta->value_len)); + } else { + service_->KVPutFailedRewind(shard_id, kv_entry); + MLOG_ERROR("KVPutHandler::Work connection read failed: err_code:{}, err_msg:{}", + std::to_string(status.errCode()), + status.errMsg()); + ret = DsErr::DataRDMATransportFailed; + simm::common::Metrics::Instance("data_server").IncErrorsTotal("Put"); + } + rsp->set_ret_code(ret); + simm::common::Metrics::Instance("data_server") + .ObserveRequestDuration("Put", + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()) + .count())); + simm::common::Metrics::Instance("data_server").IncRequestsTotal("Put"); + conn->SendResponse(*rsp, ctx, [rsp](std::shared_ptr ctx) { + if (ctx->Failed()) { + MLOG_ERROR("{} response failed: {}({})", rsp->GetTypeName(), ctx->ErrorText(), ctx->ErrorCode()); + } + }); + }; auto res = conn->read(data, meta->value_len, req->buf_addr(), rkeys, done, param); if (res != sicl::transport::Result::SICL_SUCCESS) { done(sicl::transport::Status(res)); // synchronized return @@ -145,8 +175,11 @@ void KVPutHandler::Work(const std::shared_ptr ctx, return; } rsp->set_ret_code(ret); - simm::common::Metrics::Instance("data_server").ObserveRequestDuration("Put", - static_cast(std::chrono::duration_cast(std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()).count())); + simm::common::Metrics::Instance("data_server") + .ObserveRequestDuration("Put", + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()) + .count())); simm::common::Metrics::Instance("data_server").IncRequestsTotal("Put"); if (ret != CommonErr::OK) { simm::common::Metrics::Instance("data_server").IncErrorsTotal("Put"); @@ -167,8 +200,11 @@ void KVDelHandler::Work(const std::shared_ptr ctx, auto rsp = std::make_shared(); auto ret = service_->KVDel(simm_ctx, req); rsp->set_ret_code(ret); - simm::common::Metrics::Instance("data_server").ObserveRequestDuration("Delete", - static_cast(std::chrono::duration_cast(std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()).count())); + simm::common::Metrics::Instance("data_server") + .ObserveRequestDuration("Delete", + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()) + .count())); simm::common::Metrics::Instance("data_server").IncRequestsTotal("Delete"); if (ret != CommonErr::OK) { simm::common::Metrics::Instance("data_server").IncErrorsTotal("Delete"); @@ -189,8 +225,11 @@ void KVLookupHandler::Work(const std::shared_ptr ctx, auto rsp = std::make_shared(); auto ret = service_->KVLookup(simm_ctx, req); rsp->set_ret_code(ret); - simm::common::Metrics::Instance("data_server").ObserveRequestDuration("Lookup", - static_cast(std::chrono::duration_cast(std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()).count())); + simm::common::Metrics::Instance("data_server") + .ObserveRequestDuration("Lookup", + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - simm_ctx->GetReqStartTs()) + .count())); simm::common::Metrics::Instance("data_server").IncRequestsTotal("Lookup"); if (ret != CommonErr::OK) { simm::common::Metrics::Instance("data_server").IncErrorsTotal("Lookup"); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 5b5a3b9..408c3e6 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -10,4 +10,5 @@ add_subdirectory(common) add_subdirectory(client) add_subdirectory(cluster_manager) add_subdirectory(data_server) -add_subdirectory(correctness) \ No newline at end of file +add_subdirectory(tools) +add_subdirectory(correctness) diff --git a/tests/client/test_clnt_messenger.cc b/tests/client/test_clnt_messenger.cc index a69d5c8..da003e9 100644 --- a/tests/client/test_clnt_messenger.cc +++ b/tests/client/test_clnt_messenger.cc @@ -13,6 +13,9 @@ DECLARE_bool(clnt_syncreq_enable_retry); DECLARE_uint32(clnt_syncreq_retry_count); +DECLARE_bool(clnt_use_k8s); +DECLARE_string(cm_primary_node_ip); +DECLARE_int32(cm_rpc_inter_port); DECLARE_uint32(shard_total_num); namespace simm { @@ -297,6 +300,8 @@ class ClientMessengerTestPeer { static std::string CmAddr() { return ClientMessenger::Instance().cm_addr_; } + static std::string GetCmAddress() { return ClientMessenger::Instance().get_cm_address(); } + static sicl::rpc::SiRPC *SwapRpcClient(sicl::rpc::SiRPC *replacement) { auto &messenger = ClientMessenger::Instance(); auto *original = messenger.rpc_client_; @@ -598,6 +603,22 @@ TEST_F(ClientMessengerUnitTest, ReInitRemovesStaleConnectionFromLocalContextMap) EXPECT_EQ(ClientMessengerTestPeer::ShardOwner(1), "10.0.0.1:1001"); } +TEST_F(ClientMessengerUnitTest, GetCmAddressUsesFlagToSkipK8SLookup) { + const auto old_use_k8s = FLAGS_clnt_use_k8s; + const auto old_cm_ip = FLAGS_cm_primary_node_ip; + const auto old_cm_port = FLAGS_cm_rpc_inter_port; + + FLAGS_clnt_use_k8s = false; + FLAGS_cm_primary_node_ip = "10.8.0.1"; + FLAGS_cm_rpc_inter_port = 30001; + + EXPECT_EQ(ClientMessengerTestPeer::GetCmAddress(), "10.8.0.1:30001"); + + FLAGS_clnt_use_k8s = old_use_k8s; + FLAGS_cm_primary_node_ip = old_cm_ip; + FLAGS_cm_rpc_inter_port = old_cm_port; +} + } // namespace clnt } // namespace simm diff --git a/tests/data_server/test_ds_kv_service.cc b/tests/data_server/test_ds_kv_service.cc index 2ddef33..c8bb463 100644 --- a/tests/data_server/test_ds_kv_service.cc +++ b/tests/data_server/test_ds_kv_service.cc @@ -240,6 +240,120 @@ TEST_F(KVServiceTest, TestClientHandlers) { clientPool->join(); } +TEST_F(KVServiceTest, TestGetRejectsTooSmallClientBuffer) { + auto serverPool = std::make_unique(1); + auto clientPool = std::make_unique(1); + folly::Baton<> serverReady, serverExit; + + folly::via(serverPool->getEventBase(), [&] { + auto ret = rpcServicePtr->Start(); + EXPECT_EQ(ret, CommonErr::OK); + serverReady.post(); + serverExit.wait(); + }); + + serverReady.wait(); + + folly::Baton<> clientDone; + folly::via(clientPool->getEventBase(), [&] { + sicl::rpc::SiRPC *sirpc = nullptr; + sicl::rpc::SiRPC::newInstance(sirpc, false); + sicl::rpc::RpcContext *ctx_raw = nullptr; + sicl::rpc::RpcContext::newInstance(ctx_raw); + auto ctx = std::shared_ptr(ctx_raw); + ctx->set_timeout(sicl::transport::TimerTick::TIMER_1S); + + auto conn = sirpc->connect("127.0.0.1", FLAGS_io_service_port); + ASSERT_TRUE(conn != nullptr); + + const std::string key = "test_get_too_small_buffer"; + const uint32_t shard_id = + hashkit::HashkitBase::Instance().generate_16bit_hash_value(key.c_str(), key.length()) % FLAGS_shard_total_num; + constexpr size_t kValueLen = 4096; + constexpr size_t kSmallBufLen = 512; + + auto *mempool = sirpc->GetMempool(); + sicl::transport::MemDesc *put_descr = nullptr; + ASSERT_EQ(mempool->alloc(put_descr, kValueLen), sicl::transport::Result::SICL_SUCCESS); + std::string value; + get_random_string(kValueLen, &value); + memcpy(put_descr->getAddr(), value.data(), value.size()); + + auto put_req = std::make_shared(); + auto put_rsp = std::make_shared(); + put_req->set_shard_id(shard_id); + put_req->set_key(key); + put_req->set_val_len(kValueLen); + put_req->set_buf_addr(reinterpret_cast(put_descr->getAddr())); + put_req->set_buf_ofs(0); + put_req->set_buf_len(kValueLen); + for (auto rkey : put_descr->getRemoteKeys()) { + put_req->add_buf_rkey(rkey); + } + + std::atomic put_done{false}; + sirpc->SendRequest( + conn, + static_cast(KVServerRpcType::RPC_CLIENT_KV_PUT), + *put_req, + put_rsp.get(), + ctx, + [&put_done](const google::protobuf::Message *rsp, const std::shared_ptr ctx) { + ASSERT_FALSE(ctx->Failed()); + auto *put_rsp = dynamic_cast(rsp); + ASSERT_NE(put_rsp, nullptr); + EXPECT_EQ(put_rsp->ret_code(), CommonErr::OK); + put_done.store(true); + }); + while (!put_done.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + sicl::transport::MemDesc *get_descr = nullptr; + ASSERT_EQ(mempool->alloc(get_descr, kSmallBufLen), sicl::transport::Result::SICL_SUCCESS); + memset(get_descr->getAddr(), 0, kSmallBufLen); + + auto get_req = std::make_shared(); + auto get_rsp = std::make_shared(); + get_req->set_shard_id(shard_id); + get_req->set_key(key); + get_req->set_buf_addr(reinterpret_cast(get_descr->getAddr())); + get_req->set_buf_ofs(0); + get_req->set_buf_len(kSmallBufLen); + for (auto rkey : get_descr->getRemoteKeys()) { + get_req->add_buf_rkey(rkey); + } + + std::atomic get_done{false}; + sirpc->SendRequest( + conn, + static_cast(KVServerRpcType::RPC_CLIENT_KV_GET), + *get_req, + get_rsp.get(), + ctx, + [&get_done](const google::protobuf::Message *rsp, const std::shared_ptr ctx) { + ASSERT_FALSE(ctx->Failed()); + auto *get_rsp = dynamic_cast(rsp); + ASSERT_NE(get_rsp, nullptr); + EXPECT_EQ(get_rsp->ret_code(), CommonErr::InvalidArgument); + get_done.store(true); + }); + while (!get_done.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + mempool->release(put_descr); + mempool->release(get_descr); + delete sirpc; + clientDone.post(); + }); + + clientDone.wait(); + serverExit.post(); + serverPool->join(); + clientPool->join(); +} + TEST_F(KVServiceLightTest, TestHeartbeatFailureCountResetOnSuccess) { rpcServicePtr->heartbeat_failure_count_.store(4); rpcServicePtr->cm_ready_.store(true); diff --git a/tests/tools/CMakeLists.txt b/tests/tools/CMakeLists.txt new file mode 100644 index 0000000..b97615a --- /dev/null +++ b/tests/tools/CMakeLists.txt @@ -0,0 +1,16 @@ +message(STATUS "UNIT_TEST:${UNIT_TEST}") + +file(GLOB TEST_SOURCES CONFIGURE_DEPENDS + "${CMAKE_CURRENT_SOURCE_DIR}/test_*.cc" + "${CMAKE_CURRENT_SOURCE_DIR}/test_*.cpp" +) + +foreach(test_file ${TEST_SOURCES}) + get_filename_component(test_name ${test_file} NAME_WE) + add_executable(${test_name} ${test_file}) + target_include_directories(${test_name} PRIVATE ${CMAKE_SOURCE_DIR}) + target_link_libraries(${test_name} PRIVATE + gtest + gtest_main) + add_test(NAME ${test_name} COMMAND ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/${test_name}) +endforeach() diff --git a/tests/tools/test_simm_stable_test_utils.cc b/tests/tools/test_simm_stable_test_utils.cc new file mode 100644 index 0000000..af1f87f --- /dev/null +++ b/tests/tools/test_simm_stable_test_utils.cc @@ -0,0 +1,31 @@ +#include + +#include + +#include "tools/simm_stable_test_utils.h" + +namespace simm::tools::stable_test { +namespace { + +TEST(SimmStableTestUtils, BuildWorkerKeyspaceIsStableAndUniquePerWorker) { + const auto keys_a = BuildWorkerKeyspace(7, 128, 64); + const auto keys_b = BuildWorkerKeyspace(7, 128, 64); + + ASSERT_EQ(keys_a.size(), 128); + ASSERT_EQ(keys_b.size(), 128); + EXPECT_EQ(keys_a, keys_b); + + std::unordered_set seen; + for (const auto &key : keys_a) { + EXPECT_TRUE(seen.insert(key).second); + } +} + +TEST(SimmStableTestUtils, MinimumUniqueKeyLengthPreservesFullLogicalPrefix) { + const auto min_len = MinimumUniqueKeyLength(31, 199999); + const auto key = BuildStableKey(31, 199999, min_len); + EXPECT_EQ(key, "stable_t31_s199999"); +} + +} // namespace +} // namespace simm::tools::stable_test diff --git a/tools/simm_stable_test.cc b/tools/simm_stable_test.cc index 5ae305e..7e4cf73 100644 --- a/tools/simm_stable_test.cc +++ b/tools/simm_stable_test.cc @@ -15,31 +15,42 @@ */ #include +#include +#include #include +#include +#include #include #include #include #include #include -#include +#include #include +#include +#include #include #include +#include #include #include #include -#include #include #include "common/errcode/errcode_def.h" #include "common/logging/logging.h" #include "simm/simm_common.h" #include "simm/simm_kv.h" +#include "simm_stable_test_utils.h" using steady_clock_t = std::chrono::steady_clock; -using nano_ts = std::chrono::nanoseconds; +using micro_ts = std::chrono::microseconds; + +DECLARE_string(cm_primary_node_ip); + +namespace { constexpr size_t ONE_KB = 1024; constexpr size_t ONE_MB = 1024 * ONE_KB; @@ -47,39 +58,43 @@ constexpr size_t KEY_LEN_LIMIT = 256; constexpr size_t VAL_LEN_LIMIT = 4 * ONE_MB; constexpr size_t THREAD_NUM_LIMIT = 256; constexpr size_t IODEPTH_LIMIT = 2048; +constexpr uint32_t KEYSPACE_LIMIT = 1'000'000; -DEFINE_uint32(keylimit, 256, "user key lenght limit, [1B, limit]"); -DEFINE_uint32(vallimit, 4 * ONE_MB, "user val lenght limit, [1B, limit]"); +DEFINE_uint32(keylimit, 256, "user key length limit, [1B, limit]"); +DEFINE_uint32(vallimit, 4 * ONE_MB, "user value length limit, [1B, limit]"); DEFINE_uint32(threads, 32, "threads number to call kv api concurrently"); -DEFINE_uint32(time, 60, "tests run time in seconds"); +DEFINE_uint32(time, 60, "test run time in seconds"); DEFINE_uint32(iodepth, 64, "iodepth for async io in every work thread"); -DEFINE_uint32(delratio, 20, "possibility to delete kv in one sync/async task, default is 20%"); -DEFINE_uint32(oputratio, 5, "possibility to do overwrite put operation, default is 0%, namely disabled"); -DEFINE_string(iomode, "sync", "sync or async, other modes are invalid"); -DEFINE_bool(fixed_kvsize, false, "fixed key and value size by 'keylimit' 'vallimit', or not(use random sizes)"); -DEFINE_bool(batch_mode, false, "use batch mode apis or not"); // now only cover mget/mput apis -DEFINE_uint32(batch_size, 140, "requests number in batch apis, only work when batch_mode is true"); - -DECLARE_string(cm_primary_node_ip); // specify primary cm ip for tests +DEFINE_uint32(delratio, 10, "delete ratio in workload mix, percent"); +DEFINE_uint32(oputratio, 15, "overwrite put ratio when choosing put on an existing key, percent"); +DEFINE_uint32(getratio, 45, "get ratio in workload mix, percent"); +DEFINE_uint32(putratio, 35, "put ratio in workload mix, percent"); +DEFINE_uint32(existsratio, 10, "exists ratio in workload mix, percent"); +DEFINE_string(iomode, "sync", "sync or async"); +DEFINE_bool(fixed_kvsize, false, "fixed key and value size by keylimit and vallimit"); +DEFINE_bool(batch_mode, false, "use batch mode apis or not, only supported in sync mode"); +DEFINE_uint32(batch_size, 140, "requests number in batch apis when batch_mode is true"); +DEFINE_uint32(keyspace_per_thread, + 10000, + "logical keyspace size owned by each worker thread; reused to build long-running stable workload"); +DEFINE_uint32(report_interval_inSecs, 10, "periodic report interval in seconds"); +DEFINE_bool(strict_verify_exists, + false, + "if true, treat async/sync exists miss as failure; if false, count expected miss as success"); DECLARE_LOG_MODULE("simm_client"); -// FIXME(ytji) : -// 1. cover \ and " -// 2. cover special symbols, e.g. ￥《 》 ... -static const char charset[] = - "abcdefghijklmnopqrstuvwxyz" - "ABCDEFGHIJKLMNOPQRSTUVWXYZ" - "0123456789" - "~.!@#$%^&*([{}])_-+=/><,?:;'"; -static constexpr size_t charset_size = sizeof(charset) - 1; +std::atomic g_stop_requested{false}; + +void StopSignalHandler(int /*sig*/) { + g_stop_requested.store(true, std::memory_order_release); +} std::string convert_to_readable_size(uint64_t bytes_num) { - static const char *units[] = {"B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"}; + static const char *units[] = {"B", "KB", "MB", "GB", "TB", "PB", "EB"}; double size = static_cast(bytes_num); - int unit_index = 0; - - while (size >= 1024.0 && unit_index < (int)(sizeof(units) / sizeof(units[0])) - 1) { + size_t unit_index = 0; + while (size >= 1024.0 && unit_index + 1 < std::size(units)) { size /= 1024.0; ++unit_index; } @@ -89,129 +104,320 @@ std::string convert_to_readable_size(uint64_t bytes_num) { return oss.str(); } -struct ThreadStats { - void add([[maybe_unused]] nano_ts d) { - // no latency metrics needed in stable test +struct LatencyStats { + static constexpr std::array kBucketUpperUs = + {50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 50000, 100000, 200000, 500000, 1000000, UINT64_MAX}; + + void add(micro_ts d) { + const auto us = static_cast(std::max(0, d.count())); + ++count_; + total_us_ += us; + min_us_ = std::min(min_us_, us); + max_us_ = std::max(max_us_, us); + for (size_t i = 0; i < kBucketUpperUs.size(); ++i) { + if (us <= kBucketUpperUs[i]) { + ++bucket_counts_[i]; + break; + } + } } + void merge(const LatencyStats &o) { + count_ += o.count_; + total_us_ += o.total_us_; + min_us_ = std::min(min_us_, o.min_us_); + max_us_ = std::max(max_us_, o.max_us_); + for (size_t i = 0; i < bucket_counts_.size(); ++i) { + bucket_counts_[i] += o.bucket_counts_[i]; + } + } + + double avg_us() const { return count_ == 0 ? 0.0 : static_cast(total_us_) / static_cast(count_); } + + uint64_t percentile(double pct) const { + if (count_ == 0) { + return 0; + } + const uint64_t target = static_cast(std::ceil(static_cast(count_) * pct)); + uint64_t seen = 0; + for (size_t i = 0; i < bucket_counts_.size(); ++i) { + seen += bucket_counts_[i]; + if (seen >= target) { + return kBucketUpperUs[i]; + } + } + return max_us_; + } + + uint64_t count_{0}; + uint64_t total_us_{0}; + uint64_t min_us_{std::numeric_limits::max()}; + uint64_t max_us_{0}; + std::array bucket_counts_{}; +}; + +struct ThreadStats { void merge(const ThreadStats &o) { - get_ += o.get_; - get_fails_ += o.get_fails_; - get_succs_ += o.get_succs_; - mget_ += o.mget_; - mget_fails_ += o.mget_fails_; - mget_succs_ += o.mget_succs_; put_ += o.put_; put_fails_ += o.put_fails_; put_succs_ += o.put_succs_; - mput_ += o.mput_; - mput_fails_ += o.mput_fails_; - mput_succs_ += o.mput_succs_; overwrite_put_ += o.overwrite_put_; overwrite_put_fails_ += o.overwrite_put_fails_; overwrite_put_succs_ += o.overwrite_put_succs_; + get_ += o.get_; + get_fails_ += o.get_fails_; + get_succs_ += o.get_succs_; exists_ += o.exists_; exists_fails_ += o.exists_fails_; exists_succs_ += o.exists_succs_; - mexists_ += o.mexists_; - mexists_fails_ += o.mexists_fails_; - mexists_succs_ += o.mexists_succs_; del_ += o.del_; del_fails_ += o.del_fails_; del_succs_ += o.del_succs_; + mput_ += o.mput_; + mput_fails_ += o.mput_fails_; + mput_succs_ += o.mput_succs_; + mget_ += o.mget_; + mget_fails_ += o.mget_fails_; + mget_succs_ += o.mget_succs_; data_match_ += o.data_match_; data_mismatch_ += o.data_mismatch_; + expected_miss_ += o.expected_miss_; + submit_fails_ += o.submit_fails_; put_size_bytes += o.put_size_bytes; get_size_bytes += o.get_size_bytes; + latency_.merge(o.latency_); + } + + uint64_t total_ops() const { return put_ + overwrite_put_ + get_ + exists_ + del_ + mput_ + mget_; } + + uint64_t total_failures() const { + return put_fails_ + overwrite_put_fails_ + get_fails_ + exists_fails_ + del_fails_ + mput_fails_ + mget_fails_ + + submit_fails_ + data_mismatch_; } - uint64_t get_{0}; - uint64_t get_fails_{0}; - uint64_t get_succs_{0}; - uint64_t mget_{0}; - uint64_t mget_fails_{0}; - uint64_t mget_succs_{0}; uint64_t put_{0}; uint64_t put_fails_{0}; uint64_t put_succs_{0}; - uint64_t mput_{0}; - uint64_t mput_fails_{0}; - uint64_t mput_succs_{0}; uint64_t overwrite_put_{0}; uint64_t overwrite_put_fails_{0}; uint64_t overwrite_put_succs_{0}; + uint64_t get_{0}; + uint64_t get_fails_{0}; + uint64_t get_succs_{0}; uint64_t exists_{0}; uint64_t exists_fails_{0}; uint64_t exists_succs_{0}; - uint64_t mexists_{0}; - uint64_t mexists_fails_{0}; - uint64_t mexists_succs_{0}; uint64_t del_{0}; uint64_t del_fails_{0}; uint64_t del_succs_{0}; + uint64_t mput_{0}; + uint64_t mput_fails_{0}; + uint64_t mput_succs_{0}; + uint64_t mget_{0}; + uint64_t mget_fails_{0}; + uint64_t mget_succs_{0}; uint64_t data_match_{0}; uint64_t data_mismatch_{0}; + uint64_t expected_miss_{0}; + uint64_t submit_fails_{0}; uint64_t put_size_bytes{0}; uint64_t get_size_bytes{0}; + LatencyStats latency_{}; }; -void print_stats(const ThreadStats &s, uint32_t threads) { - std::cout << "++++++++++++++++++++++++++++ STABLE TEST REPORT +++++++++++++++++++++++++\n" - << "ThreadID : " << threads << "\n" - << "GetCnt : " << s.get_ << "\n" - << "GetFails : " << s.get_fails_ << "\n" - << "GetSuccs : " << s.get_succs_ << "\n" - << "MGetCnt : " << s.mget_ << "\n" - << "MGetFails : " << s.mget_fails_ << "\n" - << "MGetSuccs : " << s.mget_succs_ << "\n" - << "PutCnt : " << s.put_ << "\n" - << "PutFails : " << s.put_fails_ << "\n" - << "PutSuccs : " << s.put_succs_ << "\n" - << "MPutCnt : " << s.mput_ << "\n" - << "MPutFails : " << s.mput_fails_ << "\n" - << "MPutSuccs : " << s.mput_succs_ << "\n" - << "OPutCnt : " << s.overwrite_put_ << "\n" - << "OPutFails : " << s.overwrite_put_fails_ << "\n" - << "OPutSuccs : " << s.overwrite_put_succs_ << "\n" - << "ExistCnt : " << s.exists_ << "\n" - << "ExistFails : " << s.exists_fails_ << "\n" - << "ExistSuccs : " << s.exists_succs_ << "\n" - << "MExistCnt : " << s.mexists_ << "\n" - << "MExistFails : " << s.mexists_fails_ << "\n" - << "MExistSuccs : " << s.mexists_succs_ << "\n" - << "DelCnt : " << s.del_ << "\n" - << "DelFails : " << s.del_fails_ << "\n" - << "DelSuccs : " << s.del_succs_ << "\n" - << "DataMatch : " << s.data_match_ << "\n" - << "DataMismatch : " << s.data_mismatch_ << "\n" - << "PutSize : " << convert_to_readable_size(s.put_size_bytes) << "\n" - << "GetSize : " << convert_to_readable_size(s.get_size_bytes) << "\n" +struct ValueSpec { + bool exists{false}; + uint32_t size{0}; + uint64_t seed{0}; +}; + +enum class OpType { + Put, + Get, + Exists, + Delete, +}; + +struct WorkerRuntime { + explicit WorkerRuntime(uint32_t tid, size_t keyspace, size_t key_len_limit) + : slots(keyspace), keys(simm::tools::stable_test::BuildWorkerKeyspace(tid, keyspace, key_len_limit)) {} + + mutable std::mutex mutex; + std::condition_variable cv; + ThreadStats stats; + std::vector slots; + std::vector keys; + size_t live_keys{0}; + size_t inflight{0}; +}; + +struct PendingSingleOp { + OpType op{OpType::Put}; + bool overwrite{false}; + size_t slot{0}; + std::string key; + ValueSpec previous_spec; + ValueSpec target_spec; + std::shared_ptr data_holder; + steady_clock_t::time_point start_ts; +}; + +uint32_t choose_size(uint32_t limit) { + return FLAGS_fixed_kvsize ? limit : folly::Random::rand32(1, limit + 1); +} + +void FillBufferFromSeed(std::span buf, uint64_t seed) { + uint64_t state = seed ^ 0x9E3779B97F4A7C15ULL; + for (size_t pos = 0; pos < buf.size(); ++pos) { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + buf[pos] = static_cast(state & 0xFFU); + } +} + +bool VerifyBufferFromSeed(std::span buf, uint64_t seed) { + uint64_t state = seed ^ 0x9E3779B97F4A7C15ULL; + for (size_t pos = 0; pos < buf.size(); ++pos) { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + if (buf[pos] != static_cast(state & 0xFFU)) { + return false; + } + } + return true; +} + +void RecordLatency(ThreadStats &stats, steady_clock_t::time_point start_ts) { + stats.latency_.add(std::chrono::duration_cast(steady_clock_t::now() - start_ts)); +} + +void print_stats(const ThreadStats &s, uint32_t threads, uint64_t elapsed_secs, bool is_delta) { + const double qps = elapsed_secs == 0 ? 0.0 : static_cast(s.total_ops()) / static_cast(elapsed_secs); + const double throughput_mb = elapsed_secs == 0 ? 0.0 + : static_cast(s.put_size_bytes + s.get_size_bytes) / + static_cast(ONE_MB) / static_cast(elapsed_secs); + + std::cout << "++++++++++++++++++++++++++++ " << (is_delta ? "STABLE TEST DELTA" : "STABLE TEST REPORT") + << " +++++++++++++++++++++++++\n" + << "Threads : " << threads << "\n" + << "ElapsedSecs : " << elapsed_secs << "\n" + << "TotalOps : " << s.total_ops() << "\n" + << "Failures : " << s.total_failures() << "\n" + << "SubmitFails : " << s.submit_fails_ << "\n" + << "PutCnt : " << s.put_ << "\n" + << "PutFails : " << s.put_fails_ << "\n" + << "PutSuccs : " << s.put_succs_ << "\n" + << "OverwriteCnt : " << s.overwrite_put_ << "\n" + << "OverwriteFails : " << s.overwrite_put_fails_ << "\n" + << "OverwriteSuccs : " << s.overwrite_put_succs_ << "\n" + << "GetCnt : " << s.get_ << "\n" + << "GetFails : " << s.get_fails_ << "\n" + << "GetSuccs : " << s.get_succs_ << "\n" + << "ExistsCnt : " << s.exists_ << "\n" + << "ExistsFails : " << s.exists_fails_ << "\n" + << "ExistsSuccs : " << s.exists_succs_ << "\n" + << "DeleteCnt : " << s.del_ << "\n" + << "DeleteFails : " << s.del_fails_ << "\n" + << "DeleteSuccs : " << s.del_succs_ << "\n" + << "MPutCnt : " << s.mput_ << "\n" + << "MPutFails : " << s.mput_fails_ << "\n" + << "MPutSuccs : " << s.mput_succs_ << "\n" + << "MGetCnt : " << s.mget_ << "\n" + << "MGetFails : " << s.mget_fails_ << "\n" + << "MGetSuccs : " << s.mget_succs_ << "\n" + << "DataMatch : " << s.data_match_ << "\n" + << "DataMismatch : " << s.data_mismatch_ << "\n" + << "ExpectedMiss : " << s.expected_miss_ << "\n" + << "PutSize : " << convert_to_readable_size(s.put_size_bytes) << "\n" + << "GetSize : " << convert_to_readable_size(s.get_size_bytes) << "\n" + << "QPS : " << std::fixed << std::setprecision(2) << qps << "\n" + << "ThroughputMiB/s : " << throughput_mb << "\n" + << "AvgLatencyUs : " << s.latency_.avg_us() << "\n" + << "P50LatencyUs : " << s.latency_.percentile(0.50) << "\n" + << "P95LatencyUs : " << s.latency_.percentile(0.95) << "\n" + << "P99LatencyUs : " << s.latency_.percentile(0.99) << "\n" + << "MaxLatencyUs : " << s.latency_.max_us_ << "\n" << std::endl; } void usage() { - std::cout - << "simm_stable_test --iomode=[sync|async] --keylimit=100 --vallimit=8192 --threads=32 --iodepth=128 --time=3600" - << std::endl; + std::cout << "simm_stable_test --iomode=[sync|async] --threads=32 --time=3600 --keyspace_per_thread=10000 " + "--getratio=45 --putratio=35 --existsratio=10 --delratio=10 --iodepth=128" + << std::endl; +} + +void PrintArgError(const std::string &msg) { + std::cerr << "[simm_stable_test] invalid args: " << msg << std::endl; } void check_args() { - bool args_valid{true}; + bool args_valid = true; - if (FLAGS_keylimit < 1 || FLAGS_keylimit > KEY_LEN_LIMIT) { + if (FLAGS_cm_primary_node_ip.empty()) { + MLOG_ERROR("cm_primary_node_ip is empty"); + PrintArgError("cm_primary_node_ip must be set"); + args_valid = false; + } else if (FLAGS_keylimit < 1 || FLAGS_keylimit > KEY_LEN_LIMIT) { MLOG_ERROR("Invalid key size upper limit : {}B, should in [1B, {}B]", FLAGS_keylimit, KEY_LEN_LIMIT); + PrintArgError("keylimit out of range"); args_valid = false; } else if (FLAGS_vallimit < 1 || FLAGS_vallimit > VAL_LEN_LIMIT) { MLOG_ERROR("Invalid val size upper limit : {}B, should in [1B, {}B]", FLAGS_vallimit, VAL_LEN_LIMIT); + PrintArgError("vallimit out of range"); args_valid = false; } else if (FLAGS_threads < 1 || FLAGS_threads > THREAD_NUM_LIMIT) { MLOG_ERROR("Invalid threads number : {}, should in [1, {}]", FLAGS_threads, THREAD_NUM_LIMIT); + PrintArgError("threads out of range"); args_valid = false; } else if (FLAGS_iodepth < 1 || FLAGS_iodepth > IODEPTH_LIMIT) { MLOG_ERROR("Invalid iodepth : {}, should in [1, {}]", FLAGS_iodepth, IODEPTH_LIMIT); + PrintArgError("iodepth out of range"); args_valid = false; } else if (FLAGS_iomode != "sync" && FLAGS_iomode != "async") { MLOG_ERROR("Invalid iomode type : {}, should be sync or async", FLAGS_iomode); + PrintArgError("iomode must be sync or async"); + args_valid = false; + } else if (FLAGS_keyspace_per_thread < 1 || FLAGS_keyspace_per_thread > KEYSPACE_LIMIT) { + MLOG_ERROR("Invalid keyspace_per_thread : {}, should in [1, {}]", FLAGS_keyspace_per_thread, KEYSPACE_LIMIT); + PrintArgError("keyspace_per_thread out of range"); + args_valid = false; + } else if (FLAGS_batch_mode && FLAGS_iomode != "sync") { + MLOG_ERROR("batch_mode currently only supports sync iomode"); + PrintArgError("batch_mode currently only supports sync iomode"); + args_valid = false; + } else if (FLAGS_batch_mode && FLAGS_batch_size < 1) { + MLOG_ERROR("batch_size should be >= 1"); + PrintArgError("batch_size should be >= 1"); + args_valid = false; + } else if (FLAGS_time < 1) { + MLOG_ERROR("time should be >= 1"); + PrintArgError("time should be >= 1"); + args_valid = false; + } else if (FLAGS_report_interval_inSecs < 1) { + MLOG_ERROR("report_interval_inSecs should be >= 1"); + PrintArgError("report_interval_inSecs should be >= 1"); + args_valid = false; + } + + if (FLAGS_getratio + FLAGS_putratio + FLAGS_existsratio + FLAGS_delratio != 100) { + MLOG_ERROR("getratio({}) + putratio({}) + existsratio({}) + delratio({}) must equal 100", + FLAGS_getratio, + FLAGS_putratio, + FLAGS_existsratio, + FLAGS_delratio); + PrintArgError("getratio + putratio + existsratio + delratio must equal 100; oputratio is part of putratio"); + args_valid = false; + } + + const auto min_key_len = simm::tools::stable_test::MinimumUniqueKeyLength( + FLAGS_threads == 0 ? 0 : FLAGS_threads - 1, FLAGS_keyspace_per_thread == 0 ? 0 : FLAGS_keyspace_per_thread - 1); + if (FLAGS_keylimit < min_key_len) { + MLOG_ERROR("Invalid key size upper limit : {}B, should be >= {}B to keep per-slot keys unique", + FLAGS_keylimit, + min_key_len); + PrintArgError("keylimit is too small to keep per-slot keys unique"); args_valid = false; } @@ -221,313 +427,672 @@ void check_args() { } } -// used to generate key or value with random contents -std::string generate_random_string(uint32_t sz_limit, bool is_key = true) { - // thread_local std::mt19937_64 rng{std::random_device{}()}; - // std::uniform_int_distribution dist(0, UINT64_MAX); - std::string s; - uint32_t pos = 0; - // uint32_t limit = folly::Random::rand32(1, sz_limit + 1); - s.resize(sz_limit); - if (is_key) { - while (pos < sz_limit) { - // uint64_t r = dist(rng); - uint64_t r = folly::Random::rand64(); - for (int j = 0; j < 10 && pos < sz_limit; ++j) { - s[pos++] = charset[r % charset_size]; - r /= charset_size; - } +bool RunStartupSelfCheck(simm::clnt::KVStore &kvstore) { + const std::string key = + "stable_test_canary_" + std::to_string(static_cast(steady_clock_t::now().time_since_epoch().count())); + constexpr uint32_t kCanarySize = 128; + constexpr uint64_t kCanarySeed = 0x13572468ABCDEF01ULL; + + try { + auto put_data = kvstore.Allocate(kCanarySize); + FillBufferFromSeed(put_data.AsRef(), kCanarySeed); + simm::clnt::DataView put_view(put_data); + const auto put_rc = kvstore.Put(key, put_view); + if (put_rc != CommonErr::OK) { + std::cerr << "[simm_stable_test] startup self-check failed: put rc=" << put_rc << std::endl; + return false; + } + + const auto exists_rc = kvstore.Exists(key); + if (exists_rc != CommonErr::OK) { + std::cerr << "[simm_stable_test] startup self-check failed: exists rc=" << exists_rc << std::endl; + return false; + } + + auto get_data = kvstore.Allocate(kCanarySize); + simm::clnt::DataView get_view(get_data); + const auto get_rc = kvstore.Get(key, get_view); + if (get_rc != static_cast(kCanarySize) || + !VerifyBufferFromSeed(std::span(get_data.AsRef().data(), kCanarySize), kCanarySeed)) { + std::cerr << "[simm_stable_test] startup self-check failed: get rc=" << get_rc << std::endl; + return false; } - } else { // generate value - const size_t step = sizeof(uint64_t); - while (pos + step <= sz_limit) { - uint64_t r = folly::Random::rand64(); - std::memcpy(&s[pos], &r, step); - pos += step; + + const auto del_rc = kvstore.Delete(key); + if (del_rc != CommonErr::OK) { + std::cerr << "[simm_stable_test] startup self-check failed: delete rc=" << del_rc << std::endl; + return false; } - if (pos < sz_limit) { - uint64_t r = folly::Random::rand64(); - std::memcpy(&s[pos], &r, sz_limit - pos); + return true; + } catch (const std::exception &ex) { + std::cerr << "[simm_stable_test] startup self-check threw exception: " << ex.what() << std::endl; + return false; + } +} + +OpType ChooseOperation(const WorkerRuntime &runtime) { + const uint32_t dice = folly::Random::rand32(0, 100); + const bool no_live_keys = runtime.live_keys == 0; + if (no_live_keys) { + return OpType::Put; + } + if (dice < FLAGS_getratio) { + return OpType::Get; + } + if (dice < FLAGS_getratio + FLAGS_putratio) { + return OpType::Put; + } + if (dice < FLAGS_getratio + FLAGS_putratio + FLAGS_existsratio) { + return OpType::Exists; + } + return OpType::Delete; +} + +size_t PickExistingSlot(const WorkerRuntime &runtime) { + size_t start = folly::Random::rand32(0, static_cast(runtime.slots.size())); + for (size_t i = 0; i < runtime.slots.size(); ++i) { + size_t idx = (start + i) % runtime.slots.size(); + if (runtime.slots[idx].exists) { + return idx; } } - return s; + return runtime.slots.size(); } -int main(int argc, char **argv) { - gflags::ParseCommandLineFlags(&argc, &argv, true); - folly::Init init(&argc, &argv); +size_t PickPutSlot(const WorkerRuntime &runtime, bool *overwrite) { + if (runtime.live_keys > 0 && folly::Random::rand32(0, 100) < FLAGS_oputratio) { + size_t idx = PickExistingSlot(runtime); + if (idx != runtime.slots.size()) { + *overwrite = true; + return idx; + } + } -#ifdef NDEBUG - simm::logging::LogConfig clnt_log_config = simm::logging::LogConfig{"/tmp/simm_clnt.log", "INFO"}; -#else - simm::logging::LogConfig clnt_log_config = simm::logging::LogConfig{"/tmp/simm_clnt.log", "DEBUG"}; -#endif - simm::logging::LoggerManager::Instance().UpdateConfig("simm_client", clnt_log_config); + size_t start = folly::Random::rand32(0, static_cast(runtime.slots.size())); + for (size_t i = 0; i < runtime.slots.size(); ++i) { + size_t idx = (start + i) % runtime.slots.size(); + if (!runtime.slots[idx].exists) { + *overwrite = false; + return idx; + } + } - check_args(); + *overwrite = true; + return start; +} - MLOG_INFO("++++++++++++++++++++++++++++++++++++++++ Start SIMM_STABLE_TEST ++++++++++++++++++++++++++++++++++++++++"); - MLOG_INFO( - "Test Args: key_size_limit:{}B | val_size_limit:{}B | threads:{} | iomode:{} | fixed_kv:{} | batchmode:{} | " - "batchsize:{}", - FLAGS_keylimit, - FLAGS_vallimit, - FLAGS_threads, - FLAGS_iomode, - FLAGS_fixed_kvsize ? "T" : "F", - FLAGS_batch_mode ? "T" : "F", - FLAGS_batch_size); - - auto test_binary_start_ts = steady_clock_t::now(); - - // +++++++++++++++++++++++++++++++++++ - // Phase : env init - // +++++++++++++++++++++++++++++++++++ - auto simm_kvstore = std::make_unique(); - if (simm_kvstore == nullptr) { - MLOG_ERROR("Make unique_ptr for simm KVStore failed, no memory!"); - std::exit(ENOMEM); - } - - // test thread pool init - folly::CPUThreadPoolExecutor thread_pool(FLAGS_threads); - - // +++++++++++++++++++++++++++++++++++ - // Phase : test - // +++++++++++++++++++++++++++++++++++ - std::atomic stop_threads{false}; - std::vector test_stats_vec(FLAGS_threads); - std::latch kvio_latch{FLAGS_threads}; - - auto sync_io_task = [&](uint32_t tid) -> void { - auto &thread_stat = test_stats_vec[tid]; - while (!stop_threads.load()) { - // generate key & value for current task - size_t k_sz = FLAGS_fixed_kvsize ? FLAGS_keylimit : folly::Random::rand32(1, FLAGS_keylimit + 1); - size_t v_sz = FLAGS_fixed_kvsize ? FLAGS_vallimit : folly::Random::rand32(1, FLAGS_vallimit + 1); - std::vector key_vec; - std::vector data_vec; - std::vector dataview_vec; - size_t kv_cnt = FLAGS_batch_mode ? FLAGS_batch_size : 1; - for (size_t i = 0; i < kv_cnt; i++) { - auto k = generate_random_string(k_sz); - key_vec.emplace_back(k); - auto v = simm_kvstore->Allocate(v_sz); - auto random_value_str = generate_random_string(v_sz, false); - std::memcpy(v.AsRef().data(), random_value_str.c_str(), v_sz); - simm::clnt::DataView dv(v); - dataview_vec.emplace_back(dv); - data_vec.emplace_back(std::move(v)); +ThreadStats SnapshotStats(const std::vector> &workers) { + ThreadStats aggregated; + for (const auto &worker : workers) { + std::lock_guard lock(worker->mutex); + aggregated.merge(worker->stats); + } + return aggregated; +} + +void RecordExpectedExistsResult(ThreadStats &stats, bool expected_exists, int rc) { + ++stats.exists_; + if (expected_exists) { + if (rc == CommonErr::OK) { + ++stats.exists_succs_; + } else { + ++stats.exists_fails_; + } + return; + } + + if (FLAGS_strict_verify_exists) { + if (rc != CommonErr::OK) { + ++stats.exists_succs_; + ++stats.expected_miss_; + } else { + ++stats.exists_fails_; + } + } else { + ++stats.exists_succs_; + if (rc != CommonErr::OK) { + ++stats.expected_miss_; + } + } +} + +void RunSyncSingleOp(uint32_t tid, simm::clnt::KVStore &kvstore, WorkerRuntime &runtime) { + PendingSingleOp op; + { + std::lock_guard lock(runtime.mutex); + op.op = ChooseOperation(runtime); + if (op.op == OpType::Put) { + op.slot = PickPutSlot(runtime, &op.overwrite); + op.previous_spec = runtime.slots[op.slot]; + op.target_spec.exists = true; + op.target_spec.size = choose_size(FLAGS_vallimit); + op.target_spec.seed = folly::Random::rand64(); + } else if (op.op == OpType::Get || op.op == OpType::Delete) { + op.slot = PickExistingSlot(runtime); + if (op.slot == runtime.slots.size()) { + op.op = OpType::Put; + op.slot = PickPutSlot(runtime, &op.overwrite); + op.previous_spec = runtime.slots[op.slot]; + op.target_spec.exists = true; + op.target_spec.size = choose_size(FLAGS_vallimit); + op.target_spec.seed = folly::Random::rand64(); + } else { + op.previous_spec = runtime.slots[op.slot]; } + } else { + op.slot = folly::Random::rand32(0, static_cast(runtime.slots.size())); + op.previous_spec = runtime.slots[op.slot]; + } + op.key = runtime.keys[op.slot]; + } - if (FLAGS_batch_mode) { // batch mode apis - auto bput_rets = simm_kvstore->MPut(key_vec, dataview_vec); - bool bput_succeed = true; - uint32_t idx = 0; - for (const auto ret : bput_rets) { - if (ret == CommonErr::OK) { - thread_stat.put_succs_++; - thread_stat.put_size_bytes += v_sz; - } else { - thread_stat.put_fails_++; - bput_succeed = false; - MLOG_ERROR("Failed to put key({}) in sync mput IO, kv_ret:{}", key_vec.at(idx), ret); - } - idx++; - } - thread_stat.put_ += FLAGS_batch_size; - thread_stat.mput_++; - if (bput_succeed) { - thread_stat.mput_succs_++; - } else { - thread_stat.mput_fails_++; + op.start_ts = steady_clock_t::now(); + if (op.op == OpType::Put) { + auto data = kvstore.Allocate(op.target_spec.size); + FillBufferFromSeed(data.AsRef(), op.target_spec.seed); + simm::clnt::DataView dv(data); + const int16_t rc = kvstore.Put(op.key, dv); + std::lock_guard lock(runtime.mutex); + if (op.overwrite) { + ++runtime.stats.overwrite_put_; + } else { + ++runtime.stats.put_; + } + RecordLatency(runtime.stats, op.start_ts); + if (rc == CommonErr::OK) { + if (op.overwrite) { + ++runtime.stats.overwrite_put_succs_; + } else { + ++runtime.stats.put_succs_; + if (!op.previous_spec.exists) { + ++runtime.live_keys; } + } + runtime.stats.put_size_bytes += op.target_spec.size; + runtime.slots[op.slot] = op.target_spec; + } else { + if (op.overwrite) { + ++runtime.stats.overwrite_put_fails_; + } else { + ++runtime.stats.put_fails_; + } + } + return; + } - // FIXME(ytji) : skip MExists test for api is unavailable now - for (const auto &k : key_vec) { - auto exist_ret = simm_kvstore->Exists(k); - thread_stat.exists_++; - if (exist_ret == CommonErr::OK) { - thread_stat.exists_succs_++; - } else { - thread_stat.exists_fails_++; - MLOG_ERROR("Failed to lookup key({}) in sync exists IO, kv_ret:{}", k, exist_ret); - } - } + if (op.op == OpType::Get) { + auto data = kvstore.Allocate(op.previous_spec.size); + simm::clnt::DataView dv(data); + const int32_t rc = kvstore.Get(op.key, dv); + std::lock_guard lock(runtime.mutex); + ++runtime.stats.get_; + RecordLatency(runtime.stats, op.start_ts); + if (rc == static_cast(op.previous_spec.size) && + VerifyBufferFromSeed(std::span(data.AsRef().data(), op.previous_spec.size), + op.previous_spec.seed)) { + ++runtime.stats.get_succs_; + ++runtime.stats.data_match_; + runtime.stats.get_size_bytes += op.previous_spec.size; + } else { + ++runtime.stats.get_fails_; + ++runtime.stats.data_mismatch_; + } + return; + } - // FIXME(ytji) : skip overwrite IO in batch scenarios + if (op.op == OpType::Exists) { + const int16_t rc = kvstore.Exists(op.key); + std::lock_guard lock(runtime.mutex); + RecordLatency(runtime.stats, op.start_ts); + RecordExpectedExistsResult(runtime.stats, op.previous_spec.exists, rc); + return; + } - std::vector mget_data_vec; - std::vector mget_dataview_vec; - for (size_t i = 0; i < kv_cnt; i++) { - auto mget_v = simm_kvstore->Allocate(v_sz); - simm::clnt::DataView mget_dv(mget_v); - mget_dataview_vec.emplace_back(mget_dv); - mget_data_vec.emplace_back(std::move(mget_v)); - } - auto bget_rets = simm_kvstore->MGet(key_vec, mget_dataview_vec); - bool bget_succeed = true; - idx = 0; - for (const auto ret : bget_rets) { - if (ret >= 0) { - thread_stat.get_succs_++; - thread_stat.get_size_bytes += v_sz; - // do mem check for every succeed get op - if (std::memcmp(data_vec.at(idx).AsRef().data(), mget_dataview_vec.at(idx).AsRef().data(), v_sz) != 0) { - thread_stat.data_mismatch_++; - MLOG_ERROR("Data mismatch for key({}) in sync mget IO", key_vec.at(idx)); - } else { - thread_stat.data_match_++; - } - } else { - thread_stat.get_fails_++; - bget_succeed = false; - MLOG_ERROR("Failed to get key({}) in sync mget IO, kv_ret:{}", key_vec.at(idx), ret); - } - idx++; - } - thread_stat.get_ += FLAGS_batch_size; - thread_stat.mget_++; - if (bget_succeed) { - thread_stat.mget_succs_++; - } else { - thread_stat.mget_fails_++; - } + const int16_t rc = kvstore.Delete(op.key); + std::lock_guard lock(runtime.mutex); + ++runtime.stats.del_; + RecordLatency(runtime.stats, op.start_ts); + if (rc == CommonErr::OK) { + ++runtime.stats.del_succs_; + if (runtime.slots[op.slot].exists) { + runtime.slots[op.slot].exists = false; + runtime.slots[op.slot].size = 0; + runtime.slots[op.slot].seed = 0; + if (runtime.live_keys > 0) { + --runtime.live_keys; + } + } + } else { + ++runtime.stats.del_fails_; + } +} - if (folly::Random::rand32(0, 100) < FLAGS_delratio) { - // FIXME(ytji) : skip MDelete test for api is unavailable now - for (const auto &k : key_vec) { - auto del_ret = simm_kvstore->Delete(k); - thread_stat.del_++; - if (del_ret != 0) { - thread_stat.del_fails_++; - MLOG_ERROR("Failed to delete key({}) in sync delete IO, kv_ret:{}", k, del_ret); - } else { - thread_stat.del_succs_++; - } - } +void RunSyncBatchOp(uint32_t tid, simm::clnt::KVStore &kvstore, WorkerRuntime &runtime) { + const size_t batch_sz = FLAGS_batch_size; + std::vector slots; + std::vector keys; + std::vector specs; + std::vector payloads; + std::vector put_views; + std::vector get_payloads; + std::vector get_views; + + slots.reserve(batch_sz); + keys.reserve(batch_sz); + specs.reserve(batch_sz); + payloads.reserve(batch_sz); + put_views.reserve(batch_sz); + get_payloads.reserve(batch_sz); + get_views.reserve(batch_sz); + + { + std::lock_guard lock(runtime.mutex); + for (size_t i = 0; i < batch_sz; ++i) { + bool overwrite = false; + const auto slot = PickPutSlot(runtime, &overwrite); + ValueSpec spec; + spec.exists = true; + spec.size = choose_size(FLAGS_vallimit); + spec.seed = folly::Random::rand64(); + slots.push_back(slot); + keys.push_back(runtime.keys[slot]); + specs.push_back(spec); + } + } + + for (const auto &spec : specs) { + auto data = kvstore.Allocate(spec.size); + FillBufferFromSeed(data.AsRef(), spec.seed); + payloads.emplace_back(std::move(data)); + } + for (auto &data : payloads) { + put_views.emplace_back(data); + } + + auto start_ts = steady_clock_t::now(); + auto put_rets = kvstore.MPut(keys, put_views); + { + std::lock_guard lock(runtime.mutex); + ++runtime.stats.mput_; + runtime.stats.put_ += batch_sz; + RecordLatency(runtime.stats, start_ts); + bool all_ok = true; + for (size_t i = 0; i < put_rets.size(); ++i) { + if (put_rets[i] == CommonErr::OK) { + ++runtime.stats.put_succs_; + runtime.stats.put_size_bytes += specs[i].size; + if (!runtime.slots[slots[i]].exists) { + ++runtime.live_keys; } + runtime.slots[slots[i]] = specs[i]; } else { - int32_t kv_ret = simm_kvstore->Put(key_vec.at(0), dataview_vec.at(0)); - thread_stat.put_++; - if (kv_ret != 0) { - thread_stat.put_fails_++; - MLOG_ERROR("Failed to put key({}) in sync put IO, kv_ret:{}", key_vec.at(0), kv_ret); - continue; - } else { - thread_stat.put_succs_++; - thread_stat.put_size_bytes += v_sz; - } + ++runtime.stats.put_fails_; + all_ok = false; + } + } + if (all_ok) { + ++runtime.stats.mput_succs_; + } else { + ++runtime.stats.mput_fails_; + } + } - kv_ret = simm_kvstore->Exists(key_vec.at(0)); - thread_stat.exists_++; - if (kv_ret != 0) { - thread_stat.exists_fails_++; - MLOG_ERROR("Failed to lookup key({}) in sync exists IO, kv_ret:{}", key_vec.at(0), kv_ret); - continue; - } else { - thread_stat.exists_succs_++; - } + for (const auto &spec : specs) { + auto data = kvstore.Allocate(spec.size); + get_payloads.emplace_back(std::move(data)); + } + for (auto &data : get_payloads) { + get_views.emplace_back(data); + } - if (folly::Random::rand32(0, 100) < FLAGS_oputratio) { - size_t overwrite_v_sz = folly::Random::rand32(1, FLAGS_vallimit + 1); - auto overwrite_v = simm_kvstore->Allocate(overwrite_v_sz); - auto random_overwrite_value_str = generate_random_string(overwrite_v_sz, false); - std::memcpy(overwrite_v.AsRef().data(), random_overwrite_value_str.c_str(), overwrite_v_sz); - simm::clnt::DataView overwrite_dv(overwrite_v); - // overwrite put - kv_ret = simm_kvstore->Put(key_vec.at(0), overwrite_dv); - thread_stat.overwrite_put_++; - if (kv_ret != 0) { - thread_stat.overwrite_put_fails_++; - MLOG_ERROR("Failed to overwrite key({}) in sync put IO, kv_ret:{}", key_vec.at(0), kv_ret); - continue; - } else { - thread_stat.overwrite_put_succs_++; - thread_stat.put_size_bytes += overwrite_v_sz; - // move overwrite kv size and data buffer for later data check - v_sz = overwrite_v_sz; - data_vec.at(0) = std::move(overwrite_v); - } - } + start_ts = steady_clock_t::now(); + auto get_rets = kvstore.MGet(keys, get_views); + { + std::lock_guard lock(runtime.mutex); + ++runtime.stats.mget_; + runtime.stats.get_ += batch_sz; + RecordLatency(runtime.stats, start_ts); + bool all_ok = true; + for (size_t i = 0; i < get_rets.size(); ++i) { + if (get_rets[i] == static_cast(specs[i].size) && + VerifyBufferFromSeed(std::span(get_payloads[i].AsRef().data(), specs[i].size), specs[i].seed)) { + ++runtime.stats.get_succs_; + ++runtime.stats.data_match_; + runtime.stats.get_size_bytes += specs[i].size; + } else { + ++runtime.stats.get_fails_; + ++runtime.stats.data_mismatch_; + all_ok = false; + } + } + if (all_ok) { + ++runtime.stats.mget_succs_; + } else { + ++runtime.stats.mget_fails_; + } + } +} + +void SubmitAsyncSingleOp(uint32_t tid, + simm::clnt::KVStore &kvstore, + std::shared_ptr runtime, + std::atomic &stop_threads) { + PendingSingleOp op; + { + std::unique_lock lock(runtime->mutex); + runtime->cv.wait( + lock, [&]() { return stop_threads.load(std::memory_order_acquire) || runtime->inflight < FLAGS_iodepth; }); + if (stop_threads.load(std::memory_order_acquire)) { + return; + } + ++runtime->inflight; + op.op = ChooseOperation(*runtime); + if (op.op == OpType::Put) { + op.slot = PickPutSlot(*runtime, &op.overwrite); + op.previous_spec = runtime->slots[op.slot]; + op.target_spec.exists = true; + op.target_spec.size = choose_size(FLAGS_vallimit); + op.target_spec.seed = folly::Random::rand64(); + } else if (op.op == OpType::Get || op.op == OpType::Delete) { + op.slot = PickExistingSlot(*runtime); + if (op.slot == runtime->slots.size()) { + op.op = OpType::Put; + op.slot = PickPutSlot(*runtime, &op.overwrite); + op.previous_spec = runtime->slots[op.slot]; + op.target_spec.exists = true; + op.target_spec.size = choose_size(FLAGS_vallimit); + op.target_spec.seed = folly::Random::rand64(); + } else { + op.previous_spec = runtime->slots[op.slot]; + } + } else { + op.slot = folly::Random::rand32(0, static_cast(runtime->slots.size())); + op.previous_spec = runtime->slots[op.slot]; + } + op.key = runtime->keys[op.slot]; + } - auto get_v = simm_kvstore->Allocate(v_sz); - simm::clnt::DataView get_dv(get_v); - kv_ret = simm_kvstore->Get(key_vec.at(0), get_dv); - thread_stat.get_++; - // Get() will return kv length or error codes( < 0) - if (kv_ret <= 0) { - thread_stat.get_fails_++; - MLOG_ERROR("Failed to get key({}) in sync get IO, kv_ret:{}", key_vec.at(0), kv_ret); - continue; + op.start_ts = steady_clock_t::now(); + + if (op.op == OpType::Put) { + auto data_holder = std::make_shared(kvstore.Allocate(op.target_spec.size)); + FillBufferFromSeed(data_holder->AsRef(), op.target_spec.seed); + simm::clnt::DataView put_view(*data_holder); + auto rc = kvstore.AsyncPut(op.key, put_view, [runtime, op, data_holder](int result) { + std::lock_guard lock(runtime->mutex); + if (op.overwrite) { + ++runtime->stats.overwrite_put_; + } else { + ++runtime->stats.put_; + } + RecordLatency(runtime->stats, op.start_ts); + if (result == CommonErr::OK) { + if (op.overwrite) { + ++runtime->stats.overwrite_put_succs_; } else { - thread_stat.get_succs_++; - thread_stat.get_size_bytes += v_sz; + ++runtime->stats.put_succs_; + if (!op.previous_spec.exists) { + ++runtime->live_keys; + } } - - if (std::memcmp(data_vec.at(0).AsRef().data(), get_v.AsRef().data(), v_sz) != 0) { - thread_stat.data_mismatch_++; - MLOG_ERROR("Data mismatch for key({}) in sync_io_task", key_vec.at(0)); + runtime->stats.put_size_bytes += op.target_spec.size; + runtime->slots[op.slot] = op.target_spec; + } else { + if (op.overwrite) { + ++runtime->stats.overwrite_put_fails_; } else { - thread_stat.data_match_++; + ++runtime->stats.put_fails_; } + } + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); + }); + if (rc != CommonErr::OK) { + std::lock_guard lock(runtime->mutex); + ++runtime->stats.submit_fails_; + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); + } + return; + } - if (folly::Random::rand32(0, 100) < FLAGS_delratio) { - kv_ret = simm_kvstore->Delete(key_vec.at(0)); - thread_stat.del_++; - if (kv_ret != 0) { - thread_stat.del_fails_++; - MLOG_ERROR("Failed to delete key({}) in sync delete IO, kv_ret:{}", key_vec.at(0), kv_ret); - } else { - thread_stat.del_succs_++; - } - } - } // else non-batch mode apis + if (op.op == OpType::Get) { + auto data_holder = std::make_shared(kvstore.Allocate(op.previous_spec.size)); + simm::clnt::DataView get_view(*data_holder); + auto rc = kvstore.AsyncGet(op.key, get_view, [runtime, op, data_holder](int result) { + std::lock_guard lock(runtime->mutex); + ++runtime->stats.get_; + RecordLatency(runtime->stats, op.start_ts); + if (result == static_cast(op.previous_spec.size) && + VerifyBufferFromSeed(std::span(data_holder->AsRef().data(), op.previous_spec.size), + op.previous_spec.seed)) { + ++runtime->stats.get_succs_; + ++runtime->stats.data_match_; + runtime->stats.get_size_bytes += op.previous_spec.size; + } else { + ++runtime->stats.get_fails_; + ++runtime->stats.data_mismatch_; + } + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); + }); + if (rc != CommonErr::OK) { + std::lock_guard lock(runtime->mutex); + ++runtime->stats.submit_fails_; + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); } - kvio_latch.count_down(); - MLOG_INFO("Thead{} for sync io mode exits", tid); - }; - [[maybe_unused]] auto async_io_task = [&]() -> void { - // TODO(ytji): to be added - }; + return; + } - auto test_ts_end = test_binary_start_ts + std::chrono::seconds(FLAGS_time); - auto io_test_start_ts = steady_clock_t::now(); + if (op.op == OpType::Exists) { + auto rc = kvstore.AsyncExists(op.key, [runtime, op](int result) { + std::lock_guard lock(runtime->mutex); + RecordLatency(runtime->stats, op.start_ts); + RecordExpectedExistsResult(runtime->stats, op.previous_spec.exists, result); + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); + }); + if (rc != CommonErr::OK) { + std::lock_guard lock(runtime->mutex); + ++runtime->stats.submit_fails_; + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); + } + return; + } - for (uint32_t i = 0; i < FLAGS_threads; ++i) { - if (FLAGS_iomode == "sync") { - MLOG_INFO("Add sync io task to thread pool, thread_id:{}", i); - thread_pool.add([i, &sync_io_task] { sync_io_task(i); }); - } else if (FLAGS_iomode == "async") { - // TODO + auto rc = kvstore.AsyncDelete(op.key, [runtime, op](int result) { + std::lock_guard lock(runtime->mutex); + ++runtime->stats.del_; + RecordLatency(runtime->stats, op.start_ts); + if (result == CommonErr::OK) { + ++runtime->stats.del_succs_; + if (runtime->slots[op.slot].exists) { + runtime->slots[op.slot].exists = false; + runtime->slots[op.slot].size = 0; + runtime->slots[op.slot].seed = 0; + if (runtime->live_keys > 0) { + --runtime->live_keys; + } + } + } else { + ++runtime->stats.del_fails_; } + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); + }); + if (rc != CommonErr::OK) { + std::lock_guard lock(runtime->mutex); + ++runtime->stats.submit_fails_; + if (runtime->inflight > 0) { + --runtime->inflight; + } + runtime->cv.notify_all(); + } +} + +} // namespace + +int main(int argc, char **argv) { + gflags::SetUsageMessage( + "simm_stable_test: long-running stability and workload tool for SiMM client/service validation"); + gflags::ParseCommandLineFlags(&argc, &argv, true); + folly::Init init(&argc, &argv); + +#ifdef NDEBUG + simm::logging::LogConfig clnt_log_config = simm::logging::LogConfig{"/tmp/simm_clnt.log", "INFO"}; +#else + simm::logging::LogConfig clnt_log_config = simm::logging::LogConfig{"/tmp/simm_clnt.log", "DEBUG"}; +#endif + simm::logging::LoggerManager::Instance().UpdateConfig("simm_client", clnt_log_config); + + check_args(); + std::signal(SIGINT, StopSignalHandler); + std::signal(SIGTERM, StopSignalHandler); + + const char *fixed_kv_mode = FLAGS_fixed_kvsize ? "T" : "F"; + const char *batch_mode = FLAGS_batch_mode ? "T" : "F"; + + MLOG_INFO("++++++++++++++++++++++++++++++++++++++++ Start SIMM_STABLE_TEST ++++++++++++++++++++++++++++++++++++++++"); + std::cout << "Test Args: key_limit=" << FLAGS_keylimit << "B" + << " val_limit=" << FLAGS_vallimit << "B" + << " threads=" << FLAGS_threads << " iomode=" << FLAGS_iomode << " fixed_kv=" << fixed_kv_mode + << " batch_mode=" << batch_mode << " batch_size=" << FLAGS_batch_size << " iodepth=" << FLAGS_iodepth + << " getratio=" << FLAGS_getratio << " putratio=" << FLAGS_putratio << " existsratio=" << FLAGS_existsratio + << " delratio=" << FLAGS_delratio << " keyspace_per_thread=" << FLAGS_keyspace_per_thread + << " report_interval=" << FLAGS_report_interval_inSecs + << " strict_verify_exists=" << (FLAGS_strict_verify_exists ? "T" : "F") << std::endl; + + std::shared_ptr simm_kvstore; + try { + simm_kvstore = std::make_shared(); + } catch (const std::exception &ex) { + std::cerr << "[simm_stable_test] failed to create KVStore: " << ex.what() << std::endl; + return EHOSTUNREACH; + } + + if (!RunStartupSelfCheck(*simm_kvstore)) { + return EHOSTUNREACH; } - while (!kvio_latch.try_wait() && steady_clock_t::now() < test_ts_end) { - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + std::atomic stop_threads{false}; + std::vector> workers; + workers.reserve(FLAGS_threads); + for (uint32_t i = 0; i < FLAGS_threads; ++i) { + workers.push_back(std::make_shared(i, FLAGS_keyspace_per_thread, FLAGS_keylimit)); } - stop_threads.store(true); - kvio_latch.wait(); - MLOG_INFO("All kvio threads jobs finished"); + auto test_start_ts = steady_clock_t::now(); + auto test_end_ts = test_start_ts + std::chrono::seconds(FLAGS_time); + ThreadStats last_snapshot; + auto reporter = std::thread([&]() { + while (!stop_threads.load(std::memory_order_acquire)) { + std::this_thread::sleep_for(std::chrono::seconds(FLAGS_report_interval_inSecs)); + if (stop_threads.load(std::memory_order_acquire)) { + break; + } + auto snapshot = SnapshotStats(workers); + ThreadStats delta = snapshot; + delta.put_ -= last_snapshot.put_; + delta.put_fails_ -= last_snapshot.put_fails_; + delta.put_succs_ -= last_snapshot.put_succs_; + delta.overwrite_put_ -= last_snapshot.overwrite_put_; + delta.overwrite_put_fails_ -= last_snapshot.overwrite_put_fails_; + delta.overwrite_put_succs_ -= last_snapshot.overwrite_put_succs_; + delta.get_ -= last_snapshot.get_; + delta.get_fails_ -= last_snapshot.get_fails_; + delta.get_succs_ -= last_snapshot.get_succs_; + delta.exists_ -= last_snapshot.exists_; + delta.exists_fails_ -= last_snapshot.exists_fails_; + delta.exists_succs_ -= last_snapshot.exists_succs_; + delta.del_ -= last_snapshot.del_; + delta.del_fails_ -= last_snapshot.del_fails_; + delta.del_succs_ -= last_snapshot.del_succs_; + delta.mput_ -= last_snapshot.mput_; + delta.mput_fails_ -= last_snapshot.mput_fails_; + delta.mput_succs_ -= last_snapshot.mput_succs_; + delta.mget_ -= last_snapshot.mget_; + delta.mget_fails_ -= last_snapshot.mget_fails_; + delta.mget_succs_ -= last_snapshot.mget_succs_; + delta.data_match_ -= last_snapshot.data_match_; + delta.data_mismatch_ -= last_snapshot.data_mismatch_; + delta.expected_miss_ -= last_snapshot.expected_miss_; + delta.submit_fails_ -= last_snapshot.submit_fails_; + delta.put_size_bytes -= last_snapshot.put_size_bytes; + delta.get_size_bytes -= last_snapshot.get_size_bytes; + last_snapshot = snapshot; + print_stats(delta, FLAGS_threads, FLAGS_report_interval_inSecs, true); + } + }); - auto io_test_elapsed = std::chrono::duration_cast(steady_clock_t::now() - io_test_start_ts); + std::vector worker_threads; + worker_threads.reserve(FLAGS_threads); + for (uint32_t tid = 0; tid < FLAGS_threads; ++tid) { + auto runtime = workers[tid]; + worker_threads.emplace_back([&, tid, runtime]() { + if (FLAGS_iomode == "sync") { + while (!stop_threads.load(std::memory_order_acquire)) { + if (FLAGS_batch_mode) { + RunSyncBatchOp(tid, *simm_kvstore, *runtime); + } else { + RunSyncSingleOp(tid, *simm_kvstore, *runtime); + } + } + } else { + while (!stop_threads.load(std::memory_order_acquire)) { + SubmitAsyncSingleOp(tid, *simm_kvstore, runtime, stop_threads); + } + std::unique_lock lock(runtime->mutex); + runtime->cv.wait(lock, [&]() { return runtime->inflight == 0; }); + } + MLOG_INFO("Stable test worker {} exits", tid); + }); + } + + while (!g_stop_requested.load(std::memory_order_acquire) && steady_clock_t::now() < test_end_ts) { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + } - MLOG_INFO("Test run time({}s) used up, all worker threads stopped, IO test duration({}s)", - FLAGS_time, - static_cast(io_test_elapsed.count())); + stop_threads.store(true, std::memory_order_release); + for (const auto &worker : workers) { + std::lock_guard lock(worker->mutex); + worker->cv.notify_all(); + } - // +++++++++++++++++++++++++++++++++++ - // Phase : stats aggregation and print - // +++++++++++++++++++++++++++++++++++ - ThreadStats aggregated_stats; - [[maybe_unused]] uint32_t tid = 1; - for (const auto &stat_entry : test_stats_vec) { - aggregated_stats.merge(stat_entry); - // print_stats(stat_entry, tid); - // tid++; + for (auto &worker : worker_threads) { + if (worker.joinable()) { + worker.join(); + } + } + if (reporter.joinable()) { + reporter.join(); } - print_stats(aggregated_stats, 0); + auto elapsed = std::chrono::duration_cast(steady_clock_t::now() - test_start_ts).count(); + auto final_stats = SnapshotStats(workers); + print_stats(final_stats, FLAGS_threads, static_cast(elapsed), false); + if (final_stats.total_ops() == 0) { + std::cerr << "[simm_stable_test] no operations completed" << std::endl; + return EIO; + } + if (final_stats.total_failures() > 0) { + std::cerr << "[simm_stable_test] detected failures during run: " << final_stats.total_failures() << std::endl; + return EIO; + } return 0; } diff --git a/tools/simm_stable_test_upgrade.md b/tools/simm_stable_test_upgrade.md new file mode 100644 index 0000000..35a8211 --- /dev/null +++ b/tools/simm_stable_test_upgrade.md @@ -0,0 +1,141 @@ +# simm_stable_test 升级说明 + +## 背景 + +原始版本的 `tools/simm_stable_test.cc` 更偏功能冒烟工具，存在几个明显短板： + +- `async` 路径基本未完成，`iodepth` 实际不生效 +- 工作负载模型过于单一，难以支撑长期稳定性验证 +- 缺少周期性统计与延迟指标，不利于长时间运行时观测 +- key 基本一次性生成，无法形成更真实的覆盖写、热点和存活数据压力 +- 停止逻辑较粗，长跑时不够稳妥 + +这次改动的目标，是把它提升成更适合持续压测、稳定性回归和故障观测的工具。 + +## 主要改动 + +### 1. 补全 async 模式 + +- 实现真正的异步提交与回调处理 +- `iodepth` 现在实际控制每个 worker 的并发异步请求数 +- 支持异步 `put/get/exists/delete` +- `async + batch` 组合当前显式禁止，避免半成品路径误用 + +### 2. 引入长期稳定 workload 模型 + +- 每个 worker 持有固定的逻辑 keyspace +- key 在 keyspace 内反复复用，而不是持续只写新 key +- 支持更贴近真实服务的混合流量： + - `putratio` + - `getratio` + - `existsratio` + - `delratio` + - `oputratio` + +这样可以覆盖： + +- 覆盖写 +- 老数据读取 +- 数据删除后再次访问 +- 热点 key 反复更新 + +### 3. 改进数据校验方式 + +- 不再为每个 key 保存整份 value 副本 +- 改为记录 `(exists, size, seed)` 元信息 +- value 内容通过 `seed` 按确定性规则生成和校验 + +收益： + +- 降低工具自身内存开销 +- 适合更长时间运行 +- 仍可对 `get` 返回内容做一致性校验 + +### 4. 增加延迟与吞吐统计 + +新增统计项： + +- 总操作数 +- 成功/失败计数 +- submit 失败计数 +- 数据匹配/不匹配计数 +- put/get 字节量 +- 延迟统计： + - avg + - p50 + - p95 + - p99 + - max + +### 5. 增加周期性报告 + +新增 `report_interval_inSecs`： + +- 周期性打印增量统计 +- 测试结束时打印最终汇总 + +这让工具更适合： + +- 长跑观测 +- 问题定位 +- 性能回归对比 + +### 6. 改进退出与收尾 + +- 注册 `SIGINT` / `SIGTERM` +- 支持收到停止信号后优雅退出 +- async worker 在停止时会等待 inflight 请求完成再退出 + +这能减少： + +- 工具自身异常中止 +- 长跑结束时统计不完整 +- 回调尚未收敛时直接退出的问题 + +### 7. 改善可用性 + +- 补充更完整的 flags +- `--help` 现在有正式 usage 信息 +- 启动时打印关键测试参数 + +## 新增/调整的重要参数 + +- `putratio` +- `getratio` +- `existsratio` +- `delratio` +- `oputratio` +- `keyspace_per_thread` +- `report_interval_inSecs` +- `strict_verify_exists` + +## 当前限制 + +- `async + batch_mode` 暂不支持 +- 工具仍以 client 视角做稳定性测试，不替代 server 端 profiling/资源分析 +- 延迟统计当前采用分桶估算 percentile，不是全量样本精确分位数 + +## 构建验证 + +已完成以下验证： + +- `build/release` 下 `simm_stable_test` 构建通过 +- `build/debug` 下 `simm_stable_test` 构建通过 +- `--help` 启动 smoke test 通过 + +测试前统一使用： + +```bash +export SICL_LOG_LEVEL=WARN +``` + +## 建议的后续增强 + +如果后面继续迭代，建议优先考虑： + +- 增加错误码分桶统计 +- 增加更明确的热点分布模型 +- 增加阶段性 summary 文件输出 +- 支持 fault injection 联动场景 +- 增加更细粒度的 async 超时/回调异常观测 + diff --git a/tools/simm_stable_test_utils.h b/tools/simm_stable_test_utils.h new file mode 100644 index 0000000..82ae49f --- /dev/null +++ b/tools/simm_stable_test_utils.h @@ -0,0 +1,45 @@ +#pragma once + +#include +#include +#include +#include + +namespace simm::tools::stable_test { + +inline std::string BuildStableKey(uint32_t tid, size_t slot, size_t key_len) { + std::string key = "stable_t" + std::to_string(tid) + "_s" + std::to_string(slot); + if (key.size() >= key_len) { + key.resize(key_len); + return key; + } + key.append(key_len - key.size(), 'x'); + return key; +} + +inline size_t MinimumUniqueKeyLength(uint32_t max_thread_id, size_t max_slot_id) { + return ("stable_t" + std::to_string(max_thread_id) + "_s" + std::to_string(max_slot_id)).size(); +} + +inline size_t StableKeyLengthForSlot(uint32_t tid, size_t slot, size_t min_len, size_t max_len) { + if (max_len <= min_len) { + return min_len; + } + uint64_t mix = static_cast(tid) * 0x9E3779B97F4A7C15ULL; + mix ^= static_cast(slot) + 0xBF58476D1CE4E5B9ULL + (mix << 6) + (mix >> 2); + const size_t span = max_len - min_len + 1; + return min_len + static_cast(mix % span); +} + +inline std::vector BuildWorkerKeyspace(uint32_t tid, size_t keyspace_size, size_t key_len_limit) { + std::vector keys; + keys.reserve(keyspace_size); + const size_t min_len = MinimumUniqueKeyLength(tid, keyspace_size == 0 ? 0 : keyspace_size - 1); + for (size_t slot = 0; slot < keyspace_size; ++slot) { + const size_t key_len = StableKeyLengthForSlot(tid, slot, min_len, key_len_limit); + keys.push_back(BuildStableKey(tid, slot, key_len)); + } + return keys; +} + +} // namespace simm::tools::stable_test