Rough implementation of buffer handling.

Many conceptual fix.

(This used to be ctdb commit 26ed351cfc883691446c792cd2d93847a31bc659)

diff --git a/ctdb/ib/ibwrapper.c b/ctdb/ib/ibwrapper.c
index 23396f7639be5f8345325512bb6718f4e765034f..fa447e90346292e71110db8f718f467895f7581f 100644 (file)
--- a/ctdb/ib/ibwrapper.c
+++ b/ctdb/ib/ibwrapper.c
@@ -42,8 +42,39 @@
 #define IBW_LASTERR_BUFSIZE 512
 static char ibw_lasterr[IBW_LASTERR_BUFSIZE];
 
-static ibw_mr *ibw_alloc_mr(ibw_ctx_priv *pctx)
+static int ibw_init_memory(ibw_conn *conn)
 {
+       ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, ibw_ctx_priv);
+       ibw_conn_priv *pconn = talloc_get_type(conn->internal, ibw_conn_priv);
+
+       int     i, num_msg;
+       ibw_wr  *p;
+
+       /* didn't find any reason to split send & recv buffer handling */
+       num_msg = pctx->opts.max_recv_wr + pctx->opts.max_send_wr;
+
+       pconn->buf = memalign(pctx->page_size, pctx->opts.max_msg_size);
+       if (!pconn->buf) {
+               sprintf(ibw_lasterr, "couldn't allocate work buf\n");
+               return -1;
+       }
+       pconn->mr = ibv_reg_mr(pctx->pd, pconn->buf, pctx->opts.bufsize, IBV_ACCESS_LOCAL_WRITE);
+       if (!pconn->mr) {
+               sprintf(ibw_lasterr, "Couldn't allocate mr\n");
+               return -1;
+       }
+
+       pconn->wr_index = talloc_size(pconn, num_msg * sizeof(ibw_wr *));
+
+       for(i=0; i<num_msg; i++) {
+               p = pconn->wr_index[i] = talloc_zero(pconn, ibw_wr);
+               p->msg = pconn->buf + (i * pconn->opts.max_msg_size);
+               p->wr_id = i;
+
+               DLIST_ADD(pconn->mr_list_avail, p);
+       }
+
+       return 0;
 }
 
 static int ibw_ctx_priv_destruct(void *ptr)
@@ -51,14 +82,6 @@ static int ibw_ctx_priv_destruct(void *ptr)
        ibw_ctx *pctx = talloc_get_type(ctx->internal, ibw_ctx_priv);
        assert(pctx!=NULL);
 
-       /* free memory regions */
-       
-       /* destroy verbs */
-       if (pctx->cq) {
-               ibv_destroy_cq(pctx->cq);
-               pctx->cq = NULL;
-       }
-
        if (pctx->verbs_channel) {
                ibv_destroy_comp_channel(pctx->verbs_channel);
                pctx->verbs_channel = NULL;
@@ -96,13 +119,6 @@ static int ibw_ctx_destruct(void *ptr)
        ibw_ctx *ctx = talloc_get_type(ptr, ibw_ctx);
        assert(ctx!=NULL);
 
-       if (pconn->cm_id) {
-               rdma_destroy_id(pconn->cm_id);
-               pconn->cm_id = NULL;
-       }
-
-       /* free memory regions */
-
        return 0;
 }
 
@@ -110,6 +126,33 @@ static int ibw_conn_priv_destruct(void *ptr)
 {
        ibw_conn *pconn = talloc_get_type(ptr, ibw_conn_priv);
        assert(pconn!=NULL);
+
+       /* free memory regions */
+       if (pconn->mr) {
+               ibv_dereg_mr(pconn->mr);
+               pconn->mr = NULL;
+       }
+       if (pconn->buf) {
+               free(pconn->buf); /* memalign-ed */
+               pconn->buf = NULL;
+       }
+
+       /* pconn->wr_index is freed by talloc */
+       /* pconn->wr_index[i] are freed by talloc */
+
+       /* destroy verbs */
+       if (pconn->cm_id->qp) {
+               ibv_destroy_qp(pconn->qp);
+               pconn->qp = NULL;
+       }
+       if (pconn->cq) {
+               ibv_destroy_cq(pconn->cq);
+               pconn->cq = NULL;
+       }
+       if (pconn->cm_id) {
+               rdma_destroy_id(pctx->cm_id);
+               pctx->cm_id = NULL;
+       }
 }
 
 static int ibw_conn_destruct(void *ptr)
