diff --git a/dependencies.sh b/dependencies.sh index bed94cec..8312c1d4 100755 --- a/dependencies.sh +++ b/dependencies.sh @@ -156,10 +156,10 @@ check_success "Failed to clone yalantinglibs" cd yalantinglibs check_success "Failed to change to yalantinglibs directory" -# Checkout version 0.5.1 -echo "Checking out yalantinglibs version 0.5.1..." -git checkout 0.5.1 -check_success "Failed to checkout yalantinglibs version 0.5.1" +# Checkout version 0.5.5 +echo "Checking out yalantinglibs version 0.5.5..." +git checkout 0.5.5 +check_success "Failed to checkout yalantinglibs version 0.5.5" mkdir -p build check_success "Failed to create build directory" diff --git a/mooncake-common/common.cmake b/mooncake-common/common.cmake index 68ee6254..d2b0a59f 100644 --- a/mooncake-common/common.cmake +++ b/mooncake-common/common.cmake @@ -169,3 +169,4 @@ set(GFLAGS_USE_TARGET_NAMESPACE "true") find_package(yaml-cpp REQUIRED) find_package(gflags REQUIRED) find_package(yalantinglibs CONFIG REQUIRED) +add_compile_definitions(YLT_ENABLE_IBV) diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h index 84d0bd49..fec60394 100644 --- a/mooncake-store/include/master_client.h +++ b/mooncake-store/include/master_client.h @@ -3,7 +3,9 @@ #include #include #include +#include #include +#include #include "client_metric.h" #include "replica.h" @@ -18,7 +20,18 @@ static const std::string kDefaultMasterAddress = "localhost:50051"; */ class MasterClient { public: - MasterClient(MasterClientMetric* metrics = nullptr) : metrics_(metrics) {} + MasterClient(MasterClientMetric* metrics = nullptr) : metrics_(metrics) { + coro_io::client_pool::pool_config + pool_conf{}; + const char* value = std::getenv("MC_RPC_PROTOCOL"); + if (value && std::string_view(value) == "rdma") { + pool_conf.client_config.socket_config = + coro_io::ib_socket_t::config_t{}; + } + client_pools_ = + std::make_shared>( + pool_conf); + } ~MasterClient(); MasterClient(const MasterClient&) = delete; @@ -75,8 +88,8 @@ class MasterClient { * @param object_infos Output parameter for object metadata * @return ErrorCode indicating success/failure */ - [[nodiscard]] - std::vector, ErrorCode>> + [[nodiscard]] std::vector< + tl::expected, ErrorCode>> BatchGetReplicaList(const std::vector& object_keys); /** @@ -235,30 +248,36 @@ class MasterClient { invoke_batch_rpc(size_t input_size, Args&&... args); /** - * @brief Accessor for the coro_rpc_client. Since coro_rpc_client cannot - * reconnect to a different address, a new coro_rpc_client is created if - * the address is different from the current one. + * @brief Accessor for the coro_rpc_client pool. Since coro_rpc_client pool + * cannot reconnect to a different address, a new coro_rpc_client pool is + * created if the address is different from the current one. */ class RpcClientAccessor { public: - void SetClient(std::shared_ptr client) { + void SetClientPool( + std::shared_ptr> + client_pool) { std::lock_guard lock(client_mutex_); - client_ = client; + client_pool_ = client_pool; } - std::shared_ptr GetClient() { + std::shared_ptr> + GetClientPool() { std::shared_lock lock(client_mutex_); - return client_; + return client_pool_; } private: mutable std::shared_mutex client_mutex_; - std::shared_ptr client_; + std::shared_ptr> + client_pool_; }; RpcClientAccessor client_accessor_; // Metrics for tracking RPC operations MasterClientMetric* metrics_; + std::shared_ptr> + client_pools_; // Mutex to insure the Connect function is atomic. mutable Mutex connect_mutex_; diff --git a/mooncake-store/src/ha_helper.cpp b/mooncake-store/src/ha_helper.cpp index c1f4dedd..796838a8 100644 --- a/mooncake-store/src/ha_helper.cpp +++ b/mooncake-store/src/ha_helper.cpp @@ -101,6 +101,11 @@ int MasterServiceSupervisor::Start() { coro_rpc::coro_rpc_server server( config_.rpc_thread_num, config_.rpc_port, config_.rpc_address, config_.rpc_conn_timeout, config_.rpc_enable_tcp_no_delay); + const char* value = std::getenv("MC_RPC_PROTOCOL"); + if (value && std::string_view(value) == "rdma") { + server.init_ibv(); + } + LOG(INFO) << "Init leader election helper..."; MasterViewHelper mv_helper; if (mv_helper.ConnectToEtcd(config_.etcd_endpoints) != ErrorCode::OK) { diff --git a/mooncake-store/src/master.cpp b/mooncake-store/src/master.cpp index 26d991c9..a3f6ca6f 100644 --- a/mooncake-store/src/master.cpp +++ b/mooncake-store/src/master.cpp @@ -52,7 +52,6 @@ DEFINE_int32(rpc_conn_timeout_seconds, 0, "Connection timeout in seconds (0 = no timeout)"); DEFINE_bool(rpc_enable_tcp_no_delay, true, "Enable TCP_NODELAY for RPC connections"); - DEFINE_validator(eviction_ratio, [](const char* flagname, double value) { if (value < 0.0 || value > 1.0) { LOG(FATAL) << "Eviction ratio must be between 0.0 and 1.0"; @@ -355,6 +354,11 @@ int main(int argc, char* argv[]) { return 1; } + const char* value = std::getenv("MC_RPC_PROTOCOL"); + std::string protocol = "tcp"; + if (value && std::string_view(value) == "rdma") { + protocol = "rdma"; + } LOG(INFO) << "Master service started on port " << master_config.rpc_port << ", max_threads=" << master_config.rpc_thread_num << ", enable_metric_reporting=" @@ -378,6 +382,7 @@ int main(int argc, char* argv[]) { << master_config.rpc_conn_timeout_seconds << ", rpc_enable_tcp_no_delay=" << master_config.rpc_enable_tcp_no_delay + << ", rpc protocol=" << protocol << ", cluster_id=" << master_config.cluster_id << ", root_fs_dir=" << master_config.root_fs_dir << ", memory_allocator=" << master_config.memory_allocator @@ -416,6 +421,10 @@ int main(int argc, char* argv[]) { master_config.rpc_address, std::chrono::seconds(master_config.rpc_conn_timeout_seconds), master_config.rpc_enable_tcp_no_delay); + const char* value = std::getenv("MC_RPC_PROTOCOL"); + if (value && std::string_view(value) == "rdma") { + server.init_ibv(); + } mooncake::WrappedMasterService wrapped_master_service( mooncake::WrappedMasterServiceConfig(master_config, version)); diff --git a/mooncake-store/src/master_client.cpp b/mooncake-store/src/master_client.cpp index a087941e..5c058f73 100644 --- a/mooncake-store/src/master_client.cpp +++ b/mooncake-store/src/master_client.cpp @@ -118,11 +118,7 @@ struct RpcNameTraits<&WrappedMasterService::GetFsdir> { template tl::expected MasterClient::invoke_rpc(Args&&... args) { - auto client = client_accessor_.GetClient(); - if (!client) { - LOG(ERROR) << "Client not available"; - return tl::make_unexpected(ErrorCode::RPC_FAIL); - } + auto pool = client_accessor_.GetClientPool(); // Increment RPC counter if (metrics_) { @@ -130,12 +126,19 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { } auto start_time = std::chrono::steady_clock::now(); - auto request_result = - client->send_request(std::forward(args)...); - return async_simple::coro::syncAwait( [&]() -> async_simple::coro::Lazy> { - auto result = co_await co_await request_result; + auto ret = co_await pool->send_request( + [&](coro_io::client_reuse_hint, + coro_rpc::coro_rpc_client& client) { + return client.send_request( + std::forward(args)...); + }); + if (!ret.has_value()) { + LOG(ERROR) << "Client not available"; + co_return tl::make_unexpected(ErrorCode::RPC_FAIL); + } + auto result = co_await std::move(ret.value()); if (!result) { LOG(ERROR) << "RPC call failed: " << result.error().msg; co_return tl::make_unexpected(ErrorCode::RPC_FAIL); @@ -155,12 +158,7 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { template std::vector> MasterClient::invoke_batch_rpc( size_t input_size, Args&&... args) { - auto client = client_accessor_.GetClient(); - if (!client) { - LOG(ERROR) << "Client not available"; - return std::vector>( - input_size, tl::make_unexpected(ErrorCode::RPC_FAIL)); - } + auto pool = client_accessor_.GetClientPool(); // Increment RPC counter if (metrics_) { @@ -168,12 +166,21 @@ std::vector> MasterClient::invoke_batch_rpc( } auto start_time = std::chrono::steady_clock::now(); - auto request_result = - client->send_request(std::forward(args)...); return async_simple::coro::syncAwait( [&]() -> async_simple::coro::Lazy< std::vector>> { - auto result = co_await co_await request_result; + auto ret = co_await pool->send_request( + [&](coro_io::client_reuse_hint, + coro_rpc::coro_rpc_client& client) { + return client.send_request( + std::forward(args)...); + }); + if (!ret.has_value()) { + LOG(ERROR) << "Client not available"; + co_return std::vector>( + input_size, tl::make_unexpected(ErrorCode::RPC_FAIL)); + } + auto result = co_await std::move(ret.value()); if (!result) { LOG(ERROR) << "Batch RPC call failed: " << result.error().msg; std::vector> error_results; @@ -207,35 +214,17 @@ ErrorCode MasterClient::Connect(const std::string& master_addr) { LOG(INFO) << "Connecting to master at " << master_addr << " from " << name; MutexLocker lock(&connect_mutex_); - if (client_addr_param_ == master_addr) { - auto client = client_accessor_.GetClient(); - auto result = - async_simple::coro::syncAwait(client->connect(master_addr)); - if (result.val() != 0) { - LOG(ERROR) << "Failed to connect to master: " << result.message(); - timer.LogResponse("error_code=", ErrorCode::RPC_FAIL); - return ErrorCode::RPC_FAIL; - } - timer.LogResponse("error_code=", ErrorCode::OK); - return ErrorCode::OK; - } else { - // Once connected to address A, the coro_rpc_client does not support - // connect to a new address B. So we need to create a new - // coro_rpc_client if the address is different from the current one. - auto client = std::make_shared(); - auto result = - async_simple::coro::syncAwait(client->connect(master_addr)); - if (result.val() != 0) { - LOG(ERROR) << "Failed to connect to master: " << result.message(); - timer.LogResponse("error_code=", ErrorCode::RPC_FAIL); - return ErrorCode::RPC_FAIL; - } - // Set the client to the accessor and update the address parameter - client_accessor_.SetClient(client); + if (client_addr_param_ != master_addr) { + lock.unlock(); + // add a new client pool to client pools. + auto client_pool = client_pools_->at(master_addr); + lock.lock(); client_addr_param_ = master_addr; - timer.LogResponse("error_code=", ErrorCode::OK); - return ErrorCode::OK; + lock.unlock(); + client_accessor_.SetClientPool(client_pool); } + timer.LogResponse("error_code=", ErrorCode::OK); + return ErrorCode::OK; } tl::expected MasterClient::ExistKey( diff --git a/mooncake-wheel/tests/test_meta_server.py b/mooncake-wheel/tests/test_meta_server.py new file mode 100644 index 00000000..2a2f27b8 --- /dev/null +++ b/mooncake-wheel/tests/test_meta_server.py @@ -0,0 +1,54 @@ +import os, sys, random +import time +import threading + +from mooncake.store import MooncakeDistributedStore + +# how to test: start mooncake_master and http_metadata_server. +# Default is tcp, if you want to use rdma, should set MC_RPC_PROTOCOL=rdma and DEVICE_NAME=rdma_xxx at first. +# ./mooncake_master +# python mooncake-wheel/mooncake/http_metadata_server.py --port 8080 +# python mooncake-wheel/tests/test_meta_server.py 127.0.0.1 8 + +import mooncake +print(mooncake.__file__) + +def test_worker(store, num_req): + for i in range(num_req): + store.is_exist(str(random.randint(0, 1000000))) + +master_host = "master.mooncake.dc" if len(sys.argv) < 2 else sys.argv[1] +num_thread = 1 if len(sys.argv) < 3 else int(sys.argv[2]) +# Initialize the store +store = MooncakeDistributedStore() +# Use TCP protocol by default for testing, also support rdma +protocol = os.getenv("MC_RPC_PROTOCOL", "tcp") +device_name = os.getenv("DEVICE_NAME", "eth0") +local_hostname = os.getenv("LOCAL_HOSTNAME", "127.0.0.1") +metadata_server = os.getenv("METADATA_ADDR", f"http://{master_host}:8080/metadata") +global_segment_size = 0 +local_buffer_size = 512 * 1024 * 1024 +master_server_address = os.getenv("MASTER_SERVER", f"{master_host}:50051") +value_length = 1 * 1024 * 1024 +retcode = store.setup(local_hostname, + metadata_server, + global_segment_size, + local_buffer_size, + protocol, + device_name, + master_server_address) +if retcode: + exit(1) +time.sleep(1) # Give some time for initialization +start_time = time.perf_counter() +threads = [] +for _ in range(num_thread): + thread = threading.Thread(target=test_worker, args=(store, 15000)) + thread.start() + threads.append(thread) +for thread in threads: + thread.join() +end_time = time.perf_counter() +total_time = end_time - start_time if end_time > start_time else 0 +ops_per_second = 15000 * num_thread / total_time if total_time > 0 else 0 +print(f"total time: {total_time} s, QPS: {ops_per_second:.2f}") \ No newline at end of file diff --git a/scripts/ascend/dependencies_ascend.sh b/scripts/ascend/dependencies_ascend.sh index 6a111347..be62b52d 100644 --- a/scripts/ascend/dependencies_ascend.sh +++ b/scripts/ascend/dependencies_ascend.sh @@ -114,7 +114,7 @@ pwd # Install yalantinglibs clone_repo_if_not_exists "yalantinglibs" "https://github.com/alibaba/yalantinglibs.git" cd yalantinglibs || exit -git checkout 0.5.1 +git checkout 0.5.5 rm -rf build mkdir -p build && cd build cmake .. -DBUILD_EXAMPLES=OFF -DBUILD_BENCHMARK=OFF -DBUILD_UNIT_TESTS=OFF diff --git a/scripts/ascend/dependencies_ascend_installation.sh b/scripts/ascend/dependencies_ascend_installation.sh index 5174fb43..79719aa8 100644 --- a/scripts/ascend/dependencies_ascend_installation.sh +++ b/scripts/ascend/dependencies_ascend_installation.sh @@ -107,7 +107,7 @@ export CPLUS_INCLUDE_PATH=$(echo $CPLUS_INCLUDE_PATH | tr ':' '\n' | grep -v "/u # Install yalantinglibs clone_repo_if_not_exists "yalantinglibs" "https://github.com/alibaba/yalantinglibs.git" cd yalantinglibs || exit -git checkout 0.5.1 +git checkout 0.5.5 rm -rf build mkdir -p build && cd build cmake .. -DBUILD_EXAMPLES=OFF -DBUILD_BENCHMARK=OFF -DBUILD_UNIT_TESTS=OFF