From f8de201625e70f195d4f8ef0a0ff19a77c1d042e Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Mon, 6 Apr 2026 17:52:02 +0000 Subject: [PATCH] Close TCP connections for RDMA clients --- src/client/messenger.cpp | 42 ++++++++++++++++++++++++++++--------- src/client/messenger.h | 8 ++++--- src/client/msgr_rdma.cpp | 21 +++++++++++++++++-- src/client/msgr_receive.cpp | 7 ++++++- src/client/msgr_send.cpp | 18 ++++++++++------ src/osd/osd_secondary.cpp | 7 +++++++ 6 files changed, 81 insertions(+), 22 deletions(-) diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index 2fdf801a..5ea47bab 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -191,9 +191,12 @@ void osd_messenger_t::init() { auto cl = cl_it->second; cl_it++; - if (!cl->osd_num && !cl->in_osd_num || cl->peer_state != PEER_CONNECTED && cl->peer_state != PEER_RDMA) + if (!cl->osd_num && !cl->in_osd_num || + cl->peer_state == PEER_CONNECTING || + cl->peer_state == PEER_RDMA_CONNECTING_IN || + cl->peer_state == PEER_STOPPED) { - // Do not run keepalive on regular clients + // Do not run keepalive on regular and unconnected clients continue; } if (cl->ping_time_remaining > 0) @@ -222,7 +225,7 @@ void osd_messenger_t::init() .opcode = OSD_OP_PING, }, }; - op->callback = [this, cl](osd_op_t *op) + op->callback = [this](osd_op_t *op) { auto cl_it = clients.find(op->client_id); if (cl_it == clients.end()) @@ -231,14 +234,15 @@ void osd_messenger_t::init() delete op; return; } - uint64_t fail_client_id = (op->reply.hdr.retval != 0 ? op->client_id : 0); - auto fail_osd_num = cl->in_osd_num ? cl->in_osd_num : cl->osd_num; + auto cl = cl_it->second; + bool failed = (op->reply.hdr.retval != 0); cl->ping_time_remaining = 0; delete op; - if (fail_client_id) + if (failed) { - fprintf(stderr, "Ping failed for OSD %ju (client %ju), disconnecting peer\n", fail_osd_num, fail_client_id); - stop_client(fail_client_id); + auto fail_osd_num = cl->in_osd_num ? cl->in_osd_num : cl->osd_num; + fprintf(stderr, "Ping failed for OSD %ju (client %ju), disconnecting peer\n", fail_osd_num, cl->client_id); + stop_client(cl->client_id); } }; cl->ping_time_remaining = osd_ping_timeout; @@ -608,6 +612,17 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events) if (epoll_events & EPOLLRDHUP) { // Stop client + if (cl->rdma_close_tcp && (cl->peer_state == PEER_RDMA || + cl->peer_state == PEER_RDMA_CONNECTING_IN || + cl->peer_state == PEER_RDMA_CONNECTING_OUT)) + { + // It's allowed to stop the TCP socket during RDMA transition + tfd->set_fd_handler(cl->peer_fd, false, NULL); + clients_by_fd.erase(cl->peer_fd); + close(cl->peer_fd); + cl->peer_fd = -1; + return; + } if (log_level > 0) { fprintf(stderr, "[OSD %ju] client %ju disconnected\n", this->osd_num, cl->client_id); @@ -687,7 +702,10 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) // Inform that we're OSD payload["osd_num"] = osd_num; } - auto features = json11::Json::object{ { "check_sequencing", true } }; + auto features = json11::Json::object{ + { "check_sequencing", true }, + { "rdma_close_tcp", true }, + }; if (use_proto_checksums) { features["proto_checksums"] = use_proto_checksums; @@ -781,6 +799,10 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) cl->proto_csum_status = MSGR_CSUM_PAYLOAD; } #ifdef WITH_RDMA + if (config["features"]["rdma_close_tcp"].bool_value()) + { + cl->rdma_close_tcp = true; + } if (!use_rdmacm && cl->rdma_conn && config["rdma_address"].is_string()) { msgr_rdma_address_t addr; @@ -805,7 +827,7 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) { fprintf(stderr, "Connected to OSD %ju using RDMA\n", cl->osd_num); } - cl->peer_state = PEER_RDMA; + cl->peer_state = (cl->rdma_close_tcp ? PEER_RDMA_CONNECTING_OUT : PEER_RDMA); // Add the initial receive request init_recv_rdma(cl); } diff --git a/src/client/messenger.h b/src/client/messenger.h index f8ff16f0..b02eea6d 100644 --- a/src/client/messenger.h +++ b/src/client/messenger.h @@ -35,9 +35,10 @@ #define PEER_CONNECTING 1 #define PEER_CONNECTED 2 -#define PEER_RDMA_CONNECTING 3 -#define PEER_RDMA 4 -#define PEER_STOPPED 5 +#define PEER_RDMA_CONNECTING_IN 3 +#define PEER_RDMA_CONNECTING_OUT 4 +#define PEER_RDMA 5 +#define PEER_STOPPED 6 #define MSGR_CSUM_PAYLOAD 1 #define MSGR_CSUM_FULL 2 @@ -126,6 +127,7 @@ struct osd_client_t std::vector recv_flags; uint64_t read_op_id = 1; bool check_sequencing = false; + bool rdma_close_tcp = false; bool enable_pg_locks = false; op_aes_xts_decrypt_t *xts_dec_ctx = NULL; size_t read_op_inline_decrypt_pos = 0; diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index 9831a1de..171fd910 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -541,7 +541,7 @@ bool osd_messenger_t::connect_rdma(uint64_t client_id, std::string rdma_address, { // Remember connection, but switch to RDMA only after sending the configuration response cl->rdma_conn = rdma_conn; - cl->peer_state = PEER_RDMA_CONNECTING; + cl->peer_state = PEER_RDMA_CONNECTING_IN; return true; } } @@ -700,7 +700,7 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) auto rc = cl->rdma_conn; if (wc[i].status != IBV_WC_SUCCESS) { - fprintf(stderr, "RDMA work request failed for client %ju", client_id); + fprintf(stderr, "RDMA work request (%s) failed for client %ju", is_send ? "send" : "recv", client_id); if (cl->osd_num) { fprintf(stderr, " (OSD %ju)", cl->osd_num); @@ -711,6 +711,23 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) } if (!is_send) { + if (cl->peer_state == PEER_RDMA_CONNECTING_OUT) + { + // First message received over RDMA - TCP socket can be closed now + if (log_level > 0) + { + fprintf(stderr, "Successfully connected with client %ju using RDMA\n", cl->client_id); + } + cl->peer_state = PEER_RDMA; + if (cl->peer_fd >= 0 && cl->rdma_close_tcp) + { + // TCP socket is not needed anymore + tfd->set_fd_handler(cl->peer_fd, false, NULL); + clients_by_fd.erase(cl->peer_fd); + close(cl->peer_fd); + cl->peer_fd = -1; + } + } rc->cur_recv--; if (!handle_read_buffer(cl, rc->recv_buffers[rc->next_recv_buf], wc[i].byte_len)) { diff --git a/src/client/msgr_receive.cpp b/src/client/msgr_receive.cpp index 90ab4455..e69a0a93 100644 --- a/src/client/msgr_receive.cpp +++ b/src/client/msgr_receive.cpp @@ -396,6 +396,10 @@ public: } }; +#ifdef WITH_RDMA +#include "msgr_rdma.h" +#endif + void osd_messenger_t::read_requests() { for (int i = 0; i < read_ready_clients.size(); i++) @@ -403,7 +407,8 @@ void osd_messenger_t::read_requests() uint64_t client_id = read_ready_clients[i]; auto cl_it = clients.find(client_id); if (cl_it == clients.end() || !cl_it->second || cl_it->second->read_msg.msg_iovlen || - cl_it->second->peer_state != PEER_CONNECTED) + cl_it->second->peer_state == PEER_CONNECTING || + cl_it->second->peer_state == PEER_STOPPED) { continue; } diff --git a/src/client/msgr_send.cpp b/src/client/msgr_send.cpp index 1e22e8e7..d584bde4 100644 --- a/src/client/msgr_send.cpp +++ b/src/client/msgr_send.cpp @@ -487,7 +487,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) } cl->write_ops.push_back(cur_op); #ifdef WITH_RDMA - if (cl->peer_state == PEER_RDMA) + if (cl->peer_state == PEER_RDMA || cl->peer_state == PEER_RDMA_CONNECTING_OUT) { try_send_rdma(cl); return; @@ -826,14 +826,20 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t cl->proto_csum_status = cl->proto_csum_status & (~MSGR_CSUM_NEG); } #ifdef WITH_RDMA - if (cl->rdma_conn && !cl->write_op && !cl->write_ops.size() && cl->peer_state == PEER_RDMA_CONNECTING) + if (cl->rdma_conn && cl->peer_state == PEER_RDMA_CONNECTING_IN && !cl->write_op && !cl->write_ops.size()) { - // FIXME: Ignore pings during RDMA state transition - if (log_level > 0) + if (cl->rdma_close_tcp) { - fprintf(stderr, "Successfully connected with client %ju using RDMA\n", cl->client_id); + cl->peer_state = PEER_RDMA_CONNECTING_OUT; + } + else + { + if (log_level > 0) + { + fprintf(stderr, "Successfully connected with client %ju using RDMA\n", cl->client_id); + } + cl->peer_state = PEER_RDMA; } - cl->peer_state = PEER_RDMA; // Add the initial receive request init_recv_rdma(cl); } diff --git a/src/osd/osd_secondary.cpp b/src/osd/osd_secondary.cpp index a61e68cf..c8933def 100644 --- a/src/osd/osd_secondary.cpp +++ b/src/osd/osd_secondary.cpp @@ -373,6 +373,13 @@ void osd_t::exec_show_config(osd_op_t *cur_op) cl->read_op_id = cur_op->req.hdr.id + 1; } auto features = json11::Json::object{ { "pg_locks", true } }; +#ifdef WITH_RDMA + if (msgr.is_rdma_enabled() && req_json["features"]["rdma_close_tcp"].bool_value()) + { + cl->rdma_close_tcp = true; + features["rdma_close_tcp"] = true; + } +#endif if (msgr.use_proto_checksums) { auto peer_csums = req_json["features"]["proto_checksums"].uint64_value();