source4/cluster/ctdb/ib/ibwrapper.c

   1 /*
   2  * Unix SMB/CIFS implementation.
   3  * Wrap Infiniband calls.
   4  *
   5  * Copyright (C) Sven Oehme <oehmes@de.ibm.com> 2006
   6  *
   7  * Major code contributions by Peter Somogyi <psomogyi@gamax.hu>
   8  *
   9  * This program is free software; you can redistribute it and/or modify
  10  * it under the terms of the GNU General Public License as published by
  11  * the Free Software Foundation; either version 3 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU General Public License
  20  * along with this program; if not, see <http://www.gnu.org/licenses/>.
  21  */
  22
  23 #include <stdlib.h>
  24 #include <string.h>
  25 #include <stdio.h>
  26 #include <errno.h>
  27 #include <sys/types.h>
  28 #include <netinet/in.h>
  29 #include <sys/socket.h>
  30 #include <netdb.h>
  31 #include <arpa/inet.h>
  32 #include <malloc.h>
  33 #include <assert.h>
  34 #include <unistd.h>
  35
  36 #include "includes.h"
  37 #include "lib/events/events.h"
  38 #include "ibwrapper.h"
  39
  40 #include <rdma/rdma_cma.h>
  41 #include "infiniband/sa-kern-abi.h"
  42
  43 #include "ibwrapper_internal.h"
  44 #include "lib/util/dlinklist.h"
  45
  46 #define IBW_LASTERR_BUFSIZE 512
  47 static char ibw_lasterr[IBW_LASTERR_BUFSIZE];
  48
  49 #define IBW_MAX_SEND_WR 256
  50 #define IBW_MAX_RECV_WR 1024
  51 #define IBW_RECV_BUFSIZE 256
  52 #define IBW_RECV_THRESHOLD (1 * 1024 * 1024)
  53
  54 static void ibw_event_handler_verbs(struct event_context *ev,
  55         struct fd_event *fde, uint16_t flags, void *private_data);
  56 static int ibw_fill_cq(struct ibw_conn *conn);
  57 static int ibw_wc_recv(struct ibw_conn *conn, struct ibv_wc *wc);
  58 static int ibw_wc_send(struct ibw_conn *conn, struct ibv_wc *wc);
  59 static int ibw_send_packet(struct ibw_conn *conn, void *buf, struct ibw_wr *p, uint32_t len);
  60
  61 static void *ibw_alloc_mr(struct ibw_ctx_priv *pctx, struct ibw_conn_priv *pconn,
  62         uint32_t n, struct ibv_mr **ppmr)
  63 {
  64         void *buf;
  65
  66         DEBUG(10, ("ibw_alloc_mr(cmid=%p, n=%u)\n", pconn->cm_id, n));
  67         buf = memalign(pctx->pagesize, n);
  68         if (!buf) {
  69                 sprintf(ibw_lasterr, "couldn't allocate memory\n");
  70                 return NULL;
  71         }
  72
  73         *ppmr = ibv_reg_mr(pconn->pd, buf, n, IBV_ACCESS_LOCAL_WRITE);
  74         if (!*ppmr) {
  75                 sprintf(ibw_lasterr, "couldn't allocate mr\n");
  76                 free(buf);
  77                 return NULL;
  78         }
  79
  80         return buf;
  81 }
  82
  83 static void ibw_free_mr(char **ppbuf, struct ibv_mr **ppmr)
  84 {
  85         DEBUG(10, ("ibw_free_mr(%u %u)\n", (uint32_t)*ppbuf, (uint32_t)*ppmr));
  86         if (*ppmr!=NULL) {
  87                 ibv_dereg_mr(*ppmr);
  88                 *ppmr = NULL;
  89         }
  90         if (*ppbuf) {
  91                 free(*ppbuf);
  92                 *ppbuf = NULL;
  93         }
  94 }
  95
  96 static int ibw_init_memory(struct ibw_conn *conn)
  97 {
  98         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
  99         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 100         struct ibw_opts *opts = &pctx->opts;
 101         int     i;
 102         struct ibw_wr   *p;
 103
 104         DEBUG(10, ("ibw_init_memory(cmid: %p)\n", pconn->cm_id));
 105         pconn->buf_send = ibw_alloc_mr(pctx, pconn,
 106                 opts->max_send_wr * opts->recv_bufsize, &pconn->mr_send);
 107         if (!pconn->buf_send) {
 108                 sprintf(ibw_lasterr, "couldn't allocate work send buf\n");
 109                 return -1;
 110         }
 111
 112         pconn->buf_recv = ibw_alloc_mr(pctx, pconn,
 113                 opts->max_recv_wr * opts->recv_bufsize, &pconn->mr_recv);
 114         if (!pconn->buf_recv) {
 115                 sprintf(ibw_lasterr, "couldn't allocate work recv buf\n");
 116                 return -1;
 117         }
 118
 119         pconn->wr_index = talloc_size(pconn, opts->max_send_wr * sizeof(struct ibw_wr *));
 120         assert(pconn->wr_index!=NULL);
 121
 122         for(i=0; i<opts->max_send_wr; i++) {
 123                 p = pconn->wr_index[i] = talloc_zero(pconn, struct ibw_wr);
 124                 p->buf = pconn->buf_send + (i * opts->recv_bufsize);
 125                 p->wr_id = i;
 126
 127                 DLIST_ADD(pconn->wr_list_avail, p);
 128         }
 129
 130         return 0;
 131 }
 132
 133 static int ibw_ctx_priv_destruct(struct ibw_ctx_priv *pctx)
 134 {
 135         DEBUG(10, ("ibw_ctx_priv_destruct(%u)\n", (uint32_t)pctx));
 136
 137         /* destroy cm */
 138         if (pctx->cm_channel) {
 139                 rdma_destroy_event_channel(pctx->cm_channel);
 140                 pctx->cm_channel = NULL;
 141         }
 142         if (pctx->cm_channel_event) {
 143                 /* TODO: do we have to do this here? */
 144                 talloc_free(pctx->cm_channel_event);
 145                 pctx->cm_channel_event = NULL;
 146         }
 147         if (pctx->cm_id) {
 148                 rdma_destroy_id(pctx->cm_id);
 149                 pctx->cm_id = NULL;
 150         }
 151
 152         return 0;
 153 }
 154
 155 static int ibw_ctx_destruct(struct ibw_ctx *ctx)
 156 {
 157         DEBUG(10, ("ibw_ctx_destruct(%u)\n", (uint32_t)ctx));
 158         return 0;
 159 }
 160
 161 static int ibw_conn_priv_destruct(struct ibw_conn_priv *pconn)
 162 {
 163         DEBUG(10, ("ibw_conn_priv_destruct(%p, cmid: %p)\n",
 164                 pconn, pconn->cm_id));
 165
 166         /* pconn->wr_index is freed by talloc */
 167         /* pconn->wr_index[i] are freed by talloc */
 168
 169         /* destroy verbs */
 170         if (pconn->cm_id!=NULL && pconn->cm_id->qp!=NULL) {
 171                 rdma_destroy_qp(pconn->cm_id);
 172                 pconn->cm_id->qp = NULL;
 173         }
 174
 175         if (pconn->cq!=NULL) {
 176                 ibv_destroy_cq(pconn->cq);
 177                 pconn->cq = NULL;
 178         }
 179
 180         if (pconn->verbs_channel!=NULL) {
 181                 ibv_destroy_comp_channel(pconn->verbs_channel);
 182                 pconn->verbs_channel = NULL;
 183         }
 184
 185         /* must be freed here because its order is important */
 186         if (pconn->verbs_channel_event) {
 187                 talloc_free(pconn->verbs_channel_event);
 188                 pconn->verbs_channel_event = NULL;
 189         }
 190
 191         /* free memory regions */
 192         ibw_free_mr(&pconn->buf_send, &pconn->mr_send);
 193         ibw_free_mr(&pconn->buf_recv, &pconn->mr_recv);
 194
 195         if (pconn->pd) {
 196                 ibv_dealloc_pd(pconn->pd);
 197                 pconn->pd = NULL;
 198                 DEBUG(10, ("pconn=%p pd deallocated\n", pconn));
 199         }
 200
 201         if (pconn->cm_id) {
 202                 rdma_destroy_id(pconn->cm_id);
 203                 pconn->cm_id = NULL;
 204                 DEBUG(10, ("pconn=%p cm_id destroyed\n", pconn));
 205         }
 206
 207         return 0;
 208 }
 209
 210 static int ibw_wr_destruct(struct ibw_wr *wr)
 211 {
 212         if (wr->buf_large!=NULL)
 213                 ibw_free_mr(&wr->buf_large, &wr->mr_large);
 214         return 0;
 215 }
 216
 217 static int ibw_conn_destruct(struct ibw_conn *conn)
 218 {
 219         DEBUG(10, ("ibw_conn_destruct(%u)\n", (uint32_t)conn));
 220
 221         /* important here: ctx is a talloc _parent_ */
 222         DLIST_REMOVE(conn->ctx->conn_list, conn);
 223         return 0;
 224 }
 225
 226 struct ibw_conn *ibw_conn_new(struct ibw_ctx *ctx, TALLOC_CTX *mem_ctx)
 227 {
 228         struct ibw_conn *conn;
 229         struct ibw_conn_priv *pconn;
 230
 231         assert(ctx!=NULL);
 232
 233         conn = talloc_zero(mem_ctx, struct ibw_conn);
 234         assert(conn!=NULL);
 235         talloc_set_destructor(conn, ibw_conn_destruct);
 236
 237         pconn = talloc_zero(conn, struct ibw_conn_priv);
 238         assert(pconn!=NULL);
 239         talloc_set_destructor(pconn, ibw_conn_priv_destruct);
 240
 241         conn->ctx = ctx;
 242         conn->internal = (void *)pconn;
 243
 244         DLIST_ADD(ctx->conn_list, conn);
 245
 246         return conn;
 247 }
 248
 249 static int ibw_setup_cq_qp(struct ibw_conn *conn)
 250 {
 251         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
 252         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 253         struct ibv_qp_init_attr init_attr;
 254         struct ibv_qp_attr attr;
 255         int rc;
 256
 257         DEBUG(10, ("ibw_setup_cq_qp(cmid: %p)\n", pconn->cm_id));
 258
 259         /* init verbs */
 260         pconn->verbs_channel = ibv_create_comp_channel(pconn->cm_id->verbs);
 261         if (!pconn->verbs_channel) {
 262                 sprintf(ibw_lasterr, "ibv_create_comp_channel failed %d\n", errno);
 263                 return -1;
 264         }
 265         DEBUG(10, ("created channel %p\n", pconn->verbs_channel));
 266
 267         pconn->verbs_channel_event = event_add_fd(pctx->ectx, NULL, /* not pconn or conn */
 268                 pconn->verbs_channel->fd, EVENT_FD_READ, ibw_event_handler_verbs, conn);
 269
 270         pconn->pd = ibv_alloc_pd(pconn->cm_id->verbs);
 271         if (!pconn->pd) {
 272                 sprintf(ibw_lasterr, "ibv_alloc_pd failed %d\n", errno);
 273                 return -1;
 274         }
 275         DEBUG(10, ("created pd %p\n", pconn->pd));
 276
 277         /* init mr */
 278         if (ibw_init_memory(conn))
 279                 return -1;
 280
 281         /* init cq */
 282         pconn->cq = ibv_create_cq(pconn->cm_id->verbs,
 283                 pctx->opts.max_recv_wr + pctx->opts.max_send_wr,
 284                 conn, pconn->verbs_channel, 0);
 285         if (pconn->cq==NULL) {
 286                 sprintf(ibw_lasterr, "ibv_create_cq failed\n");
 287                 return -1;
 288         }
 289
 290         rc = ibv_req_notify_cq(pconn->cq, 0);
 291         if (rc) {
 292                 sprintf(ibw_lasterr, "ibv_req_notify_cq failed with %d\n", rc);
 293                 return rc;
 294         }
 295
 296         /* init qp */
 297         memset(&init_attr, 0, sizeof(init_attr));
 298         init_attr.cap.max_send_wr = pctx->opts.max_send_wr;
 299         init_attr.cap.max_recv_wr = pctx->opts.max_recv_wr;
 300         init_attr.cap.max_recv_sge = 1;
 301         init_attr.cap.max_send_sge = 1;
 302         init_attr.qp_type = IBV_QPT_RC;
 303         init_attr.send_cq = pconn->cq;
 304         init_attr.recv_cq = pconn->cq;
 305
 306         rc = rdma_create_qp(pconn->cm_id, pconn->pd, &init_attr);
 307         if (rc) {
 308                 sprintf(ibw_lasterr, "rdma_create_qp failed with %d\n", rc);
 309                 return rc;
 310         }
 311         /* elase result is in pconn->cm_id->qp */
 312
 313         rc = ibv_query_qp(pconn->cm_id->qp, &attr, IBV_QP_PATH_MTU, &init_attr);
 314         if (rc) {
 315                 sprintf(ibw_lasterr, "ibv_query_qp failed with %d\n", rc);
 316                 return rc;
 317         }
 318
 319         return ibw_fill_cq(conn);
 320 }
 321
 322 static int ibw_refill_cq_recv(struct ibw_conn *conn)
 323 {
 324         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
 325         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 326         int     rc;
 327         struct ibv_sge list = {
 328                 .addr   = (uintptr_t) NULL, /* filled below */
 329                 .length = pctx->opts.recv_bufsize,
 330                 .lkey   = pconn->mr_recv->lkey /* always the same */
 331         };
 332         struct ibv_recv_wr wr = {
 333                 .wr_id      = 0, /* filled below */
 334                 .sg_list    = &list,
 335                 .num_sge    = 1,
 336         };
 337         struct ibv_recv_wr *bad_wr;
 338
 339         DEBUG(10, ("ibw_refill_cq_recv(cmid: %p)\n", pconn->cm_id));
 340
 341         list.addr = (uintptr_t) pconn->buf_recv + pctx->opts.recv_bufsize * pconn->recv_index;
 342         wr.wr_id = pconn->recv_index;
 343         pconn->recv_index = (pconn->recv_index + 1) % pctx->opts.max_recv_wr;
 344
 345         rc = ibv_post_recv(pconn->cm_id->qp, &wr, &bad_wr);
 346         if (rc) {
 347                 sprintf(ibw_lasterr, "refill/ibv_post_recv failed with %d\n", rc);
 348                 DEBUG(0, (ibw_lasterr));
 349                 return -2;
 350         }
 351
 352         return 0;
 353 }
 354
 355 static int ibw_fill_cq(struct ibw_conn *conn)
 356 {
 357         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
 358         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 359         int     i, rc;
 360         struct ibv_sge list = {
 361                 .addr   = (uintptr_t) NULL, /* filled below */
 362                 .length = pctx->opts.recv_bufsize,
 363                 .lkey   = pconn->mr_recv->lkey /* always the same */
 364         };
 365         struct ibv_recv_wr wr = {
 366                 .wr_id      = 0, /* filled below */
 367                 .sg_list    = &list,
 368                 .num_sge    = 1,
 369         };
 370         struct ibv_recv_wr *bad_wr;
 371
 372         DEBUG(10, ("ibw_fill_cq(cmid: %p)\n", pconn->cm_id));
 373
 374         for(i = pctx->opts.max_recv_wr; i!=0; i--) {
 375                 list.addr = (uintptr_t) pconn->buf_recv + pctx->opts.recv_bufsize * pconn->recv_index;
 376                 wr.wr_id = pconn->recv_index;
 377                 pconn->recv_index = (pconn->recv_index + 1) % pctx->opts.max_recv_wr;
 378
 379                 rc = ibv_post_recv(pconn->cm_id->qp, &wr, &bad_wr);
 380                 if (rc) {
 381                         sprintf(ibw_lasterr, "fill/ibv_post_recv failed with %d\n", rc);
 382                         DEBUG(0, (ibw_lasterr));
 383                         return -2;
 384                 }
 385         }
 386
 387         return 0;
 388 }
 389
 390 static int ibw_manage_connect(struct ibw_conn *conn)
 391 {
 392         struct rdma_conn_param conn_param;
 393         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 394         int     rc;
 395
 396         DEBUG(10, ("ibw_manage_connect(cmid: %p)\n", pconn->cm_id));
 397
 398         if (ibw_setup_cq_qp(conn))
 399                 return -1;
 400
 401         /* cm connect */
 402         memset(&conn_param, 0, sizeof conn_param);
 403         conn_param.responder_resources = 1;
 404         conn_param.initiator_depth = 1;
 405         conn_param.retry_count = 10;
 406
 407         rc = rdma_connect(pconn->cm_id, &conn_param);
 408         if (rc)
 409                 sprintf(ibw_lasterr, "rdma_connect error %d\n", rc);
 410
 411         return rc;
 412 }
 413
 414 static void ibw_event_handler_cm(struct event_context *ev,
 415         struct fd_event *fde, uint16_t flags, void *private_data)
 416 {
 417         int     rc;
 418         struct ibw_ctx  *ctx = talloc_get_type(private_data, struct ibw_ctx);
 419         struct ibw_ctx_priv *pctx = talloc_get_type(ctx->internal, struct ibw_ctx_priv);
 420         struct ibw_conn *conn = NULL;
 421         struct ibw_conn_priv *pconn = NULL;
 422         struct rdma_cm_id *cma_id = NULL;
 423         struct rdma_cm_event *event = NULL;
 424
 425         assert(ctx!=NULL);
 426
 427         rc = rdma_get_cm_event(pctx->cm_channel, &event);
 428         if (rc) {
 429                 ctx->state = IBWS_ERROR;
 430                 sprintf(ibw_lasterr, "rdma_get_cm_event error %d\n", rc);
 431                 goto error;
 432         }
 433         cma_id = event->id;
 434
 435         DEBUG(10, ("cma_event type %d cma_id %p (%s)\n", event->event, cma_id,
 436                   (cma_id == pctx->cm_id) ? "parent" : "child"));
 437
 438         switch (event->event) {
 439         case RDMA_CM_EVENT_ADDR_RESOLVED:
 440                 DEBUG(11, ("RDMA_CM_EVENT_ADDR_RESOLVED\n"));
 441                 /* continuing from ibw_connect ... */
 442                 rc = rdma_resolve_route(cma_id, 2000);
 443                 if (rc) {
 444                         sprintf(ibw_lasterr, "rdma_resolve_route error %d\n", rc);
 445                         goto error;
 446                 }
 447                 /* continued at RDMA_CM_EVENT_ROUTE_RESOLVED */
 448                 break;
 449
 450         case RDMA_CM_EVENT_ROUTE_RESOLVED:
 451                 DEBUG(11, ("RDMA_CM_EVENT_ROUTE_RESOLVED\n"));
 452                 /* after RDMA_CM_EVENT_ADDR_RESOLVED: */
 453                 assert(cma_id->context!=NULL);
 454                 conn = talloc_get_type(cma_id->context, struct ibw_conn);
 455
 456                 rc = ibw_manage_connect(conn);
 457                 if (rc)
 458                         goto error;
 459
 460                 break;
 461
 462         case RDMA_CM_EVENT_CONNECT_REQUEST:
 463                 DEBUG(11, ("RDMA_CM_EVENT_CONNECT_REQUEST\n"));
 464                 ctx->state = IBWS_CONNECT_REQUEST;
 465                 conn = ibw_conn_new(ctx, ctx);
 466                 pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 467                 pconn->cm_id = cma_id; /* !!! event will be freed but id not */
 468                 cma_id->context = (void *)conn;
 469                 DEBUG(10, ("pconn->cm_id %p\n", pconn->cm_id));
 470
 471                 if (ibw_setup_cq_qp(conn))
 472                         goto error;
 473
 474                 conn->state = IBWC_INIT;
 475                 pctx->connstate_func(ctx, conn);
 476
 477                 /* continued at ibw_accept when invoked by the func above */
 478                 if (!pconn->is_accepted) {
 479                         rc = rdma_reject(cma_id, NULL, 0);
 480                         if (rc)
 481                                 DEBUG(0, ("rdma_reject failed with rc=%d\n", rc));
 482                         talloc_free(conn);
 483                         DEBUG(10, ("pconn->cm_id %p wasn't accepted\n", pconn->cm_id));
 484                 }
 485
 486                 /* TODO: clarify whether if it's needed by upper layer: */
 487                 ctx->state = IBWS_READY;
 488                 pctx->connstate_func(ctx, NULL);
 489
 490                 /* NOTE: more requests can arrive until RDMA_CM_EVENT_ESTABLISHED ! */
 491                 break;
 492
 493         case RDMA_CM_EVENT_ESTABLISHED:
 494                 /* expected after ibw_accept and ibw_connect[not directly] */
 495                 DEBUG(0, ("ESTABLISHED (conn: %p)\n", cma_id->context));
 496                 conn = talloc_get_type(cma_id->context, struct ibw_conn);
 497                 assert(conn!=NULL); /* important assumption */
 498
 499                 DEBUG(10, ("ibw_setup_cq_qp succeeded (cmid=%p)\n", cma_id));
 500
 501                 /* client conn is up */
 502                 conn->state = IBWC_CONNECTED;
 503
 504                 /* both ctx and conn have changed */
 505                 pctx->connstate_func(ctx, conn);
 506                 break;
 507
 508         case RDMA_CM_EVENT_ADDR_ERROR:
 509                 sprintf(ibw_lasterr, "RDMA_CM_EVENT_ADDR_ERROR, error %d\n", event->status);
 510         case RDMA_CM_EVENT_ROUTE_ERROR:
 511                 sprintf(ibw_lasterr, "RDMA_CM_EVENT_ROUTE_ERROR, error %d\n", event->status);
 512         case RDMA_CM_EVENT_CONNECT_ERROR:
 513                 sprintf(ibw_lasterr, "RDMA_CM_EVENT_CONNECT_ERROR, error %d\n", event->status);
 514         case RDMA_CM_EVENT_UNREACHABLE:
 515                 sprintf(ibw_lasterr, "RDMA_CM_EVENT_UNREACHABLE, error %d\n", event->status);
 516         case RDMA_CM_EVENT_REJECTED:
 517                 sprintf(ibw_lasterr, "RDMA_CM_EVENT_REJECTED, error %d\n", event->status);
 518                 conn = talloc_get_type(cma_id->context, struct ibw_conn);
 519                 if (conn) {
 520                         if ((rc=rdma_ack_cm_event(event)))
 521                                 DEBUG(0, ("reject/rdma_ack_cm_event failed with %d\n", rc));
 522                         event = NULL;
 523                         pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 524                         ibw_conn_priv_destruct(pconn);
 525                 }
 526                 goto error;
 527
 528         case RDMA_CM_EVENT_DISCONNECTED:
 529                 DEBUG(11, ("RDMA_CM_EVENT_DISCONNECTED\n"));
 530                 if ((rc=rdma_ack_cm_event(event)))
 531                         DEBUG(0, ("disc/rdma_ack_cm_event failed with %d\n", rc));
 532                 event = NULL; /* don't ack more */
 533
 534                 if (cma_id!=pctx->cm_id) {
 535                         DEBUG(0, ("client DISCONNECT event cm_id=%p\n", cma_id));
 536                         conn = talloc_get_type(cma_id->context, struct ibw_conn);
 537                         conn->state = IBWC_DISCONNECTED;
 538                         pctx->connstate_func(NULL, conn);
 539                 }
 540                 break;
 541
 542         case RDMA_CM_EVENT_DEVICE_REMOVAL:
 543                 sprintf(ibw_lasterr, "cma detected device removal!\n");
 544                 goto error;
 545
 546         default:
 547                 sprintf(ibw_lasterr, "unknown event %d\n", event->event);
 548                 goto error;
 549         }
 550
 551         if (event!=NULL && (rc=rdma_ack_cm_event(event))) {
 552                 sprintf(ibw_lasterr, "rdma_ack_cm_event failed with %d\n", rc);
 553                 goto error;
 554         }
 555
 556         return;
 557 error:
 558         if (event!=NULL && (rc=rdma_ack_cm_event(event))) {
 559                 sprintf(ibw_lasterr, "rdma_ack_cm_event failed with %d\n", rc);
 560                 goto error;
 561         }
 562
 563         DEBUG(0, ("cm event handler: %s", ibw_lasterr));
 564
 565         if (cma_id!=pctx->cm_id) {
 566                 conn = talloc_get_type(cma_id->context, struct ibw_conn);
 567                 if (conn)
 568                         conn->state = IBWC_ERROR;
 569                 pctx->connstate_func(NULL, conn);
 570         } else {
 571                 ctx->state = IBWS_ERROR;
 572                 pctx->connstate_func(ctx, NULL);
 573         }
 574 }
 575
 576 static void ibw_event_handler_verbs(struct event_context *ev,
 577         struct fd_event *fde, uint16_t flags, void *private_data)
 578 {
 579         struct ibw_conn *conn = talloc_get_type(private_data, struct ibw_conn);
 580         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 581         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
 582
 583         struct ibv_wc wc;
 584         int rc;
 585         struct ibv_cq *ev_cq;
 586         void          *ev_ctx;
 587
 588         DEBUG(10, ("ibw_event_handler_verbs(%u)\n", (uint32_t)flags));
 589
 590         /* TODO: check whether if it's good to have more channels here... */
 591         rc = ibv_get_cq_event(pconn->verbs_channel, &ev_cq, &ev_ctx);
 592         if (rc) {
 593                 sprintf(ibw_lasterr, "Failed to get cq_event with %d\n", rc);
 594                 goto error;
 595         }
 596         if (ev_cq != pconn->cq) {
 597                 sprintf(ibw_lasterr, "ev_cq(%p) != pconn->cq(%p)\n", ev_cq, pconn->cq);
 598                 goto error;
 599         }
 600         rc = ibv_req_notify_cq(pconn->cq, 0);
 601         if (rc) {
 602                 sprintf(ibw_lasterr, "Couldn't request CQ notification (%d)\n", rc);
 603                 goto error;
 604         }
 605
 606         while((rc=ibv_poll_cq(pconn->cq, 1, &wc))==1) {
 607                 if (wc.status) {
 608                         sprintf(ibw_lasterr, "cq completion failed status=%d, opcode=%d, rc=%d\n",
 609                                 wc.status, wc.opcode, rc);
 610                         goto error;
 611                 }
 612
 613                 switch(wc.opcode) {
 614                 case IBV_WC_SEND:
 615                         DEBUG(10, ("send completion\n"));
 616                         if (ibw_wc_send(conn, &wc))
 617                                 goto error;
 618                         break;
 619
 620                 case IBV_WC_RDMA_WRITE:
 621                         DEBUG(10, ("rdma write completion\n"));
 622                         break;
 623
 624                 case IBV_WC_RDMA_READ:
 625                         DEBUG(10, ("rdma read completion\n"));
 626                         break;
 627
 628                 case IBV_WC_RECV:
 629                         DEBUG(10, ("recv completion\n"));
 630                         if (ibw_wc_recv(conn, &wc))
 631                                 goto error;
 632                         break;
 633
 634                 default:
 635                         sprintf(ibw_lasterr, "unknown completion %d\n", wc.opcode);
 636                         goto error;
 637                 }
 638         }
 639         if (rc!=0) {
 640                 sprintf(ibw_lasterr, "ibv_poll_cq error %d\n", rc);
 641                 goto error;
 642         }
 643
 644         ibv_ack_cq_events(pconn->cq, 1);
 645
 646         return;
 647 error:
 648         ibv_ack_cq_events(pconn->cq, 1);
 649
 650         DEBUG(0, (ibw_lasterr));
 651
 652         if (conn->state!=IBWC_ERROR) {
 653                 conn->state = IBWC_ERROR;
 654                 pctx->connstate_func(NULL, conn);
 655         }
 656 }
 657
 658 static int ibw_process_queue(struct ibw_conn *conn)
 659 {
 660         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 661         struct ibw_ctx_priv *pctx;
 662         struct ibw_wr   *p;
 663         int     rc;
 664         uint32_t        msg_size;
 665
 666         if (pconn->queue==NULL)
 667                 return 0; /* NOP */
 668
 669         p = pconn->queue;
 670
 671         /* we must have at least 1 fragment to send */
 672         assert(p->queued_ref_cnt>0);
 673         p->queued_ref_cnt--;
 674
 675         pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
 676         msg_size = (p->queued_ref_cnt) ? pctx->opts.recv_bufsize : p->queued_rlen;
 677
 678         assert(p->queued_msg!=NULL);
 679         assert(msg_size!=0);
 680
 681         DEBUG(10, ("ibw_process_queue refcnt=%d msgsize=%u\n",
 682                 p->queued_ref_cnt, msg_size));
 683
 684         rc = ibw_send_packet(conn, p->queued_msg, p, msg_size);
 685
 686         /* was this the last fragment? */
 687         if (p->queued_ref_cnt) {
 688                 p->queued_msg += pctx->opts.recv_bufsize;
 689         } else {
 690                 DLIST_REMOVE2(pconn->queue, p, qprev, qnext);
 691                 p->queued_msg = NULL;
 692         }
 693
 694         return rc;
 695 }
 696
 697 static int ibw_wc_send(struct ibw_conn *conn, struct ibv_wc *wc)
 698 {
 699         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
 700         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 701         struct ibw_wr   *p;
 702         int     send_index;
 703
 704         DEBUG(10, ("ibw_wc_send(cmid: %p, wr_id: %u, bl: %u)\n",
 705                 pconn->cm_id, (uint32_t)wc->wr_id, (uint32_t)wc->byte_len));
 706
 707         assert(pconn->cm_id->qp->qp_num==wc->qp_num);
 708         assert(wc->wr_id >= pctx->opts.max_recv_wr);
 709         send_index = wc->wr_id - pctx->opts.max_recv_wr;
 710         pconn->wr_sent--;
 711
 712         if (send_index < pctx->opts.max_send_wr) {
 713                 DEBUG(10, ("ibw_wc_send#1 %u\n", (int)wc->wr_id));
 714                 p = pconn->wr_index[send_index];
 715                 if (p->buf_large!=NULL) {
 716                         if (p->ref_cnt) {
 717                                 /* awaiting more of it... */
 718                                 p->ref_cnt--;
 719                         } else {
 720                                 ibw_free_mr(&p->buf_large, &p->mr_large);
 721                                 DLIST_REMOVE(pconn->wr_list_used, p);
 722                                 DLIST_ADD(pconn->wr_list_avail, p);
 723                         }
 724                 } else { /* nasty - but necessary */
 725                         DLIST_REMOVE(pconn->wr_list_used, p);
 726                         DLIST_ADD(pconn->wr_list_avail, p);
 727                 }
 728         } else { /* "extra" request - not optimized */
 729                 DEBUG(10, ("ibw_wc_send#2 %u\n", (int)wc->wr_id));
 730                 for(p=pconn->extra_sent; p!=NULL; p=p->next)
 731                         if ((p->wr_id + pctx->opts.max_recv_wr)==(int)wc->wr_id)
 732                                 break;
 733                 if (p==NULL) {
 734                         sprintf(ibw_lasterr, "failed to find wr_id %d\n", (int)wc->wr_id);
 735                                 return -1;
 736                 }
 737                 if (p->ref_cnt) {
 738                         p->ref_cnt--;
 739                 } else {
 740                         ibw_free_mr(&p->buf_large, &p->mr_large);
 741                         DLIST_REMOVE(pconn->extra_sent, p);
 742                         DLIST_ADD(pconn->extra_avail, p);
 743                 }
 744         }
 745
 746         return ibw_process_queue(conn);
 747 }
 748
 749 static int ibw_append_to_part(struct ibw_conn_priv *pconn,
 750         struct ibw_part *part, char **pp, uint32_t add_len, int info)
 751 {
 752         DEBUG(10, ("ibw_append_to_part: cmid=%p, (bs=%u, len=%u, tr=%u), al=%u, i=%u\n",
 753                 pconn->cm_id, part->bufsize, part->len, part->to_read, add_len, info));
 754
 755         /* allocate more if necessary - it's an "evergrowing" buffer... */
 756         if (part->len + add_len > part->bufsize) {
 757                 if (part->buf==NULL) {
 758                         assert(part->len==0);
 759                         part->buf = talloc_size(pconn, add_len);
 760                         if (part->buf==NULL) {
 761                                 sprintf(ibw_lasterr, "recv talloc_size error (%u) #%d\n",
 762                                         add_len, info);
 763                                 return -1;
 764                         }
 765                         part->bufsize = add_len;
 766                 } else {
 767                         part->buf = talloc_realloc_size(pconn,
 768                                 part->buf, part->len + add_len);
 769                         if (part->buf==NULL) {
 770                                 sprintf(ibw_lasterr, "recv realloc error (%u + %u) #%d\n",
 771                                         part->len, add_len, info);
 772                                 return -1;
 773                         }
 774                 }
 775                 part->bufsize = part->len + add_len;
 776         }
 777
 778         /* consume pp */
 779         memcpy(part->buf + part->len, *pp, add_len);
 780         *pp += add_len;
 781         part->len += add_len;
 782         part->to_read -= add_len;
 783
 784         return 0;
 785 }
 786
 787 static int ibw_wc_mem_threshold(struct ibw_conn_priv *pconn,
 788         struct ibw_part *part, uint32_t threshold)
 789 {
 790         DEBUG(10, ("ibw_wc_mem_threshold: cmid=%p, (bs=%u, len=%u, tr=%u), thr=%u\n",
 791                 pconn->cm_id, part->bufsize, part->len, part->to_read, threshold));
 792
 793         if (part->bufsize > threshold) {
 794                 DEBUG(3, ("ibw_wc_mem_threshold: cmid=%p, %u > %u\n",
 795                         pconn->cm_id, part->bufsize, threshold));
 796                 talloc_free(part->buf);
 797                 part->buf = talloc_size(pconn, threshold);
 798                 if (part->buf==NULL) {
 799                         sprintf(ibw_lasterr, "talloc_size failed\n");
 800                         return -1;
 801                 }
 802                 part->bufsize = threshold;
 803         }
 804         return 0;
 805 }
 806
 807 static int ibw_wc_recv(struct ibw_conn *conn, struct ibv_wc *wc)
 808 {
 809         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
 810         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
 811         struct ibw_part *part = &pconn->part;
 812         char    *p;
 813         uint32_t        remain = wc->byte_len;
 814
 815         DEBUG(10, ("ibw_wc_recv: cmid=%p, wr_id: %u, bl: %u\n",
 816                 pconn->cm_id, (uint32_t)wc->wr_id, remain));
 817
 818         assert(pconn->cm_id->qp->qp_num==wc->qp_num);
 819         assert((int)wc->wr_id < pctx->opts.max_recv_wr);
 820         assert(wc->byte_len <= pctx->opts.recv_bufsize);
 821
 822         p = pconn->buf_recv + ((int)wc->wr_id * pctx->opts.recv_bufsize);
 823
 824         while(remain) {
 825                 /* here always true: (part->len!=0 && part->to_read!=0) ||
 826                         (part->len==0 && part->to_read==0) */
 827                 if (part->len) { /* is there a partial msg to be continued? */
 828                         int read_len = (part->to_read<=remain) ? part->to_read : remain;
 829                         if (ibw_append_to_part(pconn, part, &p, read_len, 421))
 830                                 goto error;
 831                         remain -= read_len;
 832
 833                         if (part->len<=sizeof(uint32_t) && part->to_read==0) {
 834                                 assert(part->len==sizeof(uint32_t));
 835                                 /* set it again now... */
 836                                 part->to_read = *((uint32_t *)(part->buf)); /* TODO: ntohl */
 837                                 if (part->to_read<sizeof(uint32_t)) {
 838                                         sprintf(ibw_lasterr, "got msglen=%u #2\n", part->to_read);
 839                                         goto error;
 840                                 }
 841                                 part->to_read -= sizeof(uint32_t); /* it's already read */
 842                         }
 843
 844                         if (part->to_read==0) {
 845                                 pctx->receive_func(conn, part->buf, part->len);
 846                                 part->len = 0; /* tells not having partial data (any more) */
 847                                 if (ibw_wc_mem_threshold(pconn, part, pctx->opts.recv_threshold))
 848                                         goto error;
 849                         }
 850                 } else {
 851                         if (remain>=sizeof(uint32_t)) {
 852                                 uint32_t msglen = *(uint32_t *)p; /* TODO: ntohl */
 853                                 if (msglen<sizeof(uint32_t)) {
 854                                         sprintf(ibw_lasterr, "got msglen=%u\n", msglen);
 855                                         goto error;
 856                                 }
 857
 858                                 /* mostly awaited case: */
 859                                 if (msglen<=remain) {
 860                                         pctx->receive_func(conn, p, msglen);
 861                                         p += msglen;
 862                                         remain -= msglen;
 863                                 } else {
 864                                         part->to_read = msglen;
 865                                         /* part->len is already 0 */
 866                                         if (ibw_append_to_part(pconn, part, &p, remain, 422))
 867                                                 goto error;
 868                                         remain = 0; /* to be continued ... */
 869                                         /* part->to_read > 0 here */
 870                                 }
 871                         } else { /* edge case: */
 872                                 part->to_read = sizeof(uint32_t);
 873                                 /* part->len is already 0 */
 874                                 if (ibw_append_to_part(pconn, part, &p, remain, 423))
 875                                         goto error;
 876                                 remain = 0;
 877                                 /* part->to_read > 0 here */
 878                         }
 879                 }
 880         } /* <remain> is always decreased at least by 1 */
 881
 882         if (ibw_refill_cq_recv(conn))
 883                 goto error;
 884
 885         return 0;
 886
 887 error:
 888         DEBUG(0, ("ibw_wc_recv error: %s", ibw_lasterr));
 889         return -1;
 890 }
 891
 892 static int ibw_process_init_attrs(struct ibw_initattr *attr, int nattr, struct ibw_opts *opts)
 893 {
 894         int     i;
 895         const char *name, *value;
 896
 897         DEBUG(10, ("ibw_process_init_attrs: nattr: %d\n", nattr));
 898
 899         opts->max_send_wr = IBW_MAX_SEND_WR;
 900         opts->max_recv_wr = IBW_MAX_RECV_WR;
 901         opts->recv_bufsize = IBW_RECV_BUFSIZE;
 902         opts->recv_threshold = IBW_RECV_THRESHOLD;
 903
 904         for(i=0; i<nattr; i++) {
 905                 name = attr[i].name;
 906                 value = attr[i].value;
 907
 908                 assert(name!=NULL && value!=NULL);
 909                 if (strcmp(name, "max_send_wr")==0)
 910                         opts->max_send_wr = atoi(value);
 911                 else if (strcmp(name, "max_recv_wr")==0)
 912                         opts->max_recv_wr = atoi(value);
 913                 else if (strcmp(name, "recv_bufsize")==0)
 914                         opts->recv_bufsize = atoi(value);
 915                 else if (strcmp(name, "recv_threshold")==0)
 916                         opts->recv_threshold = atoi(value);
 917                 else {
 918                         sprintf(ibw_lasterr, "ibw_init: unknown name %s\n", name);
 919                         return -1;
 920                 }
 921         }
 922         return 0;
 923 }
 924
 925 struct ibw_ctx *ibw_init(struct ibw_initattr *attr, int nattr,
 926         void *ctx_userdata,
 927         ibw_connstate_fn_t ibw_connstate,
 928         ibw_receive_fn_t ibw_receive,
 929         struct event_context *ectx)
 930 {
 931         struct ibw_ctx *ctx = talloc_zero(NULL, struct ibw_ctx);
 932         struct ibw_ctx_priv *pctx;
 933         int     rc;
 934
 935         DEBUG(10, ("ibw_init(ctx_userdata: %p, ectx: %p)\n", ctx_userdata, ectx));
 936
 937         /* initialize basic data structures */
 938         memset(ibw_lasterr, 0, IBW_LASTERR_BUFSIZE);
 939
 940         assert(ctx!=NULL);
 941         ibw_lasterr[0] = '\0';
 942         talloc_set_destructor(ctx, ibw_ctx_destruct);
 943         ctx->ctx_userdata = ctx_userdata;
 944
 945         pctx = talloc_zero(ctx, struct ibw_ctx_priv);
 946         talloc_set_destructor(pctx, ibw_ctx_priv_destruct);
 947         ctx->internal = (void *)pctx;
 948         assert(pctx!=NULL);
 949
 950         pctx->connstate_func = ibw_connstate;
 951         pctx->receive_func = ibw_receive;
 952
 953         pctx->ectx = ectx;
 954
 955         /* process attributes */
 956         if (ibw_process_init_attrs(attr, nattr, &pctx->opts))
 957                 goto cleanup;
 958
 959         /* init cm */
 960         pctx->cm_channel = rdma_create_event_channel();
 961         if (!pctx->cm_channel) {
 962                 sprintf(ibw_lasterr, "rdma_create_event_channel error %d\n", errno);
 963                 goto cleanup;
 964         }
 965
 966         pctx->cm_channel_event = event_add_fd(pctx->ectx, pctx,
 967                 pctx->cm_channel->fd, EVENT_FD_READ, ibw_event_handler_cm, ctx);
 968
 969         rc = rdma_create_id(pctx->cm_channel, &pctx->cm_id, ctx, RDMA_PS_TCP);
 970         if (rc) {
 971                 rc = errno;
 972                 sprintf(ibw_lasterr, "rdma_create_id error %d\n", rc);
 973                 goto cleanup;
 974         }
 975         DEBUG(10, ("created cm_id %p\n", pctx->cm_id));
 976
 977         pctx->pagesize = sysconf(_SC_PAGESIZE);
 978
 979         return ctx;
 980         /* don't put code here */
 981 cleanup:
 982         DEBUG(0, (ibw_lasterr));
 983
 984         if (ctx)
 985                 talloc_free(ctx);
 986
 987         return NULL;
 988 }
 989
 990 int ibw_stop(struct ibw_ctx *ctx)
 991 {
 992         struct ibw_ctx_priv *pctx = (struct ibw_ctx_priv *)ctx->internal;
 993         struct ibw_conn *p;
 994
 995         DEBUG(10, ("ibw_stop\n"));
 996
 997         for(p=ctx->conn_list; p!=NULL; p=p->next) {
 998                 if (ctx->state==IBWC_ERROR || ctx->state==IBWC_CONNECTED) {
 999                         if (ibw_disconnect(p))
1000                                 return -1;
1001                 }
1002         }
1003
1004         ctx->state = IBWS_STOPPED;
1005         pctx->connstate_func(ctx, NULL);
1006
1007         return 0;
1008 }
1009
1010 int ibw_bind(struct ibw_ctx *ctx, struct sockaddr_in *my_addr)
1011 {
1012         struct ibw_ctx_priv *pctx = (struct ibw_ctx_priv *)ctx->internal;
1013         int     rc;
1014
1015         DEBUG(10, ("ibw_bind: addr=%s, port=%u\n",
1016                 inet_ntoa(my_addr->sin_addr), ntohs(my_addr->sin_port)));
1017         rc = rdma_bind_addr(pctx->cm_id, (struct sockaddr *) my_addr);
1018         if (rc) {
1019                 sprintf(ibw_lasterr, "rdma_bind_addr error %d\n", rc);
1020                 DEBUG(0, (ibw_lasterr));
1021                 return rc;
1022         }
1023         DEBUG(10, ("rdma_bind_addr successful\n"));
1024
1025         return 0;
1026 }
1027
1028 int ibw_listen(struct ibw_ctx *ctx, int backlog)
1029 {
1030         struct ibw_ctx_priv *pctx = talloc_get_type(ctx->internal, struct ibw_ctx_priv);
1031         int     rc;
1032
1033         DEBUG(10, ("ibw_listen\n"));
1034         rc = rdma_listen(pctx->cm_id, backlog);
1035         if (rc) {
1036                 sprintf(ibw_lasterr, "rdma_listen failed: %d\n", rc);
1037                 DEBUG(0, (ibw_lasterr));
1038                 return rc;
1039         }
1040
1041         return 0;
1042 }
1043
1044 int ibw_accept(struct ibw_ctx *ctx, struct ibw_conn *conn, void *conn_userdata)
1045 {
1046         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
1047         struct rdma_conn_param  conn_param;
1048         int     rc;
1049
1050         DEBUG(10, ("ibw_accept: cmid=%p\n", pconn->cm_id));
1051         conn->conn_userdata = conn_userdata;
1052
1053         memset(&conn_param, 0, sizeof(struct rdma_conn_param));
1054         conn_param.responder_resources = 1;
1055         conn_param.initiator_depth = 1;
1056         rc = rdma_accept(pconn->cm_id, &conn_param);
1057         if (rc) {
1058                 sprintf(ibw_lasterr, "rdma_accept failed %d\n", rc);
1059                 DEBUG(0, (ibw_lasterr));
1060                 return -1;;
1061         }
1062
1063         pconn->is_accepted = 1;
1064
1065         /* continued at RDMA_CM_EVENT_ESTABLISHED */
1066
1067         return 0;
1068 }
1069
1070 int ibw_connect(struct ibw_conn *conn, struct sockaddr_in *serv_addr, void *conn_userdata)
1071 {
1072         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
1073         struct ibw_conn_priv *pconn = NULL;
1074         int     rc;
1075
1076         assert(conn!=NULL);
1077
1078         conn->conn_userdata = conn_userdata;
1079         pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
1080         DEBUG(10, ("ibw_connect: addr=%s, port=%u\n", inet_ntoa(serv_addr->sin_addr),
1081                 ntohs(serv_addr->sin_port)));
1082
1083         /* clean previous - probably half - initialization */
1084         if (ibw_conn_priv_destruct(pconn)) {
1085                 DEBUG(0, ("ibw_connect/ibw_pconn_destruct failed for cm_id=%p\n", pconn->cm_id));
1086                 return -1;
1087         }
1088
1089         /* init cm */
1090         rc = rdma_create_id(pctx->cm_channel, &pconn->cm_id, conn, RDMA_PS_TCP);
1091         if (rc) {
1092                 rc = errno;
1093                 sprintf(ibw_lasterr, "ibw_connect/rdma_create_id error %d\n", rc);
1094                 talloc_free(conn);
1095                 return -1;
1096         }
1097         DEBUG(10, ("ibw_connect: rdma_create_id succeeded, cm_id=%p\n", pconn->cm_id));
1098
1099         rc = rdma_resolve_addr(pconn->cm_id, NULL, (struct sockaddr *) serv_addr, 2000);
1100         if (rc) {
1101                 sprintf(ibw_lasterr, "rdma_resolve_addr error %d\n", rc);
1102                 DEBUG(0, (ibw_lasterr));
1103                 talloc_free(conn);
1104                 return -1;
1105         }
1106
1107         /* continued at RDMA_CM_EVENT_ADDR_RESOLVED */
1108
1109         return 0;
1110 }
1111
1112 int ibw_disconnect(struct ibw_conn *conn)
1113 {
1114         int     rc;
1115         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
1116
1117         DEBUG(10, ("ibw_disconnect: cmid=%p\n", pconn->cm_id));
1118
1119         assert(pconn!=NULL);
1120
1121         switch(conn->state) {
1122         case IBWC_ERROR:
1123                 ibw_conn_priv_destruct(pconn); /* do this here right now */
1124                 break;
1125         case IBWC_CONNECTED:
1126                 rc = rdma_disconnect(pconn->cm_id);
1127                 if (rc) {
1128                         sprintf(ibw_lasterr, "ibw_disconnect failed with %d\n", rc);
1129                         DEBUG(0, (ibw_lasterr));
1130                         return rc;
1131                 }
1132                 break;
1133         default:
1134                 DEBUG(9, ("invalid state for disconnect: %d\n", conn->state));
1135                 break;
1136         }
1137
1138         return 0;
1139 }
1140
1141 int ibw_alloc_send_buf(struct ibw_conn *conn, void **buf, void **key, uint32_t len)
1142 {
1143         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
1144         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
1145         struct ibw_wr *p = pconn->wr_list_avail;
1146
1147         if (p!=NULL) {
1148                 DEBUG(10, ("ibw_alloc_send_buf#1: cmid=%p, len=%d\n", pconn->cm_id, len));
1149
1150                 DLIST_REMOVE(pconn->wr_list_avail, p);
1151                 DLIST_ADD(pconn->wr_list_used, p);
1152
1153                 if (len <= pctx->opts.recv_bufsize) {
1154                         *buf = (void *)p->buf;
1155                 } else {
1156                         p->buf_large = ibw_alloc_mr(pctx, pconn, len, &p->mr_large);
1157                         if (p->buf_large==NULL) {
1158                                 sprintf(ibw_lasterr, "ibw_alloc_mr#1 failed\n");
1159                                 goto error;
1160                         }
1161                         *buf = (void *)p->buf_large;
1162                 }
1163                 /* p->wr_id is already filled in ibw_init_memory */
1164         } else {
1165                 DEBUG(10, ("ibw_alloc_send_buf#2: cmid=%p, len=%d\n", pconn->cm_id, len));
1166                 /* not optimized */
1167                 p = pconn->extra_avail;
1168                 if (!p) {
1169                         p = pconn->extra_avail = talloc_zero(pconn, struct ibw_wr);
1170                         talloc_set_destructor(p, ibw_wr_destruct);
1171                         if (p==NULL) {
1172                                 sprintf(ibw_lasterr, "talloc_zero failed (emax: %u)\n", pconn->extra_max);
1173                                 goto error;
1174                         }
1175                         p->wr_id = pctx->opts.max_send_wr + pconn->extra_max;
1176                         pconn->extra_max++;
1177                         switch(pconn->extra_max) {
1178                                 case 1: DEBUG(2, ("warning: queue performed\n")); break;
1179                                 case 10: DEBUG(0, ("warning: queue reached 10\n")); break;
1180                                 case 100: DEBUG(0, ("warning: queue reached 100\n")); break;
1181                                 case 1000: DEBUG(0, ("warning: queue reached 1000\n")); break;
1182                                 default: break;
1183                         }
1184                 }
1185
1186                 p->buf_large = ibw_alloc_mr(pctx, pconn, len, &p->mr_large);
1187                 if (p->buf_large==NULL) {
1188                         sprintf(ibw_lasterr, "ibw_alloc_mr#2 failed\n");
1189                         goto error;
1190                 }
1191                 *buf = (void *)p->buf_large;
1192
1193                 DLIST_REMOVE(pconn->extra_avail, p);
1194                 /* we don't have prepared index for this, so that
1195                  * we will have to find this by wr_id later on */
1196                 DLIST_ADD(pconn->extra_sent, p);
1197         }
1198
1199         *key = (void *)p;
1200
1201         return 0;
1202 error:
1203         DEBUG(0, ("ibw_alloc_send_buf error: %s", ibw_lasterr));
1204         return -1;
1205 }
1206
1207
1208 static int ibw_send_packet(struct ibw_conn *conn, void *buf, struct ibw_wr *p, uint32_t len)
1209 {
1210         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
1211         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
1212         int     rc;
1213
1214         /* can we send it right now? */
1215         if (pconn->wr_sent<pctx->opts.max_send_wr) {
1216                 struct ibv_send_wr *bad_wr;
1217                 struct ibv_sge list = {
1218                         .addr   = (uintptr_t)buf,
1219                         .length = len,
1220                         .lkey   = pconn->mr_send->lkey
1221                 };
1222                 struct ibv_send_wr wr = {
1223                         .wr_id      = p->wr_id + pctx->opts.max_recv_wr,
1224                         .sg_list    = &list,
1225                         .num_sge    = 1,
1226                         .opcode     = IBV_WR_SEND,
1227                         .send_flags = IBV_SEND_SIGNALED,
1228                 };
1229
1230                 if (p->buf_large==NULL) {
1231                         DEBUG(10, ("ibw_send#normal(cmid: %p, wrid: %u, n: %d)\n",
1232                                 pconn->cm_id, (uint32_t)wr.wr_id, len));
1233                 } else {
1234                         DEBUG(10, ("ibw_send#large(cmid: %p, wrid: %u, n: %d)\n",
1235                                 pconn->cm_id, (uint32_t)wr.wr_id, len));
1236                         list.lkey = p->mr_large->lkey;
1237                 }
1238
1239                 rc = ibv_post_send(pconn->cm_id->qp, &wr, &bad_wr);
1240                 if (rc) {
1241                         sprintf(ibw_lasterr, "ibv_post_send error %d (%d)\n",
1242                                 rc, pconn->wr_sent);
1243                         goto error;
1244                 }
1245
1246                 pconn->wr_sent++;
1247
1248                 return rc;
1249         } /* else put the request into our own queue: */
1250
1251         DEBUG(10, ("ibw_send#queued(cmid: %p, len: %u)\n", pconn->cm_id, len));
1252
1253         /* TODO: clarify how to continue when state==IBWC_STOPPED */
1254
1255         /* to be sent by ibw_wc_send */
1256         /* regardless "normal" or [a part of] "large" packet */
1257         if (!p->queued_ref_cnt) {
1258                 DLIST_ADD_END2(pconn->queue, p, struct ibw_wr *,
1259                         qprev, qnext); /* TODO: optimize */
1260                 p->queued_msg = buf;
1261         }
1262         p->queued_ref_cnt++;
1263         p->queued_rlen = len; /* last wins; see ibw_wc_send */
1264
1265         return 0;
1266 error:
1267         DEBUG(0, (ibw_lasterr));
1268         return -1;
1269 }
1270
1271 int ibw_send(struct ibw_conn *conn, void *buf, void *key, uint32_t len)
1272 {
1273         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
1274         struct ibw_wr *p = talloc_get_type(key, struct ibw_wr);
1275         int     rc;
1276
1277         assert(len>=sizeof(uint32_t));
1278         assert((*((uint32_t *)buf)==len)); /* TODO: htonl */
1279
1280         if (len > pctx->opts.recv_bufsize) {
1281                 struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
1282                 int     rlen = len;
1283                 char    *packet = (char *)buf;
1284                 uint32_t        recv_bufsize = pctx->opts.recv_bufsize;
1285
1286                 DEBUG(10, ("ibw_send#frag(cmid: %p, buf: %p, len: %u)\n",
1287                         pconn->cm_id, buf, len));
1288
1289                 /* single threaded => no race here: */
1290                 assert(p->ref_cnt==0);
1291                 while(rlen > recv_bufsize) {
1292                         rc = ibw_send_packet(conn, packet, p, recv_bufsize);
1293                         if (rc)
1294                                 return rc;
1295                         packet += recv_bufsize;
1296                         rlen -= recv_bufsize;
1297                         p->ref_cnt++; /* not good to have it in ibw_send_packet */
1298                 }
1299                 if (rlen) {
1300                         rc = ibw_send_packet(conn, packet, p, rlen);
1301                         p->ref_cnt++; /* not good to have it in ibw_send_packet */
1302                 }
1303                 p->ref_cnt--; /* for the same handling */
1304         } else {
1305                 assert(p->ref_cnt==0);
1306                 assert(p->queued_ref_cnt==0);
1307
1308                 rc = ibw_send_packet(conn, buf, p, len);
1309         }
1310         return rc;
1311 }
1312
1313 int ibw_cancel_send_buf(struct ibw_conn *conn, void *buf, void *key)
1314 {
1315         struct ibw_ctx_priv *pctx = talloc_get_type(conn->ctx->internal, struct ibw_ctx_priv);
1316         struct ibw_conn_priv *pconn = talloc_get_type(conn->internal, struct ibw_conn_priv);
1317         struct ibw_wr *p = talloc_get_type(key, struct ibw_wr);
1318
1319         assert(p!=NULL);
1320         assert(buf!=NULL);
1321         assert(conn!=NULL);
1322
1323         if (p->buf_large!=NULL)
1324                 ibw_free_mr(&p->buf_large, &p->mr_large);
1325
1326         /* parallel case */
1327         if (p->wr_id < pctx->opts.max_send_wr) {
1328                 DEBUG(10, ("ibw_cancel_send_buf#1 %u", (int)p->wr_id));
1329                 DLIST_REMOVE(pconn->wr_list_used, p);
1330                 DLIST_ADD(pconn->wr_list_avail, p);
1331         } else { /* "extra" packet */
1332                 DEBUG(10, ("ibw_cancel_send_buf#2 %u", (int)p->wr_id));
1333                 DLIST_REMOVE(pconn->extra_sent, p);
1334                 DLIST_ADD(pconn->extra_avail, p);
1335         }
1336
1337         return 0;
1338 }
1339
1340 const char *ibw_getLastError(void)
1341 {
1342         return ibw_lasterr;
1343 }