------------------------------------------------------------ revno: 37 revision-id: [EMAIL PROTECTED] parent: [EMAIL PROTECTED] committer: Peter Somogyi <[EMAIL PROTECTED]> branch nick: ctdb timestamp: Tue 2006-12-12 19:09:16 +0100 message: Raw implementation done. Let's start compilation... modified: ib/ibwrapper.c ibwrapper.c-20061204130028-0125b4f5a72f4b11 ib/ibwrapper.h ibwrapper.h-20061204130028-32755c6266dd3c49 ib/ibwrapper_internal.h ibwrapper_internal.h-20061204130028-47f0a7e658b16ca2 === modified file 'ib/ibwrapper.c' --- a/ib/ibwrapper.c 2006-12-11 18:56:15 +0000 +++ b/ib/ibwrapper.c 2006-12-12 18:09:16 +0000 @@ -42,33 +42,34 @@ #define IBW_LASTERR_BUFSIZE 512 static char ibw_lasterr[IBW_LASTERR_BUFSIZE]; +static void ibw_event_handler_verbs(struct event_context *ev, + struct fd_event *fde, uint16_t flags, void *private_data); + 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; + int i; 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); + pconn->buf = memalign(pctx->page_size, pctx->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); + pconn->mr = ibv_reg_mr(pctx->pd, pconn->buf, + pctx->qsize * pctx->max_msg_size, IBV_ACCESS_LOCAL_WRITE); if (!pconn->mr) { - sprintf(ibw_lasterr, "Couldn't allocate mr\n"); + sprintf(ibw_lasterr, "couldn't allocate mr\n"); return -1; } - pconn->wr_index = talloc_size(pconn, num_msg * sizeof(ibw_wr *)); + pconn->wr_index = talloc_size(pconn, pctx->qsize * sizeof(ibw_wr *)); - for(i=0; i<num_msg; i++) { + for(i=0; i<pctx->qsize; i++) { p = pconn->wr_index[i] = talloc_zero(pconn, ibw_wr); - p->msg = pconn->buf + (i * pconn->opts.max_msg_size); + p->msg = pconn->buf + (i * pctx->max_msg_size); p->wr_id = i; DLIST_ADD(pconn->mr_list_avail, p); @@ -82,17 +83,6 @@ ibw_ctx *pctx = talloc_get_type(ctx->internal, ibw_ctx_priv); assert(pctx!=NULL); - if (pctx->verbs_channel) { - ibv_destroy_comp_channel(pctx->verbs_channel); - pctx->verbs_channel = NULL; - } - - if (pctx->verbs_channel_event) { - /* TODO: do we have to do this here? */ - talloc_free(pctx->verbs_channel_event); - pctx->verbs_channel_event = NULL; - } - if (pctx->pd) { ibv_dealloc_pd(pctx->pd); pctx->pd = NULL; @@ -149,6 +139,15 @@ ibv_destroy_cq(pconn->cq); pconn->cq = NULL; } + if (pconn->verbs_channel) { + ibv_destroy_comp_channel(pconn->verbs_channel); + pconn->verbs_channel = NULL; + } + if (pconn->verbs_channel_event) { + /* TODO: do we have to do this here? */ + talloc_free(pconn->verbs_channel_event); + pconn->verbs_channel_event = NULL; + } if (pconn->cm_id) { rdma_destroy_id(pctx->cm_id); pctx->cm_id = NULL; @@ -195,30 +194,44 @@ struct ibv_qp_init_attr init_attr; int rc; + /* init mr */ 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); + /* init verbs */ + pconn->verbs_channel = ibv_create_comp_channel(pconn->cm_id->verbs); + if (!pconn->verbs_channel) { + sprintf(ibw_lasterr, "ibv_create_comp_channel failed %d\n", errno); + goto cleanup; + } + DEBUG(10, "created channel %p\n", pconn->channel); + + pconn->verbs_channel_event = event_add_fd(pctx->ectx, conn, + pconn->verbs_channel->fd, EVENT_FD_READ, ibw_event_handler_verbs, conn); + + /* init cq */ + pconn->cq = ibv_create_cq(conn->cm_id->verbs, pctx->qsize, + conn, pconn->verbs_channel, 0); if (cq==NULL) { sprintf(ibw_lasterr, "ibv_create_cq failed\n"); return -1; } - rc = ibv_req_notify_cq(pctx->cq, 0); + rc = ibv_req_notify_cq(pconn->cq, 0); if (rc) { sprintf(ibw_lasterr, "ibv_req_notify_cq failed with %d\n", rc); return rc; } + /* init qp */ 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; + init_attr.send_cq = pconn->cq; + init_attr.recv_cq = pconn->cq; rc = rdma_create_qp(conn->cm_id, pctx->pd, &init_attr); if (rc) { @@ -230,8 +243,83 @@ return rc; } -static void ibw_refill_cq(ibw_conn *conn) -{ +static int ibw_refill_cq_recv(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, rc; + struct ibv_sge list = { + .addr = (uintptr_t) NULL, + .length = pctx->max_msg_size, + .lkey = pconn->mr->lkey + }; + struct ibv_recv_wr wr = { + .wr_id = 0, + .sg_list = &list, + .num_sge = 1, + }; + struct ibv_recv_wr *bad_wr; + ibw_wr *p = pconn->wr_list_avail; + + if (p==NULL) { + sprintf(ibw_last_err, "out of wr_list_avail"); + DEBUG(0, ibw_last_err); + return -1; + } + DLIST_REMOVE(pconn->wr_list_avail, p); + DLIST_ADD(pconn->wr_list_used, p); + list.addr = p->msg; + wr.wr_id = p->wr_id; + + rc = ibv_post_recv(pconn->qp, &wr, &bad_wr); + if (rc) { + sprintf(ibw_lasterr, "ibv_post_recv failed with %d\n", rc); + DEBUG(0, ibw_last_err); + return -2; + } + + return 0; +} + +static int ibw_fill_cq(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, rc; + struct ibv_sge list = { + .addr = (uintptr_t) NULL, + .length = pctx->max_msg_size, + .lkey = pconn->mr->lkey + }; + struct ibv_recv_wr wr = { + .wr_id = 0, + .sg_list = &list, + .num_sge = 1, + }; + struct ibv_recv_wr *bad_wr; + ibw_wr *p; + + for(i = pctx->opts.max_recv_wr; i!=0; i--) { + p = pconn->wr_list_avail; + if (p==NULL) { + sprintf(ibw_last_err, "out of wr_list_avail"); + DEBUG(0, ibw_last_err); + return -1; + } + DLIST_REMOVE(pconn->wr_list_avail, p); + DLIST_ADD(pconn->wr_list_used, p); + list.addr = p->msg; + wr.wr_id = p->wr_id; + + rc = ibv_post_recv(pconn->qp, &wr, &bad_wr); + if (rc) { + sprintf(ibw_lasterr, "ibv_post_recv failed with %d\n", rc); + DEBUG(0, ibw_last_err); + return -2; + } + } + + return 0; } static int ibw_manage_connect(ibw_conn *conn, struct rdma_cm_id *cma_id) @@ -266,7 +354,6 @@ ibw_conn_priv *pconn = NULL; struct rdma_cm_id *cma_id = NULL; struct rdma_cm_event *event = NULL; - int error = 0; assert(ctx!=NULL); @@ -274,8 +361,7 @@ if (rc) { ctx->state = IBWS_ERROR; sprintf(ibw_lasterr, "rdma_get_cm_event error %d\n", rc); - DEBUG(0, ibw_lasterr); - return; + goto error; } cma_id = event->id; @@ -291,7 +377,7 @@ if (rc) { ctx->state = ERROR; sprintf(ibw_lasterr, "rdma_resolve_route error %d\n", rc); - DEBUG(0, ibw_lasterr); + goto error; } /* continued at RDMA_CM_EVENT_ROUTE_RESOLVED */ break; @@ -300,12 +386,13 @@ /* after RDMA_CM_EVENT_ADDR_RESOLVED: */ assert(pctx->state==IWINT_ADDR_RESOLVED); pctx->state = IWINT_ROUTE_RESOLVED; + assert(cma_id->context!=NULL); conn = talloc_get_type(cma_id->context, ibw_conn); pconn = talloc_get_type(conn->internal, ibw_conn_priv); rc = ibw_manage_connect(conn, cma_id); if (rc) - error = 1; + goto error; break; @@ -318,7 +405,6 @@ DEBUG(10, "pconn->cm_id %p\n", pconn->cm_id); conn->state = IBWC_INIT; - pctx->connstate_func(ctx, conn); /* continued at ibw_accept when invoked by the func above */ @@ -327,7 +413,7 @@ DEBUG(10, "pconn->cm_id %p wasn't accepted\n", pconn->cm_id); } else { if (ibw_setup_cq_qp(ctx, conn)) - error = 1; + goto error; } /* TODO: clarify whether if it's needed by upper layer: */ @@ -356,9 +442,8 @@ case RDMA_CM_EVENT_CONNECT_ERROR: case RDMA_CM_EVENT_UNREACHABLE: case RDMA_CM_EVENT_REJECTED: - DEBUG(0, "cma event %d, error %d\n", event->event, event->status); - error = 1; - break; + sprintf(ibw_lasterr, "cma event %d, error %d\n", event->event, event->status); + goto error; case RDMA_CM_EVENT_DISCONNECTED: if (cma_id!=ctx->cm_id) { @@ -369,6 +454,7 @@ talloc_free(conn); + /* if we are the last... */ if (ctx->conn_list==NULL) rdma_disconnect(ctx->cm_id); } else { @@ -380,41 +466,106 @@ break; case RDMA_CM_EVENT_DEVICE_REMOVAL: - DEBUG(0, "cma detected device removal!\n"); - error = 1; - break; + sprintf(ibw_lasterr, "cma detected device removal!\n"); + goto error; default: - DEBUG(0, "unknown event %d\n", event->event); - error = 1; - break; - } - - if (error) { - DEBUG(0, ibw_lasterr); - if (cma_id!=ctx->cm_id) { - conn = talloc_get_type(cma_id->context, ibw_conn); - conn->state = IBWC_ERROR; - pctx->connstate_func(NULL, conn); - } else { - ctx->state = IBWS_ERROR; - pctx->connstate_func(ctx, NULL); - } + sprintf(ibw_lasterr, "unknown event %d\n", event->event); + goto error; } if ((rc=rdma_ack_cm_event(event))) { sprintf(ibw_lasterr, "rdma_ack_cm_event failed with %d\n"); - DEBUG(0, ibw_lasterr, rc); + goto error; + } + + return; +error: + DEBUG(0, "cm event handler: %s", ibw_lasterr); + if (cma_id!=ctx->cm_id) { + conn = talloc_get_type(cma_id->context, ibw_conn); + if (conn) + conn->state = IBWC_ERROR; + pctx->connstate_func(NULL, conn); + } else { + ctx->state = IBWS_ERROR; + pctx->connstate_func(ctx, NULL); } } static void ibw_event_handler_verbs(struct event_context *ev, struct fd_event *fde, uint16_t flags, void *private_data) { - ibw_ctx *ctx = talloc_get_type(private_data, ibw_ctx); - ibw_ctx_priv *pctx = talloc_get_type(ctx->internal, ibw_ctx_priv); - - + ibw_ctx *conn = talloc_get_type(private_data, ibw_conn); + ibw_ctx_priv *pconn = talloc_get_type(conn->internal, ibw_ctx_priv); + ibw_ctx *pctx = talloc_get_type(conn->ctx->internal, ibw_ctx_priv); + + struct ibv_wc wc; + int rc; + + rc = ibv_poll_cq(conn->cq, 1, &wc); + if (rc!=1) { + sprintf(ibw_lasterr, "ibv_poll_cq error %d\n", rc); + goto error; + } + if (wc.status) { + sprintf(ibw_lasterr, "cq completion failed status %d\n", + wc.status); + goto error; + } + + switch(wc.opcode) { + case IBV_WC_SEND: + { + ibw_wr *p; + + DEBUG(10, "send completion\n"); + assert(pconn->qp->qp_num==wc.qp_num); + assert(wc.wr_id < pctx->qsize); + p = pconn->wr_index[wc.wr_id]; + DLIST_REMOVE(pconn->wr_list_used, p); + DLIST_ADD(pconn->wr_list_avail, p); + } + break; + + case IBV_WC_RDMA_WRITE: + DEBUG(10, "rdma write completion\n"); + break; + + case IBV_WC_RDMA_READ: + DEBUG(10, "rdma read completion\n"); + break; + + case IBV_WC_RECV: + { + ibw_wr *p; + + assert(pconn->qp->qp_num==wc.qp_num); + assert(wc.wr_id < pctx->qsize); + p = pconn->wr_index[wc.wr_id]; + + DLIST_REMOVE(pconn->wr_list_used, p); + DLIST_ADD(pconn->wr_list_avail, p); + + DEBUG(10, "recv completion\n"); + assert(wc.byte_len <= pctx->max_msg_size); + + pctx->receive_func(conn, p->msg, wc.byte_len); + if (ibw_refill_cq_recv(conn)) + goto error; + } + break; + + default: + sprintf(ibw_lasterr, "unknown completion %d\n", wc.opcode); + goto error; + } + + return; +error: + DEBUG(0, ibw_lasterr); + conn->status = IBWC_ERROR; + pctx->connstate_func(NULL, conn); } static int ibw_process_init_attrs(ibw_initattr *attr, int nattr, ibw_opts *opts) @@ -424,7 +575,6 @@ 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; @@ -435,8 +585,6 @@ 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; @@ -449,7 +597,8 @@ void *ctx_userdata, ibw_connstate_fn_t ibw_connstate, ibw_receive_fn_t ibw_receive, - event_content *ectx) + event_content *ectx, + int max_msg_size) { ibw_ctx *ctx = talloc_zero(NULL, ibw_ctx); ibw_ctx_priv *pctx; @@ -487,7 +636,7 @@ pctx->cm_channel_event = event_add_fd(pctx->ectx, pctx, pctx->cm_channel->fd, EVENT_FD_READ, ibw_event_handler_cm, ctx); - rc = rdma_create_id(pctx->cm_channel, &pctx->cm_id, cb, RDMA_PS_TCP); + rc = rdma_create_id(pctx->cm_channel, &pctx->cm_id, ctx, RDMA_PS_TCP); if (rc) { rc = errno; sprintf(ibw_lasterr, "rdma_create_id error %d\n", rc); @@ -503,17 +652,9 @@ } DEBUG(10, "created pd %p\n", pctx->pd); - pctx->verbs_channel = ibv_create_comp_channel(cm_id->verbs); - if (!pctx->verbs_channel) { - sprintf(ibw_lasterr, "ibv_create_comp_channel failed %d\n", errno); - goto cleanup; - } - 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->pagesize = sysconf(_SC_PAGESIZE); + pctx->qsize = pctx->opts.max_send_wr + pctx->opts.max_recv_wr; + pctx->max_msg_size = max_msg_size; return ctx; /* don't put code here */ @@ -529,7 +670,16 @@ int ibw_stop(ibw_ctx *ctx) { ibw_ctx_priv *pctx = (ibw_ctx_priv *)ctx->internal; - + ibw_conn *p; + + for(p=ctx->conn_list; p!=NULL; p=p->next) { + if (ctx->state==IBWC_ERROR || ctx->state==IBWC_CONNECTED) { + if (ibw_disconnect(p)) + return -1; + } + } + + return 0; } int ibw_bind(ibw_ctx *ctx, struct sockaddr_in *my_addr) @@ -633,7 +783,7 @@ return 0; } -int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key, int *maxsize) +int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key) { 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); @@ -644,8 +794,6 @@ return -1; } - *maxsize = pctx->opts.max_msg_size; - DLIST_REMOVE(pctx->wr_list_avail, p); DLIST_ADD(pctx->wr_list_used, p); @@ -674,8 +822,7 @@ struct ibv_send_wr *bad_wr; assert(p->msg==(char *)buf); - - p->conn = conn; /* set it only now */ + assert(n<=pctx->max_msg_size); return ibv_post_send(conn->qp, &wr, &bad_wr); }
=== modified file 'ib/ibwrapper.h' --- a/ib/ibwrapper.h 2006-12-11 18:56:15 +0000 +++ b/ib/ibwrapper.h 2006-12-12 18:09:16 +0000 @@ -87,13 +87,15 @@ /* * settings: array of (name, value) pairs * where name is one of: - * dev_name [default is the first one] - * rx_depth [default is 500] - * mtu [default is 1024] - * ib_port [default is 1] + * max_send_wr [default is 256] + * max_recv_wr [default is 1024] + * <...> * * Must be called _ONCE_ for each node. * + * max_msg_size is the maximum size of a message + * (max_send_wr + max_recv_wr) * max_msg_size bytes allocated per connection + * * returns non-NULL on success * * talloc_free must be called for the result in IBWS_STOPPED; @@ -104,7 +106,8 @@ void *ctx_userdata, ibw_connstate_fn_t ibw_connstate, ibw_receive_fn_t ibw_receive, - event_content *ectx); + event_content *ectx, + int max_msg_size); /* * Must be called in states of (IBWS_ERROR, IBWS_READY, IBWS_CONNECT_REQUEST) @@ -178,16 +181,17 @@ * 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. + * Don't use more space than maxsize (see ibw_init). * * Returns 0 on success. */ -int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key, int *maxsize); +int ibw_alloc_send_buf(ibw_conn *conn, void **buf, void **key); /* * Send the message in one * Can be invoked any times (should fit into buffers) and at any time * (in conn->state=IBWC_CONNECTED) + * n must be less or equal than max_msg_size (see ibw_init) * * You mustn't use (buf, key) any more for sending. */ === modified file 'ib/ibwrapper_internal.h' --- a/ib/ibwrapper_internal.h 2006-12-11 18:56:15 +0000 +++ b/ib/ibwrapper_internal.h 2006-12-12 18:09:16 +0000 @@ -24,12 +24,10 @@ typedef struct _ibw_opts { 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 */ + char *msg; /* initialized in ibw_init_memory once per connection */ 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; @@ -51,28 +49,28 @@ struct rdma_event_channel *cm_channel; struct fd_event *cm_channel_event; - struct rdma_event_channel *cm_channel; - struct fd_event *cm_channel_event; - struct ibv_comp_channel *verbs_channel; - struct fd_event *verbs_channel_event; - struct ibv_pd *pd; - ibw_connstate_fn_t connstate_func; - ibw_receive_fn_t receive_func; + ibw_connstate_fn_t connstate_func; /* see ibw_init */ + ibw_receive_fn_t receive_func; /* see ibw_init */ long pagesize; /* sysconf result for memalign */ + int qsize; /* opts.max_send_wr + opts.max_recv_wr */ + int max_msg_size; /* see ibw_init */ } ibw_ctx_priv; typedef struct _ibw_conn_priv { + struct ibv_comp_channel *verbs_channel; + struct fd_event *verbs_channel_event; + struct rdma_cm_id *cm_id; /* client's cm id */ int is_accepted; struct ibv_cq *cq; /* qp is in cm_id */ struct ibv_mr *mr; - char *buf; /* fixed size (opts.bufsize) buffer for send/recv */ + char *buf; /* fixed size (qsize * opts.max_msg_size) 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_wr **wr_index; /* array[0..(qsize-1)] of (ibw_wr *) */ } ibw_conn_priv;