WIP RDMA credits

src/client/http_client.cpp

@@ -23,7 +23,7 @@
 #define READ_BUFFER_SIZE 9000
 
 static std::string ws_format_frame(int type, uint64_t size);
-static bool ws_parse_frame(std::string & buf, int & type, std::string & res);
+static bool ws_parse_frame(std::string & buf, uint8_t & type, std::string & res);
 static void parse_http_headers(std::string & res, http_response_t *parsed);
 
 struct http_co_t
@@ -70,7 +70,7 @@ struct http_co_t
     void submit_read(bool check_timeout);
     void submit_send();
     bool handle_read();
-    void post_message(int type, const std::string & msg);
+    void post_message(uint8_t type, const std::string & msg);
     void send_request(const std::string & host, const std::string & request,
         const http_options_t & options, std::function<void(const http_response_t *response)> response_callback);
 };
@@ -199,12 +199,12 @@ void http_co_t::send_request(const std::string & host, const std::string & reque
     stackout();
 }
 
-void http_post_message(http_co_t *handler, int type, const std::string & msg)
+void http_post_message(http_co_t *handler, uint8_t type, const std::string & msg)
 {
     handler->post_message(type, msg);
 }
 
-void http_co_t::post_message(int type, const std::string & msg)
+void http_co_t::post_message(uint8_t type, const std::string & msg)
 {
     stackin();
     if (state == HTTP_CO_WEBSOCKET)
@@ -608,7 +608,13 @@ bool http_co_t::handle_read()
     {
         while (ws_parse_frame(response, parsed.ws_msg_type, parsed.body))
         {
-            response_callback(&parsed);
+            if (parsed.ws_msg_type == WS_PING)
+            {
+                // Reply with WS_PONG
+                post_message(WS_PONG, "");
+            }
+            else
+                response_callback(&parsed);
             parsed.body = "";
         }
     }
@@ -698,7 +704,7 @@ static std::string ws_format_frame(int type, uint64_t size)
     return res;
 }
 
-static bool ws_parse_frame(std::string & buf, int & type, std::string & res)
+static bool ws_parse_frame(std::string & buf, uint8_t & type, std::string & res)
 {
     uint64_t hdr = 2;
     if (buf.size() < hdr)

src/client/http_client.h

@@ -32,7 +32,7 @@ struct http_response_t
     int status_code = 0;
     std::string status_line;
     std::map<std::string, std::string> headers;
-    int ws_msg_type = -1;
+    uint8_t ws_msg_type = -1;
     std::string body;
 
     void parse_json_response(std::string & error, json11::Json & r) const;
@@ -46,5 +46,5 @@ http_co_t* open_websocket(timerfd_manager_t *tfd, const std::string & host, cons
     int timeout, std::function<void(const http_response_t *msg)> on_message);
 void http_request(http_co_t *handler, const std::string & host, const std::string & request,
     const http_options_t & options, std::function<void(const http_response_t *response)> response_callback);
-void http_post_message(http_co_t *handler, int type, const std::string & msg);
+void http_post_message(http_co_t *handler, uint8_t type, const std::string & msg);
 void http_close(http_co_t *co);

src/client/messenger.cpp

@@ -749,7 +749,7 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
                     }
                 });
                 // Add the initial receive request
-                try_recv_rdma(cl);
+                init_recv_rdma(cl);
             }
         }
 #endif

src/client/messenger.h

@@ -289,7 +289,7 @@ protected:
     void try_send_rdma(osd_client_t *cl);
     void try_send_rdma_odp(osd_client_t *cl);
     void try_send_rdma_nodp(osd_client_t *cl);
-    bool try_recv_rdma(osd_client_t *cl);
+    bool init_recv_rdma(osd_client_t *cl);
     void handle_rdma_events(msgr_rdma_context_t *rdma_context);
     msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl);
 #endif

src/client/msgr_rdma.cpp

@@ -67,6 +67,20 @@ msgr_rdma_context_t::~msgr_rdma_context_t()
         ibv_close_device(context);
 }
 
+msgr_rdma_buf_t::~msgr_rdma_buf_t()
+{
+    if (buf)
+    {
+        free(buf);
+        buf = NULL;
+    }
+    if (mr)
+    {
+        ibv_dereg_mr(mr);
+        mr = NULL;
+    }
+}
+
 msgr_rdma_connection_t::~msgr_rdma_connection_t()
 {
     ctx->reserve_cqe(-max_send-max_recv);
@@ -84,26 +98,6 @@ msgr_rdma_connection_t::~msgr_rdma_connection_t()
     if (qp)
         ibv_destroy_qp(qp);
 #endif
-    if (recv_buffers.size())
-    {
-        for (auto b: recv_buffers)
-        {
-            if (b.mr)
-                ibv_dereg_mr(b.mr);
-            free(b.buf);
-        }
-        recv_buffers.clear();
-    }
-    if (send_out.mr)
-    {
-        ibv_dereg_mr(send_out.mr);
-        send_out.mr = NULL;
-    }
-    if (send_out.buf)
-    {
-        free(send_out.buf);
-        send_out.buf = NULL;
-    }
     send_out_size = 0;
 }
 
@@ -437,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,
@@ -493,12 +489,12 @@ int msgr_rdma_connection_t::connect(msgr_rdma_address_t *dest)
             .grh        = {
                 .dgid = dest->gid,
                 .sgid_index = conn->ctx->gid_index,
-                .hop_limit = 1, // FIXME can it vary?
+                .hop_limit = 64, // FIXME can it vary?
             },
             .dlid       = dest->lid,
             .sl         = 0, // service level
             .src_path_bits = 0,
