[GH-ISSUE #297] How to config rpc parameters for benchmark? #182

New issue

Open

opened 2026-10-26 09:46:43 +00:00 by bsakaguchi · 0 comments

bsakaguchi commented

2026-10-26 09:46:43 +00:00

Owner

Originally created by @qicosmos on GitHub (Jun 16, 2025).
Original GitHub issue: https://github.com/deepseek-ai/3FS/issues/297

I wrote a simple test case to benchmark the 3FS RPC (rdma), but found that it couldn't fully utilize the available bandwidth.
The test was conducted in a 200Gb/s environment, but the throughput is not expected. I suspect that some misconfiguration is causing this result.
Could someone please share how to properly configure the RPC server and client for benchmarking purposes？

benchmark code in test_net.cpp:

#include <chrono>
#include <folly/Random.h>
#include <folly/Unit.h>
#include <folly/executors/CPUThreadPoolExecutor.h>
#include <folly/executors/IOThreadPoolExecutor.h>
#include <folly/experimental/coro/BlockingWait.h>
#include <folly/experimental/coro/Collect.h>
#include <folly/experimental/coro/GtestHelpers.h>
#include <folly/experimental/coro/Invoke.h>
#include <folly/init/Init.h>
#include <folly/logging/xlog.h>
#include <gflags/gflags.h>
#include <gtest/gtest.h>
#include <memory>
#include <thread>
#include <tuple>
#include <utility>
#include <vector>

#include "common/net/Client.h"
#include "common/net/Listener.h"
#include "common/net/Server.h"
#include "common/net/ib/IBDevice.h"
#include "common/net/sync/Client.h"
#include "common/serde/ClientContext.h"
#include "common/utils/Address.h"
#include "common/utils/Coroutine.h"
#include "tests/GtestHelpers.h"
#include "tests/common/net/Echo.h"
#include "tests/common/net/ib/SetupIB.h"
#include <ylt/metric/summary.hpp>

DEFINE_uint32(req_len, 13, "request data len");
DEFINE_uint32(par, 10, "parallel");
DEFINE_uint32(port, 0, "port");
DEFINE_string(host, "10.0.0.42:9004", "host");
DEFINE_uint32(tcp, 1, "enable tcp, otherwise rdma");
DEFINE_uint32(dur, 20, "test duration");

using namespace hf3fs;
using namespace hf3fs::net;
using namespace hf3fs::net::test;

std::atomic<uint64_t> g_count = 0;
ylt::metric::summary_t g_latency{"Latency(us) of rpc call", "help",
                                 std::vector{0.5, 0.9, 0.95, 0.99},
                                 std::chrono::seconds{60}};

class EchoServiceImpl : public serde::ServiceWrapper<EchoServiceImpl, Echo> {
 public:
  CoTryTask<EchoRsp> echo(serde::CallContext &ctx, const EchoReq &req) {
    EchoRsp rsp;
    rsp.val = "hello for test";
    co_return rsp;
  }

  CoTryTask<HelloRsp> hello(serde::CallContext &ctx, const HelloReq &req) {
    HelloRsp rsp;
    rsp.val = "Hello, " + req.val;
    rsp.idx = ++idx_;
    co_return rsp;
  }

  CoTryTask<HelloRsp> fail(serde::CallContext &ctx, const HelloReq &) {
    fmt::print("Request from {}\n", ctx.transport()->describe());
    co_return makeError(RPCCode::kInvalidMessageType, "failed");
  }

 private:
  uint32_t idx_ = 0;
};

int main(int argc, char **argv) {
  folly::Init init(&argc, &argv);
  if(FLAGS_tcp == 0) {
  static IBConfig config;
  config.set_allow_unknown_zone(true);
  config.set_device_filter({"mlx5_bond_0"});
  auto ib = IBManager::start(config);
  std::cout << "is empty " << IBDevice::all().empty() << "\n";
  XLOGF_IF(FATAL, ib.hasError(), "IBManager start failed, result {}", ib.error());
  assert(IBDevice::all().empty());
  }

  if (FLAGS_port > 0) {
    hf3fs::net::Server::Config serverConfig;
    if(FLAGS_tcp == 1) {
      serverConfig.groups(0).set_network_type(Address::TCP);
    }else {
      serverConfig.groups(0).set_network_type(Address::RDMA);
    }
    serverConfig.groups(0).listener().set_listen_port(FLAGS_port);
    serverConfig.groups(0).io_worker().transport_pool().set_max_connections(2000);
    serverConfig.groups(0).io_worker().set_num_event_loop(std::thread::hardware_concurrency());
    serverConfig.groups(0).io_worker().set_read_write_tcp_in_event_thread(true);
    serverConfig.groups(0).io_worker().ibsocket().set_buf_size(2 * 1024 * 1024);
    serverConfig.thread_pool().set_num_io_threads(std::thread::hardware_concurrency());
    serverConfig.thread_pool().set_num_proc_threads(std::thread::hardware_concurrency());
    // serverConfig.thread_pool().set_num_bg_threads(std::thread::hardware_concurrency());
    std::cout << "start server, port " << FLAGS_port << ", enable tcp " << FLAGS_tcp  << "\n";
    hf3fs::net::Server server{serverConfig};
    server.addSerdeService(std::make_unique<EchoServiceImpl>());
    server.setup();
    server.start();

    std::string wait_quit;
    std::cin >> wait_quit;
  }
  std::cout << "start client, remote addr " << FLAGS_host << ", concurrency " << FLAGS_par << ", req_len "
            << FLAGS_req_len << "\n";
  hf3fs::net::Client::Config clientConfig{};
  clientConfig.io_worker().transport_pool().set_max_connections(std::thread::hardware_concurrency());
  clientConfig.io_worker().set_read_write_tcp_in_event_thread(true);
  clientConfig.io_worker().set_num_event_loop(std::thread::hardware_concurrency());
  clientConfig.thread_pool().set_num_connect_threads(std::thread::hardware_concurrency());
  clientConfig.thread_pool().set_num_io_threads(std::thread::hardware_concurrency());
  clientConfig.thread_pool().set_num_proc_threads(std::thread::hardware_concurrency());
  clientConfig.set_default_timeout(5_s);
  clientConfig.io_worker().ibsocket().set_buf_size(2*1024*1024);
  hf3fs::net::Client client{clientConfig};
  Address addr = Address::fromString(FLAGS_host, Address::RDMA);
  if(FLAGS_tcp == 1) {
      addr = Address::fromString(FLAGS_host, Address::TCP);
  }

  auto ctx = client.serdeCtx(addr);
  client.start();

  EchoReq req;
  req.val = std::string(FLAGS_req_len, 'A');
  auto task = [&]() -> CoTask<void> {
    Echo<> client;
    while (true) {
      auto start = std::chrono::steady_clock::now();
      auto result = co_await client.echo(ctx, req);
      if (result.hasValue()) {
        g_count += FLAGS_req_len;
        auto now = std::chrono::steady_clock::now();
        g_latency.observe(std::chrono::duration_cast<std::chrono::microseconds>(now - start).count());
      }
    }
  };

  std::cout << std::fixed << std::setprecision(2);
  g_count = 0;
  std::thread thd([] {
    size_t total = 0;
    for (uint32_t i = 0; i < FLAGS_dur; i++) {
      auto start = std::chrono::system_clock::now();
      std::this_thread::sleep_for(std::chrono::seconds{1});
      auto c = g_count.exchange(0);
      total += c;
      auto end = std::chrono::system_clock::now();
      auto dur = (end - start) / std::chrono::milliseconds(1);
      double val = (8.0 * c * 1000) / (1000'000'000ll * dur);
      std::cout << "qps " << c / FLAGS_req_len << ", Throughput:" << val << " Gb/s\n";
    }
    std::cout << "# Benchmark result \n";
    double val = (8.0 * total) / (1000'000'000ll * FLAGS_dur);
    std::cout << "avg qps " << total / (FLAGS_req_len * FLAGS_dur) << " and throughput:" << val << " Gb/s in duration "
              << FLAGS_dur << "\n";
    std::string str_rate;
    g_latency.serialize(str_rate);
    std::cout << str_rate << "\n";
  });

  folly::CPUThreadPoolExecutor exec(std::thread::hardware_concurrency());
  for (size_t i = 0; i < FLAGS_par; i++) {
    task().scheduleOn(folly::Executor::getKeepAliveToken(exec)).start();
  }
  exec.join();
  return 0;
}

rdma benchmark command line:

./test_net -port 9004 -tcp 0

./test_net -host server_ip:9004 -req_len 8388608 -par 16 -dur 15

some test results:

When sending 8M+ data to test, the throughout become lower and lower, and the cpu is very high.

I'm eager to hear your suggestions.

Originally created by @qicosmos on GitHub (Jun 16, 2025). Original GitHub issue: https://github.com/deepseek-ai/3FS/issues/297 I wrote a simple test case to benchmark the 3FS RPC (rdma), but found that it couldn't fully utilize the available bandwidth. The test was conducted in a 200Gb/s environment, but the throughput is not expected. I suspect that some misconfiguration is causing this result. Could someone please share how to properly configure the RPC server and client for benchmarking purposes？ benchmark code in test_net.cpp: ```cpp #include <chrono> #include <folly/Random.h> #include <folly/Unit.h> #include <folly/executors/CPUThreadPoolExecutor.h> #include <folly/executors/IOThreadPoolExecutor.h> #include <folly/experimental/coro/BlockingWait.h> #include <folly/experimental/coro/Collect.h> #include <folly/experimental/coro/GtestHelpers.h> #include <folly/experimental/coro/Invoke.h> #include <folly/init/Init.h> #include <folly/logging/xlog.h> #include <gflags/gflags.h> #include <gtest/gtest.h> #include <memory> #include <thread> #include <tuple> #include <utility> #include <vector> #include "common/net/Client.h" #include "common/net/Listener.h" #include "common/net/Server.h" #include "common/net/ib/IBDevice.h" #include "common/net/sync/Client.h" #include "common/serde/ClientContext.h" #include "common/utils/Address.h" #include "common/utils/Coroutine.h" #include "tests/GtestHelpers.h" #include "tests/common/net/Echo.h" #include "tests/common/net/ib/SetupIB.h" #include <ylt/metric/summary.hpp> DEFINE_uint32(req_len, 13, "request data len"); DEFINE_uint32(par, 10, "parallel"); DEFINE_uint32(port, 0, "port"); DEFINE_string(host, "10.0.0.42:9004", "host"); DEFINE_uint32(tcp, 1, "enable tcp, otherwise rdma"); DEFINE_uint32(dur, 20, "test duration"); using namespace hf3fs; using namespace hf3fs::net; using namespace hf3fs::net::test; std::atomic<uint64_t> g_count = 0; ylt::metric::summary_t g_latency{"Latency(us) of rpc call", "help", std::vector{0.5, 0.9, 0.95, 0.99}, std::chrono::seconds{60}}; class EchoServiceImpl : public serde::ServiceWrapper<EchoServiceImpl, Echo> { public: CoTryTask<EchoRsp> echo(serde::CallContext &ctx, const EchoReq &req) { EchoRsp rsp; rsp.val = "hello for test"; co_return rsp; } CoTryTask<HelloRsp> hello(serde::CallContext &ctx, const HelloReq &req) { HelloRsp rsp; rsp.val = "Hello, " + req.val; rsp.idx = ++idx_; co_return rsp; } CoTryTask<HelloRsp> fail(serde::CallContext &ctx, const HelloReq &) { fmt::print("Request from {}\n", ctx.transport()->describe()); co_return makeError(RPCCode::kInvalidMessageType, "failed"); } private: uint32_t idx_ = 0; }; int main(int argc, char **argv) { folly::Init init(&argc, &argv); if(FLAGS_tcp == 0) { static IBConfig config; config.set_allow_unknown_zone(true); config.set_device_filter({"mlx5_bond_0"}); auto ib = IBManager::start(config); std::cout << "is empty " << IBDevice::all().empty() << "\n"; XLOGF_IF(FATAL, ib.hasError(), "IBManager start failed, result {}", ib.error()); assert(IBDevice::all().empty()); } if (FLAGS_port > 0) { hf3fs::net::Server::Config serverConfig; if(FLAGS_tcp == 1) { serverConfig.groups(0).set_network_type(Address::TCP); }else { serverConfig.groups(0).set_network_type(Address::RDMA); } serverConfig.groups(0).listener().set_listen_port(FLAGS_port); serverConfig.groups(0).io_worker().transport_pool().set_max_connections(2000); serverConfig.groups(0).io_worker().set_num_event_loop(std::thread::hardware_concurrency()); serverConfig.groups(0).io_worker().set_read_write_tcp_in_event_thread(true); serverConfig.groups(0).io_worker().ibsocket().set_buf_size(2 * 1024 * 1024); serverConfig.thread_pool().set_num_io_threads(std::thread::hardware_concurrency()); serverConfig.thread_pool().set_num_proc_threads(std::thread::hardware_concurrency()); // serverConfig.thread_pool().set_num_bg_threads(std::thread::hardware_concurrency()); std::cout << "start server, port " << FLAGS_port << ", enable tcp " << FLAGS_tcp << "\n"; hf3fs::net::Server server{serverConfig}; server.addSerdeService(std::make_unique<EchoServiceImpl>()); server.setup(); server.start(); std::string wait_quit; std::cin >> wait_quit; } std::cout << "start client, remote addr " << FLAGS_host << ", concurrency " << FLAGS_par << ", req_len " << FLAGS_req_len << "\n"; hf3fs::net::Client::Config clientConfig{}; clientConfig.io_worker().transport_pool().set_max_connections(std::thread::hardware_concurrency()); clientConfig.io_worker().set_read_write_tcp_in_event_thread(true); clientConfig.io_worker().set_num_event_loop(std::thread::hardware_concurrency()); clientConfig.thread_pool().set_num_connect_threads(std::thread::hardware_concurrency()); clientConfig.thread_pool().set_num_io_threads(std::thread::hardware_concurrency()); clientConfig.thread_pool().set_num_proc_threads(std::thread::hardware_concurrency()); clientConfig.set_default_timeout(5_s); clientConfig.io_worker().ibsocket().set_buf_size(2*1024*1024); hf3fs::net::Client client{clientConfig}; Address addr = Address::fromString(FLAGS_host, Address::RDMA); if(FLAGS_tcp == 1) { addr = Address::fromString(FLAGS_host, Address::TCP); } auto ctx = client.serdeCtx(addr); client.start(); EchoReq req; req.val = std::string(FLAGS_req_len, 'A'); auto task = [&]() -> CoTask<void> { Echo<> client; while (true) { auto start = std::chrono::steady_clock::now(); auto result = co_await client.echo(ctx, req); if (result.hasValue()) { g_count += FLAGS_req_len; auto now = std::chrono::steady_clock::now(); g_latency.observe(std::chrono::duration_cast<std::chrono::microseconds>(now - start).count()); } } }; std::cout << std::fixed << std::setprecision(2); g_count = 0; std::thread thd([] { size_t total = 0; for (uint32_t i = 0; i < FLAGS_dur; i++) { auto start = std::chrono::system_clock::now(); std::this_thread::sleep_for(std::chrono::seconds{1}); auto c = g_count.exchange(0); total += c; auto end = std::chrono::system_clock::now(); auto dur = (end - start) / std::chrono::milliseconds(1); double val = (8.0 * c * 1000) / (1000'000'000ll * dur); std::cout << "qps " << c / FLAGS_req_len << ", Throughput:" << val << " Gb/s\n"; } std::cout << "# Benchmark result \n"; double val = (8.0 * total) / (1000'000'000ll * FLAGS_dur); std::cout << "avg qps " << total / (FLAGS_req_len * FLAGS_dur) << " and throughput:" << val << " Gb/s in duration " << FLAGS_dur << "\n"; std::string str_rate; g_latency.serialize(str_rate); std::cout << str_rate << "\n"; }); folly::CPUThreadPoolExecutor exec(std::thread::hardware_concurrency()); for (size_t i = 0; i < FLAGS_par; i++) { task().scheduleOn(folly::Executor::getKeepAliveToken(exec)).start(); } exec.join(); return 0; } ``` rdma benchmark command line: ``` ./test_net -port 9004 -tcp 0 ``` ``` ./test_net -host server_ip:9004 -req_len 8388608 -par 16 -dur 15 ``` some test results: ![img](https://github.com/user-attachments/assets/2c0e2696-5bbb-4868-b9f3-3c25eb7ca0df) When sending 8M+ data to test, the throughout become lower and lower, and the cpu is very high. I'm eager to hear your suggestions.

No labels

pull-request

No milestone

No project

No assignees

1 participant

Notifications

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference

deepseek-ai/3FS#182

No description provided.

Rows
Columns