diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index 15fc33c1..487fec33 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -431,6 +431,8 @@ msgr_rdma_connection_t *msgr_rdma_connection_t::create(msgr_rdma_context_t *ctx, conn->max_recv = max_recv; conn->max_sge = max_sge; conn->max_msg = max_msg; + conn->in_credit = max_recv; + conn->out_credit = max_recv; ibv_qp_init_attr init_attr = { .send_cq = ctx->cq, @@ -571,9 +573,11 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge) .wr_id = (uint64_t)(cl->peer_fd*2+1), .sg_list = sge, .num_sge = op_sge, - .opcode = IBV_WR_SEND, + .opcode = IBV_WR_SEND_WITH_IMM, .send_flags = IBV_SEND_SIGNALED, + .imm_data = cl->rdma_conn->in_credit, }; + cl->rdma_conn->in_credit = 0; int err = ibv_post_send(cl->rdma_conn->qp, &wr, &bad_wr); if (err || bad_wr) { @@ -581,6 +585,7 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge) exit(1); } cl->rdma_conn->cur_send++; + cl->rdma_conn->out_credit--; } static int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len) @@ -608,7 +613,7 @@ static int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len) void osd_messenger_t::try_send_rdma_odp(osd_client_t *cl) { auto rc = cl->rdma_conn; - if (!cl->send_list.size() || rc->cur_send >= rc->max_send) + if (!cl->send_list.size() || rc->cur_send >= rc->max_send || !rc->out_credit) { return; } @@ -623,7 +628,7 @@ void osd_messenger_t::try_send_rdma_odp(osd_client_t *cl) try_send_rdma_wr(cl, sge, op_sge); op_sge = 0; op_size = 0; - if (rc->cur_send >= rc->max_send) + if (rc->cur_send >= rc->max_send || !rc->out_credit) { break; } @@ -672,7 +677,7 @@ void osd_messenger_t::try_send_rdma_nodp(osd_client_t *cl) uint8_t *dst = NULL; int dst_len = 0; int copied = 1; - while (!rc->send_out_full && copied > 0 && rc->cur_send < rc->max_send) + while (!rc->send_out_full && copied > 0 && rc->cur_send < rc->max_send && rc->out_credit > 0) { dst = (uint8_t*)rc->send_out.buf + rc->send_out_pos; dst_len = (rc->send_out_pos < rc->send_out_size ? rc->send_out_size-rc->send_out_pos : rc->send_done_pos-rc->send_out_pos); @@ -696,6 +701,16 @@ void osd_messenger_t::try_send_rdma_nodp(osd_client_t *cl) rc->send_sizes.push_back(copied); } } + if (rc->cur_send < rc->max_send && rc->in_credit > 0) + { + ibv_sge sge = { + .addr = (uintptr_t)NULL, + .length = (uint32_t)0, + .lkey = 0, + }; + try_send_rdma_wr(cl, &sge, 1); + rc->send_sizes.push_back(0); + } } void osd_messenger_t::try_send_rdma(osd_client_t *cl) @@ -726,6 +741,7 @@ static void try_recv_rdma_wr(osd_client_t *cl, void *buf) exit(1); } cl->rdma_conn->cur_recv++; + cl->rdma_conn->in_credit++; } bool osd_messenger_t::init_recv_rdma(osd_client_t *cl) @@ -799,6 +815,10 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) if (!is_send) { // Reset OSD ping state - client is obviously alive + if (wc[i].wc_flags & IBV_WC_WITH_IMM) + { + rc->out_credit += wc[i].imm_data; + } cl->ping_time_remaining = 0; cl->idle_time_remaining = osd_idle_timeout; rc->cur_recv--; diff --git a/src/client/msgr_rdma.h b/src/client/msgr_rdma.h index 0e93c57e..4fe70762 100644 --- a/src/client/msgr_rdma.h +++ b/src/client/msgr_rdma.h @@ -81,6 +81,8 @@ struct msgr_rdma_connection_t int send_out_pos = 0, send_done_pos = 0, send_out_size = 0; bool send_out_full = false; + uint32_t out_credit = 0, in_credit = 0; + ~msgr_rdma_connection_t(); static msgr_rdma_connection_t *create(msgr_rdma_context_t *ctx, uint32_t max_send, uint32_t max_recv, uint32_t max_sge, uint32_t max_msg); int connect(msgr_rdma_address_t *dest);