Close TCP connections for RDMA clients

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 cl = cl_it->second;
-                        auto fail_osd_num = cl->in_osd_num ? cl->in_osd_num : cl->osd_num;
+                        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);
+                            auto fail_osd_num = cl->in_osd_num ? cl->in_osd_num : cl->osd_num;
-                            stop_client(fail_client_id);
+                            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 <osd_num>
         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);
             }

src/client/messenger.h

@@ -35,9 +35,10 @@
 
 #define PEER_CONNECTING 1
 #define PEER_CONNECTED 2
-#define PEER_RDMA_CONNECTING 3
+#define PEER_RDMA_CONNECTING_IN 3
-#define PEER_RDMA 4
+#define PEER_RDMA_CONNECTING_OUT 4
-#define PEER_STOPPED 5
+#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<int> 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;

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

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;
         }

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 (cl->rdma_close_tcp)
-            if (log_level > 0)
             {
-                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);
         }

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