engines/rdma.c

   1 /*
   2  * RDMA I/O engine
   3  *
   4  * RDMA I/O engine based on the IB verbs and RDMA/CM user space libraries.
   5  * Supports both RDMA memory semantics and channel semantics
   6  *   for the InfiniBand, RoCE and iWARP protocols.
   7  *
   8  * This I/O engine is disabled by default. To enable it, execute:
   9  *
  10  * $ export EXTFLAGS+=" -DFIO_HAVE_RDMA "
  11  * $ export EXTLIBS+=" -libverbs -lrdmacm "
  12  *
  13  * before running make. You will need the Linux RDMA software as well, either
  14  * from your Linux distributor or directly from openfabrics.org:
  15  *
  16  * http://www.openfabrics.org/downloads/OFED/
  17  *
  18  * Exchanging steps of RDMA ioengine control messages:
  19  *      1. client side sends test mode (RDMA_WRITE/RDMA_READ/SEND)
  20  *         to server side.
  21  *      2. server side parses test mode, and sends back confirmation
  22  *         to client side. In RDMA WRITE/READ test, this confirmation
  23  *         includes memory information, such as rkey, address.
  24  *      3. client side initiates test loop.
  25  *      4. In RDMA WRITE/READ test, client side sends a completion
  26  *         notification to server side. Server side updates its
  27  *         td->done as true.
  28  *
  29  */
  30 #include <stdio.h>
  31 #include <stdlib.h>
  32 #include <unistd.h>
  33 #include <errno.h>
  34 #include <assert.h>
  35 #include <netinet/in.h>
  36 #include <arpa/inet.h>
  37 #include <netdb.h>
  38 #include <sys/poll.h>
  39 #include <sys/types.h>
  40 #include <sys/socket.h>
  41 #include <sys/time.h>
  42 #include <sys/resource.h>
  43
  44 #include <byteswap.h>
  45 #include <pthread.h>
  46 #include <inttypes.h>
  47
  48 #include "../fio.h"
  49 #include "../hash.h"
  50
  51 #include <rdma/rdma_cma.h>
  52 #include <infiniband/arch.h>
  53
  54 #define FIO_RDMA_MAX_IO_DEPTH    512
  55
  56 enum rdma_io_mode {
  57         FIO_RDMA_UNKNOWN = 0,
  58         FIO_RDMA_MEM_WRITE,
  59         FIO_RDMA_MEM_READ,
  60         FIO_RDMA_CHA_SEND,
  61         FIO_RDMA_CHA_RECV
  62 };
  63
  64 struct remote_u {
  65         uint64_t buf;
  66         uint32_t rkey;
  67         uint32_t size;
  68 };
  69
  70 struct rdma_info_blk {
  71         uint32_t mode;          /* channel semantic or memory semantic */
  72         uint32_t nr;            /* client: io depth
  73                                    server: number of records for memory semantic
  74                                  */
  75         struct remote_u rmt_us[FIO_RDMA_MAX_IO_DEPTH];
  76 };
  77
  78 struct rdma_io_u_data {
  79         uint64_t wr_id;
  80         struct ibv_send_wr sq_wr;
  81         struct ibv_recv_wr rq_wr;
  82         struct ibv_sge rdma_sgl;
  83 };
  84
  85 struct rdmaio_data {
  86         int is_client;
  87         enum rdma_io_mode rdma_protocol;
  88         char host[64];
  89         struct sockaddr_in addr;
  90
  91         struct ibv_recv_wr rq_wr;
  92         struct ibv_sge recv_sgl;
  93         struct rdma_info_blk recv_buf;
  94         struct ibv_mr *recv_mr;
  95
  96         struct ibv_send_wr sq_wr;
  97         struct ibv_sge send_sgl;
  98         struct rdma_info_blk send_buf;
  99         struct ibv_mr *send_mr;
 100
 101         struct ibv_comp_channel *channel;
 102         struct ibv_cq *cq;
 103         struct ibv_pd *pd;
 104         struct ibv_qp *qp;
 105
 106         pthread_t cmthread;
 107         struct rdma_event_channel *cm_channel;
 108         struct rdma_cm_id *cm_id;
 109         struct rdma_cm_id *child_cm_id;
 110
 111         int cq_event_num;
 112
 113         struct remote_u *rmt_us;
 114         int rmt_nr;
 115         struct io_u **io_us_queued;
 116         int io_u_queued_nr;
 117         struct io_u **io_us_flight;
 118         int io_u_flight_nr;
 119         struct io_u **io_us_completed;
 120         int io_u_completed_nr;
 121
 122         struct frand_state rand_state;
 123 };
 124
 125 static int client_recv(struct thread_data *td, struct ibv_wc *wc)
 126 {
 127         struct rdmaio_data *rd = td->io_ops->data;
 128
 129         if (wc->byte_len != sizeof(rd->recv_buf)) {
 130                 log_err("Received bogus data, size %d\n", wc->byte_len);
 131                 return 1;
 132         }
 133
 134         /* store mr info for MEMORY semantic */
 135         if ((rd->rdma_protocol == FIO_RDMA_MEM_WRITE) ||
 136             (rd->rdma_protocol == FIO_RDMA_MEM_READ)) {
 137                 /* struct flist_head *entry; */
 138                 int i = 0;
 139
 140                 rd->rmt_nr = ntohl(rd->recv_buf.nr);
 141
 142                 for (i = 0; i < rd->rmt_nr; i++) {
 143                         rd->rmt_us[i].buf = ntohll(rd->recv_buf.rmt_us[i].buf);
 144                         rd->rmt_us[i].rkey = ntohl(rd->recv_buf.rmt_us[i].rkey);
 145                         rd->rmt_us[i].size = ntohl(rd->recv_buf.rmt_us[i].size);
 146
 147                         dprint(FD_IO,
 148                                "fio: Received rkey %x addr %" PRIx64
 149                                " len %d from peer\n", rd->rmt_us[i].rkey,
 150                                rd->rmt_us[i].buf, rd->rmt_us[i].size);
 151                 }
 152         }
 153
 154         return 0;
 155 }
 156
 157 static int server_recv(struct thread_data *td, struct ibv_wc *wc)
 158 {
 159         struct rdmaio_data *rd = td->io_ops->data;
 160
 161         if (wc->wr_id == FIO_RDMA_MAX_IO_DEPTH) {
 162                 rd->rdma_protocol = ntohl(rd->recv_buf.mode);
 163
 164                 /* CHANNEL semantic, do nothing */
 165                 if (rd->rdma_protocol == FIO_RDMA_CHA_SEND)
 166                         rd->rdma_protocol = FIO_RDMA_CHA_RECV;
 167         }
 168
 169         return 0;
 170 }
 171
 172 static int cq_event_handler(struct thread_data *td, enum ibv_wc_opcode opcode)
 173 {
 174         struct rdmaio_data *rd = td->io_ops->data;
 175         struct ibv_wc wc;
 176         struct rdma_io_u_data *r_io_u_d;
 177         int ret;
 178         int compevnum = 0;
 179         int i;
 180
 181         while ((ret = ibv_poll_cq(rd->cq, 1, &wc)) == 1) {
 182                 ret = 0;
 183                 compevnum++;
 184
 185                 if (wc.status) {
 186                         log_err("fio: cq completion status %d(%s)\n",
 187                                 wc.status, ibv_wc_status_str(wc.status));
 188                         return -1;
 189                 }
 190
 191                 switch (wc.opcode) {
 192
 193                 case IBV_WC_RECV:
 194                         if (rd->is_client == 1)
 195                                 client_recv(td, &wc);
 196                         else
 197                                 server_recv(td, &wc);
 198
 199                         if (wc.wr_id == FIO_RDMA_MAX_IO_DEPTH)
 200                                 break;
 201
 202                         for (i = 0; i < rd->io_u_flight_nr; i++) {
 203                                 r_io_u_d = rd->io_us_flight[i]->engine_data;
 204
 205                                 if (wc.wr_id == r_io_u_d->rq_wr.wr_id) {
 206                                         rd->io_us_flight[i]->resid =
 207                                             rd->io_us_flight[i]->buflen
 208                                             - wc.byte_len;
 209
 210                                         rd->io_us_flight[i]->error = 0;
 211
 212                                         rd->io_us_completed[rd->
 213                                                             io_u_completed_nr]
 214                                             = rd->io_us_flight[i];
 215                                         rd->io_u_completed_nr++;
 216                                         break;
 217                                 }
 218                         }
 219                         if (i == rd->io_u_flight_nr)
 220                                 log_err("fio: recv wr %" PRId64 " not found\n",
 221                                         wc.wr_id);
 222                         else {
 223                                 /* put the last one into middle of the list */
 224                                 rd->io_us_flight[i] =
 225                                     rd->io_us_flight[rd->io_u_flight_nr - 1];
 226                                 rd->io_u_flight_nr--;
 227                         }
 228
 229                         break;
 230
 231                 case IBV_WC_SEND:
 232                 case IBV_WC_RDMA_WRITE:
 233                 case IBV_WC_RDMA_READ:
 234                         if (wc.wr_id == FIO_RDMA_MAX_IO_DEPTH)
 235                                 break;
 236
 237                         for (i = 0; i < rd->io_u_flight_nr; i++) {
 238                                 r_io_u_d = rd->io_us_flight[i]->engine_data;
 239
 240                                 if (wc.wr_id == r_io_u_d->sq_wr.wr_id) {
 241                                         rd->io_us_completed[rd->
 242                                                             io_u_completed_nr]
 243                                             = rd->io_us_flight[i];
 244                                         rd->io_u_completed_nr++;
 245                                         break;
 246                                 }
 247                         }
 248                         if (i == rd->io_u_flight_nr)
 249                                 log_err("fio: send wr %" PRId64 " not found\n",
 250                                         wc.wr_id);
 251                         else {
 252                                 /* put the last one into middle of the list */
 253                                 rd->io_us_flight[i] =
 254                                     rd->io_us_flight[rd->io_u_flight_nr - 1];
 255                                 rd->io_u_flight_nr--;
 256                         }
 257
 258                         break;
 259
 260                 default:
 261                         log_info("fio: unknown completion event %d\n",
 262                                  wc.opcode);
 263                         return -1;
 264                 }
 265                 rd->cq_event_num++;
 266         }
 267         if (ret) {
 268                 log_err("fio: poll error %d\n", ret);
 269                 return 1;
 270         }
 271
 272         return compevnum;
 273 }
 274
 275 /*
 276  * Return -1 for error and 'nr events' for a positive number
 277  * of events
 278  */
 279 static int rdma_poll_wait(struct thread_data *td, enum ibv_wc_opcode opcode)
 280 {
 281         struct rdmaio_data *rd = td->io_ops->data;
 282         struct ibv_cq *ev_cq;
 283         void *ev_ctx;
 284         int ret;
 285
 286         if (rd->cq_event_num > 0) {     /* previous left */
 287                 rd->cq_event_num--;
 288                 return 0;
 289         }
 290
 291 again:
 292         if (ibv_get_cq_event(rd->channel, &ev_cq, &ev_ctx) != 0) {
 293                 log_err("fio: Failed to get cq event!\n");
 294                 return -1;
 295         }
 296         if (ev_cq != rd->cq) {
 297                 log_err("fio: Unknown CQ!\n");
 298                 return -1;
 299         }
 300         if (ibv_req_notify_cq(rd->cq, 0) != 0) {
 301                 log_err("fio: Failed to set notify!\n");
 302                 return -1;
 303         }
 304
 305         ret = cq_event_handler(td, opcode);
 306         if (ret < 1)
 307                 goto again;
 308
 309         ibv_ack_cq_events(rd->cq, ret);
 310
 311         rd->cq_event_num--;
 312
 313         return ret;
 314 }
 315
 316 static int fio_rdmaio_setup_qp(struct thread_data *td)
 317 {
 318         struct rdmaio_data *rd = td->io_ops->data;
 319         struct ibv_qp_init_attr init_attr;
 320         int qp_depth = td->o.iodepth * 2;       /* 2 times of io depth */
 321
 322         if (rd->is_client == 0)
 323                 rd->pd = ibv_alloc_pd(rd->child_cm_id->verbs);
 324         else
 325                 rd->pd = ibv_alloc_pd(rd->cm_id->verbs);
 326
 327         if (rd->pd == NULL) {
 328                 log_err("fio: ibv_alloc_pd fail\n");
 329                 return 1;
 330         }
 331
 332         if (rd->is_client == 0)
 333                 rd->channel = ibv_create_comp_channel(rd->child_cm_id->verbs);
 334         else
 335                 rd->channel = ibv_create_comp_channel(rd->cm_id->verbs);
 336         if (rd->channel == NULL) {
 337                 log_err("fio: ibv_create_comp_channel fail\n");
 338                 goto err1;
 339         }
 340
 341         if (qp_depth < 16)
 342                 qp_depth = 16;
 343
 344         if (rd->is_client == 0)
 345                 rd->cq = ibv_create_cq(rd->child_cm_id->verbs,
 346                                        qp_depth, rd, rd->channel, 0);
 347         else
 348                 rd->cq = ibv_create_cq(rd->cm_id->verbs,
 349                                        qp_depth, rd, rd->channel, 0);
 350         if (rd->cq == NULL) {
 351                 log_err("fio: ibv_create_cq failed\n");
 352                 goto err2;
 353         }
 354
 355         if (ibv_req_notify_cq(rd->cq, 0) != 0) {
 356                 log_err("fio: ibv_create_cq failed\n");
 357                 goto err3;
 358         }
 359
 360         /* create queue pair */
 361         memset(&init_attr, 0, sizeof(init_attr));
 362         init_attr.cap.max_send_wr = qp_depth;
 363         init_attr.cap.max_recv_wr = qp_depth;
 364         init_attr.cap.max_recv_sge = 1;
 365         init_attr.cap.max_send_sge = 1;
 366         init_attr.qp_type = IBV_QPT_RC;
 367         init_attr.send_cq = rd->cq;
 368         init_attr.recv_cq = rd->cq;
 369
 370         if (rd->is_client == 0) {
 371                 if (rdma_create_qp(rd->child_cm_id, rd->pd, &init_attr) != 0) {
 372                         log_err("fio: rdma_create_qp failed\n");
 373                         goto err3;
 374                 }
 375                 rd->qp = rd->child_cm_id->qp;
 376         } else {
 377                 if (rdma_create_qp(rd->cm_id, rd->pd, &init_attr) != 0) {
 378                         log_err("fio: rdma_create_qp failed\n");
 379                         goto err3;
 380                 }
 381                 rd->qp = rd->cm_id->qp;
 382         }
 383
 384         return 0;
 385
 386 err3:
 387         ibv_destroy_cq(rd->cq);
 388 err2:
 389         ibv_destroy_comp_channel(rd->channel);
 390 err1:
 391         ibv_dealloc_pd(rd->pd);
 392
 393         return 1;
 394 }
 395
 396 static int fio_rdmaio_setup_control_msg_buffers(struct thread_data *td)
 397 {
 398         struct rdmaio_data *rd = td->io_ops->data;
 399
 400         rd->recv_mr = ibv_reg_mr(rd->pd, &rd->recv_buf, sizeof(rd->recv_buf),
 401                                  IBV_ACCESS_LOCAL_WRITE);
 402         if (rd->recv_mr == NULL) {
 403                 log_err("fio: recv_buf reg_mr failed\n");
 404                 return 1;
 405         }
 406
 407         rd->send_mr = ibv_reg_mr(rd->pd, &rd->send_buf, sizeof(rd->send_buf),
 408                                  0);
 409         if (rd->send_mr == NULL) {
 410                 log_err("fio: send_buf reg_mr failed\n");
 411                 ibv_dereg_mr(rd->recv_mr);
 412                 return 1;
 413         }
 414
 415         /* setup work request */
 416         /* recv wq */
 417         rd->recv_sgl.addr = (uint64_t) (unsigned long)&rd->recv_buf;
 418         rd->recv_sgl.length = sizeof(rd->recv_buf);
 419         rd->recv_sgl.lkey = rd->recv_mr->lkey;
 420         rd->rq_wr.sg_list = &rd->recv_sgl;
 421         rd->rq_wr.num_sge = 1;
 422         rd->rq_wr.wr_id = FIO_RDMA_MAX_IO_DEPTH;
 423
 424         /* send wq */
 425         rd->send_sgl.addr = (uint64_t) (unsigned long)&rd->send_buf;
 426         rd->send_sgl.length = sizeof(rd->send_buf);
 427         rd->send_sgl.lkey = rd->send_mr->lkey;
 428
 429         rd->sq_wr.opcode = IBV_WR_SEND;
 430         rd->sq_wr.send_flags = IBV_SEND_SIGNALED;
 431         rd->sq_wr.sg_list = &rd->send_sgl;
 432         rd->sq_wr.num_sge = 1;
 433         rd->sq_wr.wr_id = FIO_RDMA_MAX_IO_DEPTH;
 434
 435         return 0;
 436 }
 437
 438 static int get_next_channel_event(struct thread_data *td,
 439                                   struct rdma_event_channel *channel,
 440                                   enum rdma_cm_event_type wait_event)
 441 {
 442         struct rdmaio_data *rd = td->io_ops->data;
 443         struct rdma_cm_event *event;
 444         int ret;
 445
 446         ret = rdma_get_cm_event(channel, &event);
 447         if (ret) {
 448                 log_err("fio: rdma_get_cm_event: %d\n", ret);
 449                 return 1;
 450         }
 451
 452         if (event->event != wait_event) {
 453                 log_err("fio: event is %s instead of %s\n",
 454                         rdma_event_str(event->event),
 455                         rdma_event_str(wait_event));
 456                 return 1;
 457         }
 458
 459         switch (event->event) {
 460         case RDMA_CM_EVENT_CONNECT_REQUEST:
 461                 rd->child_cm_id = event->id;
 462                 break;
 463         default:
 464                 break;
 465         }
 466
 467         rdma_ack_cm_event(event);
 468
 469         return 0;
 470 }
 471
 472 static int fio_rdmaio_prep(struct thread_data *td, struct io_u *io_u)
 473 {
 474         struct rdmaio_data *rd = td->io_ops->data;
 475         struct rdma_io_u_data *r_io_u_d;
 476
 477         r_io_u_d = io_u->engine_data;
 478
 479         switch (rd->rdma_protocol) {
 480         case FIO_RDMA_MEM_WRITE:
 481         case FIO_RDMA_MEM_READ:
 482                 r_io_u_d->rdma_sgl.addr = (uint64_t) (unsigned long)io_u->buf;
 483                 r_io_u_d->rdma_sgl.lkey = io_u->mr->lkey;
 484                 r_io_u_d->sq_wr.wr_id = r_io_u_d->wr_id;
 485                 r_io_u_d->sq_wr.send_flags = IBV_SEND_SIGNALED;
 486                 r_io_u_d->sq_wr.sg_list = &r_io_u_d->rdma_sgl;
 487                 r_io_u_d->sq_wr.num_sge = 1;
 488                 break;
 489         case FIO_RDMA_CHA_SEND:
 490                 r_io_u_d->rdma_sgl.addr = (uint64_t) (unsigned long)io_u->buf;
 491                 r_io_u_d->rdma_sgl.lkey = io_u->mr->lkey;
 492                 r_io_u_d->rdma_sgl.length = io_u->buflen;
 493                 r_io_u_d->sq_wr.wr_id = r_io_u_d->wr_id;
 494                 r_io_u_d->sq_wr.opcode = IBV_WR_SEND;
 495                 r_io_u_d->sq_wr.send_flags = IBV_SEND_SIGNALED;
 496                 r_io_u_d->sq_wr.sg_list = &r_io_u_d->rdma_sgl;
 497                 r_io_u_d->sq_wr.num_sge = 1;
 498                 break;
 499         case FIO_RDMA_CHA_RECV:
 500                 r_io_u_d->rdma_sgl.addr = (uint64_t) (unsigned long)io_u->buf;
 501                 r_io_u_d->rdma_sgl.lkey = io_u->mr->lkey;
 502                 r_io_u_d->rdma_sgl.length = io_u->buflen;
 503                 r_io_u_d->rq_wr.wr_id = r_io_u_d->wr_id;
 504                 r_io_u_d->rq_wr.sg_list = &r_io_u_d->rdma_sgl;
 505                 r_io_u_d->rq_wr.num_sge = 1;
 506                 break;
 507         default:
 508                 log_err("fio: unknown rdma protocol - %d\n", rd->rdma_protocol);
 509                 break;
 510         }
 511
 512         return 0;
 513 }
 514
 515 static struct io_u *fio_rdmaio_event(struct thread_data *td, int event)
 516 {
 517         struct rdmaio_data *rd = td->io_ops->data;
 518         struct io_u *io_u;
 519         int i;
 520
 521         io_u = rd->io_us_completed[0];
 522         for (i = 0; i < rd->io_u_completed_nr - 1; i++)
 523                 rd->io_us_completed[i] = rd->io_us_completed[i + 1];
 524
 525         rd->io_u_completed_nr--;
 526
 527         dprint_io_u(io_u, "fio_rdmaio_event");
 528
 529         return io_u;
 530 }
 531
 532 static int fio_rdmaio_getevents(struct thread_data *td, unsigned int min,
 533                                 unsigned int max, struct timespec *t)
 534 {
 535         struct rdmaio_data *rd = td->io_ops->data;
 536         enum ibv_wc_opcode comp_opcode;
 537         comp_opcode = IBV_WC_RDMA_WRITE;
 538         struct ibv_cq *ev_cq;
 539         void *ev_ctx;
 540         int ret, r = 0;
 541
 542         switch (rd->rdma_protocol) {
 543         case FIO_RDMA_MEM_WRITE:
 544                 comp_opcode = IBV_WC_RDMA_WRITE;
 545                 break;
 546         case FIO_RDMA_MEM_READ:
 547                 comp_opcode = IBV_WC_RDMA_READ;
 548                 break;
 549         case FIO_RDMA_CHA_SEND:
 550                 comp_opcode = IBV_WC_SEND;
 551                 break;
 552         case FIO_RDMA_CHA_RECV:
 553                 comp_opcode = IBV_WC_RECV;
 554                 break;
 555         default:
 556                 log_err("fio: unknown rdma protocol - %d\n", rd->rdma_protocol);
 557                 break;
 558         }
 559
 560         if (rd->cq_event_num > 0) {     /* previous left */
 561                 rd->cq_event_num--;
 562                 return 0;
 563         }
 564
 565 again:
 566         if (ibv_get_cq_event(rd->channel, &ev_cq, &ev_ctx) != 0) {
 567                 log_err("fio: Failed to get cq event!\n");
 568                 return -1;
 569         }
 570         if (ev_cq != rd->cq) {
 571                 log_err("fio: Unknown CQ!\n");
 572                 return -1;
 573         }
 574         if (ibv_req_notify_cq(rd->cq, 0) != 0) {
 575                 log_err("fio: Failed to set notify!\n");
 576                 return -1;
 577         }
 578
 579         ret = cq_event_handler(td, comp_opcode);
 580         if (ret < 1)
 581                 goto again;
 582
 583         ibv_ack_cq_events(rd->cq, ret);
 584
 585         r += ret;
 586         if (r < min)
 587                 goto again;
 588
 589         rd->cq_event_num -= r;
 590
 591         return r;
 592 }
 593
 594 static int fio_rdmaio_send(struct thread_data *td, struct io_u **io_us,
 595                            unsigned int nr)
 596 {
 597         struct rdmaio_data *rd = td->io_ops->data;
 598         struct ibv_send_wr *bad_wr;
 599 #if 0
 600         enum ibv_wc_opcode comp_opcode;
 601         comp_opcode = IBV_WC_RDMA_WRITE;
 602 #endif
 603         int i;
 604         long index;
 605         struct rdma_io_u_data *r_io_u_d;
 606
 607         r_io_u_d = NULL;
 608
 609         for (i = 0; i < nr; i++) {
 610                 /* RDMA_WRITE or RDMA_READ */
 611                 switch (rd->rdma_protocol) {
 612                 case FIO_RDMA_MEM_WRITE:
 613                         /* compose work request */
 614                         r_io_u_d = io_us[i]->engine_data;
 615                         index = __rand(&rd->rand_state) % rd->rmt_nr;
 616                         r_io_u_d->sq_wr.opcode = IBV_WR_RDMA_WRITE;
 617                         r_io_u_d->sq_wr.wr.rdma.rkey = rd->rmt_us[index].rkey;
 618                         r_io_u_d->sq_wr.wr.rdma.remote_addr = \
 619                                 rd->rmt_us[index].buf;
 620                         r_io_u_d->sq_wr.sg_list->length = io_us[i]->buflen;
 621                         break;
 622                 case FIO_RDMA_MEM_READ:
 623                         /* compose work request */
 624                         r_io_u_d = io_us[i]->engine_data;
 625                         index = __rand(&rd->rand_state) % rd->rmt_nr;
 626                         r_io_u_d->sq_wr.opcode = IBV_WR_RDMA_READ;
 627                         r_io_u_d->sq_wr.wr.rdma.rkey = rd->rmt_us[index].rkey;
 628                         r_io_u_d->sq_wr.wr.rdma.remote_addr = \
 629                                 rd->rmt_us[index].buf;
 630                         r_io_u_d->sq_wr.sg_list->length = io_us[i]->buflen;
 631                         break;
 632                 case FIO_RDMA_CHA_SEND:
 633                         r_io_u_d = io_us[i]->engine_data;
 634                         r_io_u_d->sq_wr.opcode = IBV_WR_SEND;
 635                         r_io_u_d->sq_wr.send_flags = IBV_SEND_SIGNALED;
 636                         break;
 637                 default:
 638                         log_err("fio: unknown rdma protocol - %d\n",
 639                                 rd->rdma_protocol);
 640                         break;
 641                 }
 642
 643                 if (ibv_post_send(rd->qp, &r_io_u_d->sq_wr, &bad_wr) != 0) {
 644                         log_err("fio: ibv_post_send fail\n");
 645                         return -1;
 646                 }
 647
 648                 dprint_io_u(io_us[i], "fio_rdmaio_send");
 649         }
 650
 651         /* wait for completion
 652            rdma_poll_wait(td, comp_opcode); */
 653
 654         return i;
 655 }
 656
 657 static int fio_rdmaio_recv(struct thread_data *td, struct io_u **io_us,
 658                            unsigned int nr)
 659 {
 660         struct rdmaio_data *rd = td->io_ops->data;
 661         struct ibv_recv_wr *bad_wr;
 662         struct rdma_io_u_data *r_io_u_d;
 663         int i;
 664
 665         i = 0;
 666         if (rd->rdma_protocol == FIO_RDMA_CHA_RECV) {
 667                 /* post io_u into recv queue */
 668                 for (i = 0; i < nr; i++) {
 669                         r_io_u_d = io_us[i]->engine_data;
 670                         if (ibv_post_recv(rd->qp, &r_io_u_d->rq_wr, &bad_wr) !=
 671                             0) {
 672                                 log_err("fio: ibv_post_recv fail\n");
 673                                 return 1;
 674                         }
 675                 }
 676         } else if ((rd->rdma_protocol == FIO_RDMA_MEM_READ)
 677                    || (rd->rdma_protocol == FIO_RDMA_MEM_WRITE)) {
 678                 /* re-post the rq_wr */
 679                 if (ibv_post_recv(rd->qp, &rd->rq_wr, &bad_wr) != 0) {
 680                         log_err("fio: ibv_post_recv fail\n");
 681                         return 1;
 682                 }
 683
 684                 rdma_poll_wait(td, IBV_WC_RECV);
 685
 686                 dprint(FD_IO, "fio: recv FINISH message\n");
 687                 td->done = 1;
 688                 return 0;
 689         }
 690
 691         return i;
 692 }
 693
 694 static int fio_rdmaio_queue(struct thread_data *td, struct io_u *io_u)
 695 {
 696         struct rdmaio_data *rd = td->io_ops->data;
 697
 698         fio_ro_check(td, io_u);
 699
 700         if (rd->io_u_queued_nr == (int)td->o.iodepth)
 701                 return FIO_Q_BUSY;
 702
 703         rd->io_us_queued[rd->io_u_queued_nr] = io_u;
 704         rd->io_u_queued_nr++;
 705
 706         dprint_io_u(io_u, "fio_rdmaio_queue");
 707
 708         return FIO_Q_QUEUED;
 709 }
 710
 711 static void fio_rdmaio_queued(struct thread_data *td, struct io_u **io_us,
 712                               unsigned int nr)
 713 {
 714         struct rdmaio_data *rd = td->io_ops->data;
 715         struct timeval now;
 716         unsigned int i;
 717
 718         if (!fio_fill_issue_time(td))
 719                 return;
 720
 721         fio_gettime(&now, NULL);
 722
 723         for (i = 0; i < nr; i++) {
 724                 struct io_u *io_u = io_us[i];
 725
 726                 /* queued -> flight */
 727                 rd->io_us_flight[rd->io_u_flight_nr] = io_u;
 728                 rd->io_u_flight_nr++;
 729
 730                 memcpy(&io_u->issue_time, &now, sizeof(now));
 731                 io_u_queued(td, io_u);
 732         }
 733 }
 734
 735 static int fio_rdmaio_commit(struct thread_data *td)
 736 {
 737         struct rdmaio_data *rd = td->io_ops->data;
 738         struct io_u **io_us;
 739         int ret;
 740
 741         if (!rd->io_us_queued)
 742                 return 0;
 743
 744         io_us = rd->io_us_queued;
 745         do {
 746                 /* RDMA_WRITE or RDMA_READ */
 747                 if (rd->is_client)
 748                         ret = fio_rdmaio_send(td, io_us, rd->io_u_queued_nr);
 749                 else if (!rd->is_client)
 750                         ret = fio_rdmaio_recv(td, io_us, rd->io_u_queued_nr);
 751                 else
 752                         ret = 0;        /* must be a SYNC */
 753
 754                 if (ret > 0) {
 755                         fio_rdmaio_queued(td, io_us, ret);
 756                         io_u_mark_submit(td, ret);
 757                         rd->io_u_queued_nr -= ret;
 758                         io_us += ret;
 759                         ret = 0;
 760                 } else
 761                         break;
 762         } while (rd->io_u_queued_nr);
 763
 764         return ret;
 765 }
 766
 767 static int fio_rdmaio_connect(struct thread_data *td, struct fio_file *f)
 768 {
 769         struct rdmaio_data *rd = td->io_ops->data;
 770         struct rdma_conn_param conn_param;
 771         struct ibv_send_wr *bad_wr;
 772
 773         memset(&conn_param, 0, sizeof(conn_param));
 774         conn_param.responder_resources = 1;
 775         conn_param.initiator_depth = 1;
 776         conn_param.retry_count = 10;
 777
 778         if (rdma_connect(rd->cm_id, &conn_param) != 0) {
 779                 log_err("fio: rdma_connect fail\n");
 780                 return 1;
 781         }
 782
 783         if (get_next_channel_event
 784             (td, rd->cm_channel, RDMA_CM_EVENT_ESTABLISHED) != 0) {
 785                 log_err("fio: wait for RDMA_CM_EVENT_ESTABLISHED\n");
 786                 return 1;
 787         }
 788
 789         /* send task request */
 790         rd->send_buf.mode = htonl(rd->rdma_protocol);
 791         rd->send_buf.nr = htonl(td->o.iodepth);
 792
 793         if (ibv_post_send(rd->qp, &rd->sq_wr, &bad_wr) != 0) {
 794                 log_err("fio: ibv_post_send fail");
 795                 return 1;
 796         }
 797
 798         rdma_poll_wait(td, IBV_WC_SEND);
 799
 800         /* wait for remote MR info from server side */
 801         rdma_poll_wait(td, IBV_WC_RECV);
 802
 803         /* In SEND/RECV test, it's a good practice to setup the iodepth of
 804          * of the RECV side deeper than that of the SEND side to
 805          * avoid RNR (receiver not ready) error. The
 806          * SEND side may send so many unsolicited message before
 807          * RECV side commits sufficient recv buffers into recv queue.
 808          * This may lead to RNR error. Here, SEND side pauses for a while
 809          * during which RECV side commits sufficient recv buffers.
 810          */
 811         usleep(500000);
 812
 813         return 0;
 814 }
 815
 816 static int fio_rdmaio_accept(struct thread_data *td, struct fio_file *f)
 817 {
 818         struct rdmaio_data *rd = td->io_ops->data;
 819         struct rdma_conn_param conn_param;
 820         struct ibv_send_wr *bad_wr;
 821
 822         /* rdma_accept() - then wait for accept success */
 823         memset(&conn_param, 0, sizeof(conn_param));
 824         conn_param.responder_resources = 1;
 825         conn_param.initiator_depth = 1;
 826
 827         if (rdma_accept(rd->child_cm_id, &conn_param) != 0) {
 828                 log_err("fio: rdma_accept\n");
 829                 return 1;
 830         }
 831
 832         if (get_next_channel_event
 833             (td, rd->cm_channel, RDMA_CM_EVENT_ESTABLISHED) != 0) {
 834                 log_err("fio: wait for RDMA_CM_EVENT_ESTABLISHED\n");
 835                 return 1;
 836         }
 837
 838         /* wait for request */
 839         rdma_poll_wait(td, IBV_WC_RECV);
 840
 841         if (ibv_post_send(rd->qp, &rd->sq_wr, &bad_wr) != 0) {
 842                 log_err("fio: ibv_post_send fail");
 843                 return 1;
 844         }
 845
 846         rdma_poll_wait(td, IBV_WC_SEND);
 847
 848         return 0;
 849 }
 850
 851 static int fio_rdmaio_open_file(struct thread_data *td, struct fio_file *f)
 852 {
 853         if (td_read(td))
 854                 return fio_rdmaio_accept(td, f);
 855         else
 856                 return fio_rdmaio_connect(td, f);
 857 }
 858
 859 static int fio_rdmaio_close_file(struct thread_data *td, struct fio_file *f)
 860 {
 861         struct rdmaio_data *rd = td->io_ops->data;
 862         struct ibv_send_wr *bad_wr;
 863
 864         /* unregister rdma buffer */
 865
 866         /*
 867          * Client sends notification to the server side
 868          */
 869         /* refer to: http://linux.die.net/man/7/rdma_cm */
 870         if ((rd->is_client == 1) && ((rd->rdma_protocol == FIO_RDMA_MEM_WRITE)
 871                                      || (rd->rdma_protocol ==
 872                                          FIO_RDMA_MEM_READ))) {
 873                 if (ibv_post_send(rd->qp, &rd->sq_wr, &bad_wr) != 0) {
 874                         log_err("fio: ibv_post_send fail");
 875                         return 1;
 876                 }
 877
 878                 dprint(FD_IO, "fio: close infomation sent success\n");
 879                 rdma_poll_wait(td, IBV_WC_SEND);
 880         }
 881
 882         if (rd->is_client == 1)
 883                 rdma_disconnect(rd->cm_id);
 884         else {
 885                 rdma_disconnect(rd->child_cm_id);
 886 #if 0
 887                 rdma_disconnect(rd->cm_id);
 888 #endif
 889         }
 890
 891 #if 0
 892         if (get_next_channel_event(td, rd->cm_channel, RDMA_CM_EVENT_DISCONNECTED) != 0) {
 893                 log_err("fio: wait for RDMA_CM_EVENT_DISCONNECTED\n");
 894                 return 1;
 895         }
 896 #endif
 897
 898         ibv_destroy_cq(rd->cq);
 899         ibv_destroy_qp(rd->qp);
 900
 901         if (rd->is_client == 1)
 902                 rdma_destroy_id(rd->cm_id);
 903         else {
 904                 rdma_destroy_id(rd->child_cm_id);
 905                 rdma_destroy_id(rd->cm_id);
 906         }
 907
 908         ibv_destroy_comp_channel(rd->channel);
 909         ibv_dealloc_pd(rd->pd);
 910
 911         return 0;
 912 }
 913
 914 static int fio_rdmaio_setup_connect(struct thread_data *td, const char *host,
 915                                     unsigned short port)
 916 {
 917         struct rdmaio_data *rd = td->io_ops->data;
 918         struct ibv_recv_wr *bad_wr;
 919         int err;
 920
 921         rd->addr.sin_family = AF_INET;
 922         rd->addr.sin_port = htons(port);
 923
 924         if (inet_aton(host, &rd->addr.sin_addr) != 1) {
 925                 struct hostent *hent;
 926
 927                 hent = gethostbyname(host);
 928                 if (!hent) {
 929                         td_verror(td, errno, "gethostbyname");
 930                         return 1;
 931                 }
 932
 933                 memcpy(&rd->addr.sin_addr, hent->h_addr, 4);
 934         }
 935
 936         /* resolve route */
 937         err = rdma_resolve_addr(rd->cm_id, NULL, (struct sockaddr *)&rd->addr, 2000);
 938         if (err != 0) {
 939                 log_err("fio: rdma_resolve_addr: %d\n", err);
 940                 return 1;
 941         }
 942
 943         err = get_next_channel_event(td, rd->cm_channel, RDMA_CM_EVENT_ADDR_RESOLVED);
 944         if (err != 0) {
 945                 log_err("fio: get_next_channel_event: %d\n", err);
 946                 return 1;
 947         }
 948
 949         /* resolve route */
 950         err = rdma_resolve_route(rd->cm_id, 2000);
 951         if (err != 0) {
 952                 log_err("fio: rdma_resolve_route: %d\n", err);
 953                 return 1;
 954         }
 955
 956         err = get_next_channel_event(td, rd->cm_channel, RDMA_CM_EVENT_ROUTE_RESOLVED);
 957         if (err != 0) {
 958                 log_err("fio: get_next_channel_event: %d\n", err);
 959                 return 1;
 960         }
 961
 962         /* create qp and buffer */
 963         if (fio_rdmaio_setup_qp(td) != 0)
 964                 return 1;
 965
 966         if (fio_rdmaio_setup_control_msg_buffers(td) != 0)
 967                 return 1;
 968
 969         /* post recv buf */
 970         err = ibv_post_recv(rd->qp, &rd->rq_wr, &bad_wr);
 971         if (err != 0) {
 972                 log_err("fio: ibv_post_recv fail: %d\n", err);
 973                 return 1;
 974         }
 975
 976         return 0;
 977 }
 978
 979 static int fio_rdmaio_setup_listen(struct thread_data *td, short port)
 980 {
 981         struct rdmaio_data *rd = td->io_ops->data;
 982         struct ibv_recv_wr *bad_wr;
 983
 984         rd->addr.sin_family = AF_INET;
 985         rd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
 986         rd->addr.sin_port = htons(port);
 987
 988         /* rdma_listen */
 989         if (rdma_bind_addr(rd->cm_id, (struct sockaddr *)&rd->addr) != 0) {
 990                 log_err("fio: rdma_bind_addr fail\n");
 991                 return 1;
 992         }
 993
 994         if (rdma_listen(rd->cm_id, 3) != 0) {
 995                 log_err("fio: rdma_listen fail\n");
 996                 return 1;
 997         }
 998
 999         /* wait for CONNECT_REQUEST */
1000         if (get_next_channel_event
1001             (td, rd->cm_channel, RDMA_CM_EVENT_CONNECT_REQUEST) != 0) {
1002                 log_err("fio: wait for RDMA_CM_EVENT_CONNECT_REQUEST\n");
1003                 return 1;
1004         }
1005
1006         if (fio_rdmaio_setup_qp(td) != 0)
1007                 return 1;
1008
1009         if (fio_rdmaio_setup_control_msg_buffers(td) != 0)
1010                 return 1;
1011
1012         /* post recv buf */
1013         if (ibv_post_recv(rd->qp, &rd->rq_wr, &bad_wr) != 0) {
1014                 log_err("fio: ibv_post_recv fail\n");
1015                 return 1;
1016         }
1017
1018         return 0;
1019 }
1020
1021 static int fio_rdmaio_init(struct thread_data *td)
1022 {
1023         struct rdmaio_data *rd = td->io_ops->data;
1024         struct flist_head *entry;
1025         unsigned int max_bs;
1026         unsigned int port;
1027         char host[64], buf[128];
1028         char *sep, *portp, *modep;
1029         int ret, i = 0;
1030         struct rlimit rl;
1031
1032         if (td_rw(td)) {
1033                 log_err("fio: rdma connections must be read OR write\n");
1034                 return 1;
1035         }
1036         if (td_random(td)) {
1037                 log_err("fio: RDMA network IO can't be random\n");
1038                 return 1;
1039         }
1040
1041         /* check RLIMIT_MEMLOCK */
1042         if (getrlimit(RLIMIT_MEMLOCK, &rl) != 0) {
1043                 log_err("fio: getrlimit fail: %d(%s)\n",
1044                         errno, strerror(errno));
1045                 return 1;
1046         }
1047
1048         /* soft limit */
1049         if ((rl.rlim_cur != RLIM_INFINITY)
1050             && (rl.rlim_cur < td->orig_buffer_size)) {
1051                 log_err("fio: soft RLIMIT_MEMLOCK is: %" PRId64 "\n",
1052                         rl.rlim_cur);
1053                 log_err("fio: total block size is:    %zd\n",
1054                         td->orig_buffer_size);
1055                 /* try to set larger RLIMIT_MEMLOCK */
1056                 rl.rlim_cur = rl.rlim_max;
1057                 if (setrlimit(RLIMIT_MEMLOCK, &rl) != 0) {
1058                         log_err("fio: setrlimit fail: %d(%s)\n",
1059                                 errno, strerror(errno));
1060                         log_err("fio: you may try enlarge MEMLOCK by root\n");
1061                         log_err("# ulimit -l unlimited\n");
1062                         return 1;
1063                 }
1064         }
1065
1066         strcpy(buf, td->o.filename);
1067
1068         sep = strchr(buf, '/');
1069         if (!sep)
1070                 goto bad_host;
1071
1072         *sep = '\0';
1073         sep++;
1074         strcpy(host, buf);
1075         if (!strlen(host))
1076                 goto bad_host;
1077
1078         modep = NULL;
1079         portp = sep;
1080         sep = strchr(portp, '/');
1081         if (sep) {
1082                 *sep = '\0';
1083                 modep = sep + 1;
1084         }
1085
1086         port = strtol(portp, NULL, 10);
1087         if (!port || port > 65535)
1088                 goto bad_host;
1089
1090         if (modep) {
1091                 if (!strncmp("rdma_write", modep, strlen(modep)) ||
1092                     !strncmp("RDMA_WRITE", modep, strlen(modep)))
1093                         rd->rdma_protocol = FIO_RDMA_MEM_WRITE;
1094                 else if (!strncmp("rdma_read", modep, strlen(modep)) ||
1095                          !strncmp("RDMA_READ", modep, strlen(modep)))
1096                         rd->rdma_protocol = FIO_RDMA_MEM_READ;
1097                 else if (!strncmp("send", modep, strlen(modep)) ||
1098                          !strncmp("SEND", modep, strlen(modep)))
1099                         rd->rdma_protocol = FIO_RDMA_CHA_SEND;
1100                 else
1101                         goto bad_host;
1102         } else
1103                 rd->rdma_protocol = FIO_RDMA_MEM_WRITE;
1104
1105         rd->cq_event_num = 0;
1106
1107         rd->cm_channel = rdma_create_event_channel();
1108         if (!rd->cm_channel) {
1109                 log_err("fio: rdma_create_event_channel fail\n");
1110                 return 1;
1111         }
1112
1113         ret = rdma_create_id(rd->cm_channel, &rd->cm_id, rd, RDMA_PS_TCP);
1114         if (ret) {
1115                 log_err("fio: rdma_create_id fail\n");
1116                 return 1;
1117         }
1118
1119         if ((rd->rdma_protocol == FIO_RDMA_MEM_WRITE) ||
1120             (rd->rdma_protocol == FIO_RDMA_MEM_READ)) {
1121                 rd->rmt_us =
1122                         malloc(FIO_RDMA_MAX_IO_DEPTH * sizeof(struct remote_u));
1123                 memset(rd->rmt_us, 0,
1124                         FIO_RDMA_MAX_IO_DEPTH * sizeof(struct remote_u));
1125                 rd->rmt_nr = 0;
1126         }
1127
1128         rd->io_us_queued = malloc(td->o.iodepth * sizeof(struct io_u *));
1129         memset(rd->io_us_queued, 0, td->o.iodepth * sizeof(struct io_u *));
1130         rd->io_u_queued_nr = 0;
1131
1132         rd->io_us_flight = malloc(td->o.iodepth * sizeof(struct io_u *));
1133         memset(rd->io_us_flight, 0, td->o.iodepth * sizeof(struct io_u *));
1134         rd->io_u_flight_nr = 0;
1135
1136         rd->io_us_completed = malloc(td->o.iodepth * sizeof(struct io_u *));
1137         memset(rd->io_us_completed, 0, td->o.iodepth * sizeof(struct io_u *));
1138         rd->io_u_completed_nr = 0;
1139
1140         if (td_read(td)) {      /* READ as the server */
1141                 rd->is_client = 0;
1142                 /* server rd->rdma_buf_len will be setup after got request */
1143                 ret = fio_rdmaio_setup_listen(td, port);
1144         } else {                /* WRITE as the client */
1145                 rd->is_client = 1;
1146                 ret = fio_rdmaio_setup_connect(td, host, port);
1147         }
1148
1149         max_bs = max(td->o.max_bs[DDIR_READ], td->o.max_bs[DDIR_WRITE]);
1150         /* register each io_u in the free list */
1151         flist_for_each(entry, &td->io_u_freelist) {
1152                 struct io_u *io_u = flist_entry(entry, struct io_u, list);
1153
1154                 io_u->engine_data = malloc(sizeof(struct rdma_io_u_data));
1155                 memset(io_u->engine_data, 0, sizeof(struct rdma_io_u_data));
1156                 ((struct rdma_io_u_data *)io_u->engine_data)->wr_id = i;
1157
1158                 io_u->mr = ibv_reg_mr(rd->pd, io_u->buf, max_bs,
1159                                       IBV_ACCESS_LOCAL_WRITE |
1160                                       IBV_ACCESS_REMOTE_READ |
1161                                       IBV_ACCESS_REMOTE_WRITE);
1162                 if (io_u->mr == NULL) {
1163                         log_err("fio: ibv_reg_mr io_u failed\n");
1164                         return 1;
1165                 }
1166
1167                 rd->send_buf.rmt_us[i].buf =
1168                     htonll((uint64_t) (unsigned long)io_u->buf);
1169                 rd->send_buf.rmt_us[i].rkey = htonl(io_u->mr->rkey);
1170                 rd->send_buf.rmt_us[i].size = htonl(max_bs);
1171
1172 #if 0
1173                 log_info("fio: Send rkey %x addr %" PRIx64 " len %d to client\n", io_u->mr->rkey, io_u->buf, max_bs); */
1174 #endif
1175                 i++;
1176         }
1177
1178         rd->send_buf.nr = htonl(i);
1179
1180         return ret;
1181 bad_host:
1182         log_err("fio: bad rdma host/port/protocol: %s\n", td->o.filename);
1183         return 1;
1184 }
1185
1186 static void fio_rdmaio_cleanup(struct thread_data *td)
1187 {
1188         struct rdmaio_data *rd = td->io_ops->data;
1189
1190         if (rd)
1191                 free(rd);
1192 }
1193
1194 static int fio_rdmaio_setup(struct thread_data *td)
1195 {
1196         struct rdmaio_data *rd;
1197
1198         if (!td->io_ops->data) {
1199                 rd = malloc(sizeof(*rd));
1200
1201                 memset(rd, 0, sizeof(*rd));
1202                 init_rand_seed(&rd->rand_state, (unsigned int) GOLDEN_RATIO_PRIME);
1203                 td->io_ops->data = rd;
1204         }
1205
1206         return 0;
1207 }
1208
1209 static struct ioengine_ops ioengine_rw = {
1210         .name           = "rdma",
1211         .version        = FIO_IOOPS_VERSION,
1212         .setup          = fio_rdmaio_setup,
1213         .init           = fio_rdmaio_init,
1214         .prep           = fio_rdmaio_prep,
1215         .queue          = fio_rdmaio_queue,
1216         .commit         = fio_rdmaio_commit,
1217         .getevents      = fio_rdmaio_getevents,
1218         .event          = fio_rdmaio_event,
1219         .cleanup        = fio_rdmaio_cleanup,
1220         .open_file      = fio_rdmaio_open_file,
1221         .close_file     = fio_rdmaio_close_file,
1222         .flags          = FIO_DISKLESSIO | FIO_UNIDIR | FIO_PIPEIO,
1223 };
1224
1225 static void fio_init fio_rdmaio_register(void)
1226 {
1227         register_ioengine(&ioengine_rw);
1228 }
1229
1230 static void fio_exit fio_rdmaio_unregister(void)
1231 {
1232         unregister_ioengine(&ioengine_rw);
1233 }