chenyu520
/
Mooncake
forked from mooncake-track/Mooncake
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[coro_rpc] use client pool and enable rdma (#789)

This commit is contained in:
qicosmos 2025-09-02 00:40:21 +08:00 committed by GitHub
parent 506231b615
commit 185c5d229b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 140 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dependencies.sh
View File

@ -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"

1
mooncake-common/common.cmake
View File

@ -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)

41
mooncake-store/include/master_client.h
View File

@ -3,7 +3,9 @@
#include <memory>
#include <string>
#include <vector>
#include <cstdlib>
#include <ylt/coro_rpc/coro_rpc_client.hpp>
#include <ylt/coro_io/client_pool.hpp>
#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<coro_rpc::coro_rpc_client>::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<coro_io::client_pools<coro_rpc::coro_rpc_client>>(
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<tl::expected<std::vector<Replica::Descriptor>, ErrorCode>>
[[nodiscard]] std::vector<
tl::expected<std::vector<Replica::Descriptor>, ErrorCode>>
BatchGetReplicaList(const std::vector<std::string>& 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<coro_rpc::coro_rpc_client> client) {
void SetClientPool(
std::shared_ptr<coro_io::client_pool<coro_rpc::coro_rpc_client>>
client_pool) {
std::lock_guard<std::shared_mutex> lock(client_mutex_);
client_ = client;
client_pool_ = client_pool;
}
std::shared_ptr<coro_rpc::coro_rpc_client> GetClient() {
std::shared_ptr<coro_io::client_pool<coro_rpc::coro_rpc_client>>
GetClientPool() {
std::shared_lock<std::shared_mutex> lock(client_mutex_);
return client_;
return client_pool_;
}
private:
mutable std::shared_mutex client_mutex_;
std::shared_ptr<coro_rpc::coro_rpc_client> client_;
std::shared_ptr<coro_io::client_pool<coro_rpc::coro_rpc_client>>
client_pool_;
};
RpcClientAccessor client_accessor_;
// Metrics for tracking RPC operations
MasterClientMetric* metrics_;
std::shared_ptr<coro_io::client_pools<coro_rpc::coro_rpc_client>>
client_pools_;
// Mutex to insure the Connect function is atomic.
mutable Mutex connect_mutex_;

5
mooncake-store/src/ha_helper.cpp
View File

@ -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) {

11
mooncake-store/src/master.cpp
View File

@ -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));

79
mooncake-store/src/master_client.cpp
View File

@ -118,11 +118,7 @@ struct RpcNameTraits<&WrappedMasterService::GetFsdir> {
template <auto ServiceMethod, typename ReturnType, typename... Args>
tl::expected<ReturnType, ErrorCode> 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<ReturnType, ErrorCode> MasterClient::invoke_rpc(Args&&... args) {
}
auto start_time = std::chrono::steady_clock::now();
auto request_result =
client->send_request<ServiceMethod>(std::forward<Args>(args)...);
return async_simple::coro::syncAwait(
[&]() -> async_simple::coro::Lazy<tl::expected<ReturnType, ErrorCode>> {
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<ServiceMethod>(
std::forward<Args>(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<ReturnType, ErrorCode> MasterClient::invoke_rpc(Args&&... args) {
template <auto ServiceMethod, typename ResultType, typename... Args>
std::vector<tl::expected<ResultType, ErrorCode>> 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<tl::expected<ResultType, ErrorCode>>(
input_size, tl::make_unexpected(ErrorCode::RPC_FAIL));
}
auto pool = client_accessor_.GetClientPool();
// Increment RPC counter
if (metrics_) {
@ -168,12 +166,21 @@ std::vector<tl::expected<ResultType, ErrorCode>> MasterClient::invoke_batch_rpc(
}
auto start_time = std::chrono::steady_clock::now();
auto request_result =
client->send_request<ServiceMethod>(std::forward<Args>(args)...);
return async_simple::coro::syncAwait(
[&]() -> async_simple::coro::Lazy<
std::vector<tl::expected<ResultType, ErrorCode>>> {
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<ServiceMethod>(
std::forward<Args>(args)...);
});
if (!ret.has_value()) {
LOG(ERROR) << "Client not available";
co_return std::vector<tl::expected<ResultType, ErrorCode>>(
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<tl::expected<ResultType, ErrorCode>> 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<coro_rpc::coro_rpc_client>();
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<bool, ErrorCode> MasterClient::ExistKey(

54
mooncake-wheel/tests/test_meta_server.py Normal file
View File

@ -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}")

2
scripts/ascend/dependencies_ascend.sh
View File

@ -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

2
scripts/ascend/dependencies_ascend_installation.sh
View File

@ -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