@@ -145,12 +188,60 @@ static ibw_conn *ibw_conn_new(ibw_ctx *ctx)
        return conn;
 }
 
-static int ibw_manage_connect(struct rdma_cm_id *cma_id)
+static int ibw_setup_cq_qp(ibw_conn *conn)
+{
+       ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, ibw_ctx_priv);
+       ibw_conn_priv *pconn = talloc_get_type(conn->internal, ibw_conn_priv);
+       struct ibv_qp_init_attr init_attr;
+       int rc;
+
+       if (ibw_init_memory(conn))
+               return -1;
+
+       pctx->cq = ibv_create_cq(conn->cm_id->verbs, pctx->opts.max_send_wr + pctx->opts.max_recv_wr,
+               ctx, ctx->verbs_channel, 0);
+       if (cq==NULL) {
+               sprintf(ibw_lasterr, "ibv_create_cq failed\n");
+               return -1;
+       }
+
+       rc = ibv_req_notify_cq(pctx->cq, 0);
+       if (rc) {
+               sprintf(ibw_lasterr, "ibv_req_notify_cq failed with %d\n", rc);
+               return rc;
+       }
+
+       memset(&init_attr, 0, sizeof(init_attr));
+       init_attr.cap.max_send_wr = pctx->opts.max_send_wr;
+       init_attr.cap.max_recv_wr = pctx->opts.max_recv_wr;
+       init_attr.cap.max_recv_sge = 1;
+       init_attr.cap.max_send_sge = 1;
+       init_attr.qp_type = IBV_QPT_RC;
+       init_attr.send_cq = ctx->cq;
+       init_attr.recv_cq = ctx->cq;
+
+       rc = rdma_create_qp(conn->cm_id, pctx->pd, &init_attr);
+       if (rc) {
+               sprintf(ibw_lasterr, "rdma_create_qp (%d) failed with %d\n", is_server, rc);
+               return rc;
+       }
+       /* elase result is in pconn->cm_id->qp */
+
+       return rc;
+}
+
+static void ibw_refill_cq(ibw_conn *conn)
+{
+}
+
+static int ibw_manage_connect(ibw_conn *conn, struct rdma_cm_id *cma_id)
 {
        struct rdma_conn_param conn_param;
        int     rc;
 
-       /* TODO: setup verbs... */
+       rc = ibw_setup_cq_qp(conn);
+       if (rc)
+               return -1;
 
        /* cm connect */
        memset(&conn_param, 0, sizeof conn_param);
@@ -179,7 +270,7 @@ static void ibw_event_handler_cm(struct event_context *ev,
 
        assert(ctx!=NULL);
 
-       rc = rdma_get_cm_event(cb->cm_channel, &event);
+       rc = rdma_get_cm_event(pctx->cm_channel, &event);
        if (rc) {
                ctx->state = IBWS_ERROR;
                sprintf(ibw_lasterr, "rdma_get_cm_event error %d\n", rc);
@@ -198,7 +289,7 @@ static void ibw_event_handler_cm(struct event_context *ev,
                pctx->state = IWINT_ADDR_RESOLVED;
                rc = rdma_resolve_route(cma_id, 2000);
                if (rc) {
-                       cb->state = ERROR;
+                       ctx->state = ERROR;
                        sprintf(ibw_lasterr, "rdma_resolve_route error %d\n", rc);
                        DEBUG(0, ibw_lasterr);
                }
@@ -212,7 +303,7 @@ static void ibw_event_handler_cm(struct event_context *ev,
                conn = talloc_get_type(cma_id->context, ibw_conn);
                pconn = talloc_get_type(conn->internal, ibw_conn_priv);
 
-               rc = ibw_manage_connect(cma_id);
+               rc = ibw_manage_connect(conn, cma_id);
                if (rc)
                        error = 1;
 
@@ -234,6 +325,9 @@ static void ibw_event_handler_cm(struct event_context *ev,
                if (!pconn->is_accepted) {
                        talloc_free(conn);
                        DEBUG(10, "pconn->cm_id %p wasn't accepted\n", pconn->cm_id);
+               } else {
+                       if (ibw_setup_cq_qp(ctx, conn))
+                               error = 1;
                }
 
                /* TODO: clarify whether if it's needed by upper layer: */
@@ -245,12 +339,12 @@ static void ibw_event_handler_cm(struct event_context *ev,
 
        case RDMA_CM_EVENT_ESTABLISHED:
                /* expected after ibw_accept and ibw_connect[not directly] */
-               DEBUG(0, "ESTABLISHED\n");
-               ctx->state = IBWS_READY;
+               DEBUG(0, "ESTABLISHED (conn: %u)\n", cma_id->context);
                conn = talloc_get_type(cma_id->context, ibw_conn);
                assert(conn!=NULL); /* important assumption */
                pconn = talloc_get_type(conn->internal, ibw_conn_priv);
 
+               /* client conn is up */
                conn->state = IBWC_CONNECTED;
 
                /* both ctx and conn have changed */
@@ -274,9 +368,13 @@ static void ibw_event_handler_cm(struct event_context *ev,
                        pctx->connstate_func(NULL, conn);
 
                        talloc_free(conn);
+
+                       if (ctx->conn_list==NULL)
+                               rdma_disconnect(ctx->cm_id);
                } else {
                        DEBUG(0, "server DISCONNECT event\n");
                        ctx->state = IBWS_STOPPED; /* ??? TODO: try it... */
+                       /* talloc_free(ctx) should be called within or after this func */
                        pctx->connstate_func(ctx, NULL);
                }
                break;
@@ -313,28 +411,32 @@ static void ibw_event_handler_cm(struct event_context *ev,
 static void ibw_event_handler_verbs(struct event_context *ev,
        struct fd_event *fde, uint16_t flags, void *private_data)
 {
-       int     rc;
        ibw_ctx *ctx = talloc_get_type(private_data, ibw_ctx);
        ibw_ctx_priv *pctx = talloc_get_type(ctx->internal, ibw_ctx_priv);
 
+
 }
 
 static int ibw_process_init_attrs(ibw_initattr *attr, int nattr, ibw_opts *opts)
 {
-       int     i;
+       int     i, mtu;
        char *name, *value;
-       
+
+       opts->max_send_wr = 256;
+       opts->max_recv_wr = 1024;
+       opts->max_msg_size = 1024;
+
        for(i=0; i<nattr; i++) {
                name = attr[i].name;
                value = attr[i].value;
 
                assert(name!=NULL && value!=NULL);
-               if (strcmp(name, "dev_name")==0)
-                       opts->opts.dev_name = talloc_strdup(ctx, value);
-               else if (strcmp(name, "rx_depth")==0)
-                       opts->rx_depth = atoi(value);
-               else if (strcmp(name, "mtu")==0)
-                       opts->mtu = atoi(value);
+               if (strcmp(name, "max_send_wr")==0)
+                       opts->max_send_wr = atoi(value);
+               else if (strcmp(name, "max_recv_wr")==0)
+                       opts->max_recv_wr = atoi(value);
+               else if (strcmp(name, "max_msg_size")==0)
+                       opts->bufsize = atoi(value);
                else {
                        sprintf(ibw_lasterr, "ibw_init: unknown name %s\n", name);
                        return -1;
@@ -378,8 +480,7 @@ ibw_ctx *ibw_init(ibw_initattr *attr, int nattr,
        /* init cm */
        pctx->cm_channel = rdma_create_event_channel();
        if (!pctx->cm_channel) {
-               ret = errno;
-               sprintf(ibw_lasterr, "rdma_create_event_channel error %d\n", ret);
+               sprintf(ibw_lasterr, "rdma_create_event_channel error %d\n", errno);
                goto cleanup;
        }
 
@@ -404,23 +505,21 @@ ibw_ctx *ibw_init(ibw_initattr *attr, int nattr,
 
        pctx->verbs_channel = ibv_create_comp_channel(cm_id->verbs);
        if (!pctx->verbs_channel) {
-               sprintf(stderr, "ibv_create_comp_channel failed %d\n", errno);
+               sprintf(ibw_lasterr, "ibv_create_comp_channel failed %d\n", errno);
                goto cleanup;
        }
-       DEBUG_LOG("created channel %p\n", pctx->channel);
+       DEBUG(10, "created channel %p\n", pctx->channel);
 
        pctx->verbs_channel_event = event_add_fd(pctx->ectx, pctx,
                pctx->verbs_channel->fd, EVENT_FD_READ, ibw_event_handler_verbs, ctx);
 
-       pctx->cq = ibv_create_cq(cm_id->verbs, pctx->opts.rx_depth, ctx,
-               ctx->verbs_channel, 0);
-
-       /* allocate ib memory regions */
+       pctx->pagesize = sysconf(_SC_PAGESIZE);
 
        return ctx;
-
+       /* don't put code here */
 cleanup:
        DEBUG(0, ibw_lasterr);
+
        if (ctx)
                talloc_free(ctx);
 
@@ -455,7 +554,7 @@ int ibw_listen(ibw_ctx *ctx, int backlog)
        int     rc;
 
        DEBUG_LOG("rdma_listen...\n");
-       rc = rdma_listen(cb->cm_id, backlog);
+       rc = rdma_listen(pctx->cm_id, backlog);
        if (rc) {
                sprintf(ibw_lasterr, "rdma_listen failed: %d\n", ret);
                DEBUG(0, ibw_lasterr);
@@ -471,6 +570,8 @@ int ibw_accept(ibw_ctx *ctx, ibw_conn *conn, void *conn_userdata)
        ibw_conn_priv *pconn = talloc_get_type(conn->internal, ibw_conn_priv);
        struct rdma_conn_param  conn_param;
 
+       conn->conn_userdata = conn_userdata;
+
        memset(&conn_param, 0, sizeof(struct rdma_conn_param));
        conn_param.responder_resources = 1;
        conn_param.initiator_depth = 1;
@@ -522,21 +623,61 @@ int ibw_connect(ibw_ctx *ctx, struct sockaddr_in *serv_addr, void *conn_userdata
 void ibw_disconnect(ibw_conn *conn)
 {
        ibw_conn_priv *pconn = talloc_get_type(conn->internal, ibw_conn_priv);
-       
+       ibw_ctx *ctx = conn->ctx;
+       ibw_ctx_priv *pctx = talloc_get_type(ctx->internal);
+
+       rdma_disconnect(pctx->cm_id);
+
+       /* continued at RDMA_CM_EVENT_DISCONNECTED */
+
        return 0;
 }
 
-int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key, int n)
+int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key, int *maxsize)
 {
        ibw_conn_priv *pconn = talloc_get_type(conn->internal, ibw_conn_priv);
+       ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, ibw_ctx_priv);
+       ibw_wr *p = pctx->wr_list_avail;
 
-       return 0;
+       if (p==NULL) {
+               sprintf(ibw_last_err, "insufficient wr chunks\n");
+               return -1;
+       }
+
+       *maxsize = pctx->opts.max_msg_size;
+
+       DLIST_REMOVE(pctx->wr_list_avail, p);
+       DLIST_ADD(pctx->wr_list_used, p);
+
+       *buf = (void *)p->msg;
+       *key = (void *)p;
+
+       return pctx->buf;
 }
 
 int ibw_send(ibw_conn *conn, void *buf, void *key, int n)
 {
-       ibw_conn_priv *pconn = (ibw_ctx_priv *)ctx->internal;
-       return 0;
+       ibw_ctx_priv pctx = talloc_get_type(conn->ctx->internal, ibw_ctx_priv);
+       ibw_wr *p = talloc_get_type(key, ibw_wr);
+       struct ibv_sge list = {
+               .addr   = (uintptr_t) p->msg,
+               .length = n,
+               .lkey   = pctx->mr->lkey
+       };
+       struct ibv_send_wr wr = {
+               .wr_id      = p->wr_id,
+               .sg_list    = &list,
+               .num_sge    = 1,
+               .opcode     = IBV_WR_SEND,
+               .send_flags = IBV_SEND_SIGNALED,
+       };
+       struct ibv_send_wr *bad_wr;
+
+       assert(p->msg==(char *)buf);
+
+       p->conn = conn; /* set it only now */
+
+       return ibv_post_send(conn->qp, &wr, &bad_wr);
 }
 
 const char *ibw_getLastError()

diff --git a/ctdb/ib/ibwrapper.h b/ctdb/ib/ibwrapper.h
index 9b9666a60c48ef61a909f9974369d0a60bf45f81..e731fa981dc5ac99373dda09331042b3e7e7f803 100644 (file)
--- a/ctdb/ib/ibwrapper.h
+++ b/ctdb/ib/ibwrapper.h
@@ -178,10 +178,11 @@ void ibw_disconnect(ibw_conn *conn);
  * You have to use this buf to fill in before send.
  * It's just to avoid memcpy.in ibw_send.
  * Use the same (buf, key) pair with ibw_send.
+ * Don't use more space than maxsize.
  *
  * Returns 0 on success.
  */
-int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key, int n);
+int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key, int *maxsize);
 
 /*
  * Send the message in one

diff --git a/ctdb/ib/ibwrapper_internal.h b/ctdb/ib/ibwrapper_internal.h
index 8a1a445abb3764fcca40e1614bd1ef261ccda6d2..b022c92c717a7e91f0caaa634371490a850d9d9a 100644 (file)
--- a/ctdb/ib/ibwrapper_internal.h
+++ b/ctdb/ib/ibwrapper_internal.h
@@ -21,18 +21,19 @@
  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
-typedef struct _ibw_mr {
-       struct ibv_mr *mr;
-       struct _ibw_mr *next, *prev;
-} ibw_mr;
-
 typedef struct _ibw_opts {
-       char    *dev_name;
-       int     rx_depth;
-       int     mtu;
-       int     ib_port;
+       int     max_send_wr;
+       int     max_recv_wr;
+       int     max_msg_size;
 } ibw_opts;
 
+typedef struct _ibw_wr {
+       char    *msg; /* initialized in ibw_init_memory once */
+       ibw_conn *conn; /*valid only when in wr_list_used */
+       int     wr_id; /* position in wr_index list; also used as wr id */
+       struct _ibw_wr *next, *prev; /* in wr_list_avail or wr_list_used */
+} ibw_wr;
+
 typedef enum {
        IWINT_INIT = 0,
        IWINT_ADDR_RESOLVED,
@@ -41,11 +42,6 @@ typedef enum {
 } ibw_state_ctx;
 
 typedef struct _ibw_ctx_priv {
-       ibw_mr *avail_first;
-       ibw_mr *avail_last;
-       ibw_mr *used_first;
-       ibw_mr *used_last;
-
        struct event_context *ectx;
 
        ibw_opts opts;
@@ -64,22 +60,19 @@ typedef struct _ibw_ctx_priv {
 
        ibw_connstate_fn_t connstate_func;
        ibw_receive_fn_t receive_func;
+
+       long    pagesize; /* sysconf result for memalign */
 } ibw_ctx_priv;
 
 typedef struct _ibw_conn_priv {
-       struct ibv_cq   *cq;
-       struct ibv_qp   *qp;
-
        struct rdma_cm_id *cm_id; /* client's cm id */
        int     is_accepted;
-} ibw_conn_priv;
 
-/* 
- * Must be called in all cases after selecting/polling
- * for FDs set via ibw_add_event_fn_t.
- *
- * fd_index: fd identifier passed in ibw_add_event_fn_t
- * with the same fd was set there.
- */
-//int ibw_process_event(ibw_ctx *ctx, int fd_index);
+       struct ibv_cq   *cq; /* qp is in cm_id */
+       struct ibv_mr *mr;
+       char *buf; /* fixed size (opts.bufsize) buffer for send/recv */
+       ibw_wr *wr_list_avail;
+       ibw_wr *wr_list_used;
+       ibw_wr **wr_index; /* array[0..(max_send_wr + max_recv_wr)-1] of (ibw_wr *) */
+} ibw_conn_priv;