-            .is_global  = (uint8_t)(dest->gid.global.interface_id ? 1 : 0),
+            .is_global  = (uint8_t)(dest->gid.global.subnet_prefix || dest->gid.global.interface_id ? 1 : 0),
             .port_num   = conn->ctx->ib_port,
         },
         .max_rd_atomic  = 1,
@@ -577,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)
     {
@@ -587,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)
@@ -614,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;
     }
@@ -629,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;
             }
@@ -663,7 +662,7 @@ void osd_messenger_t::try_send_rdma_nodp(osd_client_t *cl)
     {
         // Allocate send ring buffer, if not yet
         rc->send_out_size = rc->max_msg*rdma_max_send;
-        rc->send_out.buf = malloc_or_die(rc->send_out_size);
+        rc->send_out.buf = (uint8_t*)malloc_or_die(rc->send_out_size);
         if (!rc->ctx->odp)
         {
             rc->send_out.mr = ibv_reg_mr(rc->ctx->pd, rc->send_out.buf, rc->send_out_size, 0);
@@ -678,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);
@@ -702,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)
@@ -712,12 +721,12 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
         try_send_rdma_nodp(cl);
 }
 
-static void try_recv_rdma_wr(osd_client_t *cl, msgr_rdma_buf_t b)
+static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
 {
     ibv_sge sge = {
-        .addr = (uintptr_t)b.buf,
+        .addr = (uintptr_t)buf,
         .length = (uint32_t)cl->rdma_conn->max_msg,
-        .lkey = cl->rdma_conn->ctx->odp ? cl->rdma_conn->ctx->mr->lkey : b.mr->lkey,
+        .lkey = cl->rdma_conn->ctx->odp ? cl->rdma_conn->ctx->mr->lkey : cl->rdma_conn->recv_buf.mr->lkey,
     };
     ibv_recv_wr *bad_wr = NULL;
     ibv_recv_wr wr = {
@@ -732,24 +741,26 @@ static void try_recv_rdma_wr(osd_client_t *cl, msgr_rdma_buf_t b)
         exit(1);
     }
     cl->rdma_conn->cur_recv++;
+    cl->rdma_conn->in_credit++;
 }
 
-bool osd_messenger_t::try_recv_rdma(osd_client_t *cl)
+bool osd_messenger_t::init_recv_rdma(osd_client_t *cl)
 {
     auto rc = cl->rdma_conn;
-    while (rc->cur_recv < rc->max_recv)
+    assert(!rc->recv_buf.buf);
+    rc->recv_buf.buf = (uint8_t*)malloc_or_die(rc->max_msg * rc->max_recv);
+    if (!rc->ctx->odp)
     {
-        msgr_rdma_buf_t b;
+        rc->recv_buf.mr = ibv_reg_mr(rc->ctx->pd, rc->recv_buf.buf, rc->max_msg * rc->max_recv, IBV_ACCESS_LOCAL_WRITE);
-        b.buf = malloc_or_die(rc->max_msg);
+        if (!rc->recv_buf.mr)
-        if (!rc->ctx->odp)
         {
-            b.mr = ibv_reg_mr(rc->ctx->pd, b.buf, rc->max_msg, IBV_ACCESS_LOCAL_WRITE);
+            fprintf(stderr, "Failed to register RDMA memory region: %s\n", strerror(errno));
-            if (!b.mr)
+            exit(1);
-            {
-                fprintf(stderr, "Failed to register RDMA memory region: %s\n", strerror(errno));
-                exit(1);
-            }
         }
+    }
+    for (uint32_t i = 0; i < rc->max_recv; i++)
+    {
+        uint8_t *b = rc->recv_buf.buf + i*rc->max_msg;
         rc->recv_buffers.push_back(b);
         try_recv_rdma_wr(cl, b);
     }
@@ -804,10 +815,14 @@ 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--;
-                if (!handle_read_buffer(cl, rc->recv_buffers[rc->next_recv_buf].buf, wc[i].byte_len))
+                if (!handle_read_buffer(cl, rc->recv_buffers[rc->next_recv_buf], wc[i].byte_len))
                 {
                     // handle_read_buffer may stop the client
                     clear_immediate_ops(client_id);

src/client/msgr_rdma.h

@@ -54,8 +54,10 @@ struct msgr_rdma_context_t
 
 struct msgr_rdma_buf_t
 {
-    void *buf = NULL;
+    uint8_t *buf = NULL;
     ibv_mr *mr = NULL;
+
+    ~msgr_rdma_buf_t();
 };
 
 struct msgr_rdma_connection_t
@@ -72,12 +74,15 @@ struct msgr_rdma_connection_t
     int cur_send = 0, cur_recv = 0;
     int send_pos = 0, send_buf_pos = 0;
     int next_recv_buf = 0;
-    std::vector<msgr_rdma_buf_t> recv_buffers;
+    std::vector<void*> recv_buffers;
+    msgr_rdma_buf_t recv_buf;
     std::vector<uint64_t> send_sizes;
     msgr_rdma_buf_t send_out;
     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);

src/client/msgr_rdmacm.cpp

@@ -531,7 +531,7 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
         fprintf(stderr, "Successfully connected with OSD %ju using RDMA-CM\n", peer_osd);
     }
     // Add initial receive request(s)
-    try_recv_rdma(cl);
+    init_recv_rdma(cl);
     if (peer_osd)
     {
         check_peer_config(cl);

src/client/msgr_send.cpp

@@ -363,7 +363,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
                 }
             });
             // Add the initial receive request
-            try_recv_rdma(cl);
+            init_recv_rdma(cl);
         }
 #endif
     }