WIP RDMA credits

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

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