diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index 5ea47bab..15ee8a6a 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -192,9 +192,9 @@ void osd_messenger_t::init() auto cl = cl_it->second; cl_it++; 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) + cl->peer_state != PEER_CONNECTED && + cl->peer_state != PEER_RDMA_CONNECTING_OUT && + cl->peer_state != PEER_RDMA) { // Do not run keepalive on regular and unconnected clients continue; @@ -204,10 +204,21 @@ void osd_messenger_t::init() cl->ping_time_remaining--; if (!cl->ping_time_remaining) { - // Ping timed out, stop the client - fprintf(stderr, "Ping timed out for OSD %ju (client %ju), disconnecting peer\n", - cl->in_osd_num ? cl->in_osd_num : cl->osd_num, cl->client_id); - clients_to_stop.push_back(cl->client_id); + // Ping timed out + if (cl->peer_state == PEER_RDMA_CONNECTING_OUT && cl->peer_fd >= 0) + { + fprintf(stderr, "Ping timed out for OSD %ju over RDMA (client %ju), switching back to TCP\n", + cl->in_osd_num ? cl->in_osd_num : cl->osd_num, cl->client_id); + delete cl->rdma_conn; + cl->rdma_conn = NULL; + cl->peer_state = PEER_CONNECTED; + } + else + { + fprintf(stderr, "Ping timed out for OSD %ju (client %ju), disconnecting peer\n", + cl->in_osd_num ? cl->in_osd_num : cl->osd_num, cl->client_id); + clients_to_stop.push_back(cl->client_id); + } } } else if (cl->idle_time_remaining > 0) @@ -241,8 +252,18 @@ void osd_messenger_t::init() if (failed) { 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); + if (cl->peer_state == PEER_RDMA_CONNECTING_OUT && cl->peer_fd >= 0) + { + fprintf(stderr, "Ping failed for OSD %ju over RDMA (client %ju), switching back to TCP\n", fail_osd_num, cl->client_id); + delete cl->rdma_conn; + cl->rdma_conn = NULL; + cl->peer_state = PEER_CONNECTED; + } + else + { + 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; @@ -830,6 +851,11 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) cl->peer_state = (cl->rdma_close_tcp ? PEER_RDMA_CONNECTING_OUT : PEER_RDMA); // Add the initial receive request init_recv_rdma(cl); + // Check the connection by using a ping + cl->ping_time_remaining = osd_ping_timeout; + cl->idle_time_remaining = 0; + delete op; + return; } } #endif diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index 171fd910..8be29175 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -643,6 +643,7 @@ static void try_recv_rdma_wr(osd_client_t *cl, void *buf) bool osd_messenger_t::init_recv_rdma(osd_client_t *cl) { + return true; auto rc = cl->rdma_conn; assert(!rc->recv_buf.buf); rc->recv_buf.buf = (uint8_t*)malloc_or_die(rc->max_msg * rc->max_recv); @@ -698,6 +699,11 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) continue; } auto rc = cl->rdma_conn; + if (!rc) + { + // Connection destroyed (fallback to TCP) + continue; + } if (wc[i].status != IBV_WC_SUCCESS) { fprintf(stderr, "RDMA work request (%s) failed for client %ju", is_send ? "send" : "recv", client_id); @@ -727,6 +733,11 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) close(cl->peer_fd); cl->peer_fd = -1; } + if (cl->osd_num) + { + osd_peers[cl->osd_num] = cl; + on_connect_peer(cl->osd_num, 0, cl->client_id); + } } 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 e69a0a93..02e94178 100644 --- a/src/client/msgr_receive.cpp +++ b/src/client/msgr_receive.cpp @@ -489,6 +489,15 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl) { return; } + if (cl->peer_state == PEER_RDMA_CONNECTING_IN || cl->peer_state == PEER_RDMA_CONNECTING_OUT) + { + // Data received over TCP, abort RDMA connection + fprintf(stderr, "Data received for RDMA client %ju over TCP, falling back to TCP\n", cl->client_id); + delete cl->rdma_conn; + cl->rdma_conn = NULL; + cl->peer_state = PEER_CONNECTED; + assert(cl->peer_fd >= 0); + } if (cl->peer_state == PEER_STOPPED) { if (cl->refs <= 0)