1 /* 2 * Copyright (c) 2005 Ammasso, Inc. All rights reserved. 3 * Copyright (c) 2006 Open Grid Computing, Inc. All rights reserved. 4 * 5 * This software is available to you under a choice of one of two 6 * licenses. You may choose to be licensed under the terms of the GNU 7 * General Public License (GPL) Version 2, available from the file 8 * COPYING in the main directory of this source tree, or the 9 * OpenIB.org BSD license below: 10 * 11 * Redistribution and use in source and binary forms, with or 12 * without modification, are permitted provided that the following 13 * conditions are met: 14 * 15 * - Redistributions of source code must retain the above 16 * copyright notice, this list of conditions and the following 17 * disclaimer. 18 * 19 * - Redistributions in binary form must reproduce the above 20 * copyright notice, this list of conditions and the following 21 * disclaimer in the documentation and/or other materials 22 * provided with the distribution. 23 * 24 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 25 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF 26 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 27 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS 28 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN 29 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN 30 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 31 * SOFTWARE. 32 */ 33 34 #include <getopt.h> 35 #include <stdlib.h> 36 #include <string.h> 37 #include <stdio.h> 38 #include <errno.h> 39 #include <sys/types.h> 40 #include <netinet/in.h> 41 #include <sys/socket.h> 42 #include <netdb.h> 43 #include <byteswap.h> 44 #include <semaphore.h> 45 #include <arpa/inet.h> 46 #include <pthread.h> 47 #include <inttypes.h> 48 49 #include <rdma/rdma_cma.h> 50 #include <infiniband/arch.h> 51 52 static int debug = 0; 53 #define DEBUG_LOG if (debug) printf 54 55 /* 56 * rping "ping/pong" loop: 57 * client sends source rkey/addr/len 58 * server receives source rkey/add/len 59 * server rdma reads "ping" data from source 60 * server sends "go ahead" on rdma read completion 61 * client sends sink rkey/addr/len 62 * server receives sink rkey/addr/len 63 * server rdma writes "pong" data to sink 64 * server sends "go ahead" on rdma write completion 65 * <repeat loop> 66 */ 67 68 /* 69 * These states are used to signal events between the completion handler 70 * and the main client or server thread. 71 * 72 * Once CONNECTED, they cycle through RDMA_READ_ADV, RDMA_WRITE_ADV, 73 * and RDMA_WRITE_COMPLETE for each ping. 74 */ 75 enum test_state { 76 IDLE = 1, 77 CONNECT_REQUEST, 78 ADDR_RESOLVED, 79 ROUTE_RESOLVED, 80 CONNECTED, 81 RDMA_READ_ADV, 82 RDMA_READ_COMPLETE, 83 RDMA_WRITE_ADV, 84 RDMA_WRITE_COMPLETE, 85 ERROR 86 }; 87 88 struct rping_rdma_info { 89 uint64_t buf; 90 uint32_t rkey; 91 uint32_t size; 92 }; 93 94 /* 95 * Default max buffer size for IO... 96 */ 97 #define RPING_BUFSIZE 64*1024 98 #define RPING_SQ_DEPTH 16 99 100 /* Default string for print data and 101 * minimum buffer size 102 */ 103 #define _stringify( _x ) # _x 104 #define stringify( _x ) _stringify( _x ) 105 106 #define RPING_MSG_FMT "rdma-ping-%d: " 107 #define RPING_MIN_BUFSIZE sizeof(stringify(INT_MAX)) + sizeof(RPING_MSG_FMT) 108 109 /* 110 * Control block struct. 111 */ 112 struct rping_cb { 113 int server; /* 0 iff client */ 114 pthread_t cqthread; 115 pthread_t persistent_server_thread; 116 struct ibv_comp_channel *channel; 117 struct ibv_cq *cq; 118 struct ibv_pd *pd; 119 struct ibv_qp *qp; 120 121 struct ibv_recv_wr rq_wr; /* recv work request record */ 122 struct ibv_sge recv_sgl; /* recv single SGE */ 123 struct rping_rdma_info recv_buf;/* malloc'd buffer */ 124 struct ibv_mr *recv_mr; /* MR associated with this buffer */ 125 126 struct ibv_send_wr sq_wr; /* send work request record */ 127 struct ibv_sge send_sgl; 128 struct rping_rdma_info send_buf;/* single send buf */ 129 struct ibv_mr *send_mr; 130 131 struct ibv_send_wr rdma_sq_wr; /* rdma work request record */ 132 struct ibv_sge rdma_sgl; /* rdma single SGE */ 133 char *rdma_buf; /* used as rdma sink */ 134 struct ibv_mr *rdma_mr; 135 136 uint32_t remote_rkey; /* remote guys RKEY */ 137 uint64_t remote_addr; /* remote guys TO */ 138 uint32_t remote_len; /* remote guys LEN */ 139 140 char *start_buf; /* rdma read src */ 141 struct ibv_mr *start_mr; 142 143 enum test_state state; /* used for cond/signalling */ 144 sem_t sem; 145 146 struct sockaddr_storage sin; 147 uint16_t port; /* dst port in NBO */ 148 int verbose; /* verbose logging */ 149 int count; /* ping count */ 150 int size; /* ping data size */ 151 int validate; /* validate ping data */ 152 153 /* CM stuff */ 154 pthread_t cmthread; 155 struct rdma_event_channel *cm_channel; 156 struct rdma_cm_id *cm_id; /* connection on client side,*/ 157 /* listener on service side. */ 158 struct rdma_cm_id *child_cm_id; /* connection on server side */ 159 }; 160 161 static int rping_cma_event_handler(struct rdma_cm_id *cma_id, 162 struct rdma_cm_event *event) 163 { 164 int ret = 0; 165 struct rping_cb *cb = cma_id->context; 166 167 DEBUG_LOG("cma_event type %s cma_id %p (%s)\n", 168 rdma_event_str(event->event), cma_id, 169 (cma_id == cb->cm_id) ? "parent" : "child"); 170 171 switch (event->event) { 172 case RDMA_CM_EVENT_ADDR_RESOLVED: 173 cb->state = ADDR_RESOLVED; 174 ret = rdma_resolve_route(cma_id, 2000); 175 if (ret) { 176 cb->state = ERROR; 177 perror("rdma_resolve_route"); 178 sem_post(&cb->sem); 179 } 180 break; 181 182 case RDMA_CM_EVENT_ROUTE_RESOLVED: 183 cb->state = ROUTE_RESOLVED; 184 sem_post(&cb->sem); 185 break; 186 187 case RDMA_CM_EVENT_CONNECT_REQUEST: 188 cb->state = CONNECT_REQUEST; 189 cb->child_cm_id = cma_id; 190 DEBUG_LOG("child cma %p\n", cb->child_cm_id); 191 sem_post(&cb->sem); 192 break; 193 194 case RDMA_CM_EVENT_ESTABLISHED: 195 DEBUG_LOG("ESTABLISHED\n"); 196 197 /* 198 * Server will wake up when first RECV completes. 199 */ 200 if (!cb->server) { 201 cb->state = CONNECTED; 202 } 203 sem_post(&cb->sem); 204 break; 205 206 case RDMA_CM_EVENT_ADDR_ERROR: 207 case RDMA_CM_EVENT_ROUTE_ERROR: 208 case RDMA_CM_EVENT_CONNECT_ERROR: 209 case RDMA_CM_EVENT_UNREACHABLE: 210 case RDMA_CM_EVENT_REJECTED: 211 fprintf(stderr, "cma event %s, error %d\n", 212 rdma_event_str(event->event), event->status); 213 sem_post(&cb->sem); 214 ret = -1; 215 break; 216 217 case RDMA_CM_EVENT_DISCONNECTED: 218 fprintf(stderr, "%s DISCONNECT EVENT...\n", 219 cb->server ? "server" : "client"); 220 sem_post(&cb->sem); 221 break; 222 223 case RDMA_CM_EVENT_DEVICE_REMOVAL: 224 fprintf(stderr, "cma detected device removal!!!!\n"); 225 ret = -1; 226 break; 227 228 default: 229 fprintf(stderr, "unhandled event: %s, ignoring\n", 230 rdma_event_str(event->event)); 231 break; 232 } 233 234 return ret; 235 } 236 237 static int server_recv(struct rping_cb *cb, struct ibv_wc *wc) 238 { 239 if (wc->byte_len != sizeof(cb->recv_buf)) { 240 fprintf(stderr, "Received bogus data, size %d\n", wc->byte_len); 241 return -1; 242 } 243 244 cb->remote_rkey = ntohl(cb->recv_buf.rkey); 245 cb->remote_addr = ntohll(cb->recv_buf.buf); 246 cb->remote_len = ntohl(cb->recv_buf.size); 247 DEBUG_LOG("Received rkey %x addr %" PRIx64 " len %d from peer\n", 248 cb->remote_rkey, cb->remote_addr, cb->remote_len); 249 250 if (cb->state <= CONNECTED || cb->state == RDMA_WRITE_COMPLETE) 251 cb->state = RDMA_READ_ADV; 252 else 253 cb->state = RDMA_WRITE_ADV; 254 255 return 0; 256 } 257 258 static int client_recv(struct rping_cb *cb, struct ibv_wc *wc) 259 { 260 if (wc->byte_len != sizeof(cb->recv_buf)) { 261 fprintf(stderr, "Received bogus data, size %d\n", wc->byte_len); 262 return -1; 263 } 264 265 if (cb->state == RDMA_READ_ADV) 266 cb->state = RDMA_WRITE_ADV; 267 else 268 cb->state = RDMA_WRITE_COMPLETE; 269 270 return 0; 271 } 272 273 static int rping_cq_event_handler(struct rping_cb *cb) 274 { 275 struct ibv_wc wc; 276 struct ibv_recv_wr *bad_wr; 277 int ret; 278 279 while ((ret = ibv_poll_cq(cb->cq, 1, &wc)) == 1) { 280 ret = 0; 281 282 if (wc.status) { 283 fprintf(stderr, "cq completion failed status %d\n", 284 wc.status); 285 if (wc.status != IBV_WC_WR_FLUSH_ERR) 286 ret = -1; 287 goto error; 288 } 289 290 switch (wc.opcode) { 291 case IBV_WC_SEND: 292 DEBUG_LOG("send completion\n"); 293 break; 294 295 case IBV_WC_RDMA_WRITE: 296 DEBUG_LOG("rdma write completion\n"); 297 cb->state = RDMA_WRITE_COMPLETE; 298 sem_post(&cb->sem); 299 break; 300 301 case IBV_WC_RDMA_READ: 302 DEBUG_LOG("rdma read completion\n"); 303 cb->state = RDMA_READ_COMPLETE; 304 sem_post(&cb->sem); 305 break; 306 307 case IBV_WC_RECV: 308 DEBUG_LOG("recv completion\n"); 309 ret = cb->server ? server_recv(cb, &wc) : 310 client_recv(cb, &wc); 311 if (ret) { 312 fprintf(stderr, "recv wc error: %d\n", ret); 313 goto error; 314 } 315 316 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr); 317 if (ret) { 318 fprintf(stderr, "post recv error: %d\n", ret); 319 goto error; 320 } 321 sem_post(&cb->sem); 322 break; 323 324 default: 325 DEBUG_LOG("unknown!!!!! completion\n"); 326 ret = -1; 327 goto error; 328 } 329 } 330 if (ret) { 331 fprintf(stderr, "poll error %d\n", ret); 332 goto error; 333 } 334 return 0; 335 336 error: 337 cb->state = ERROR; 338 sem_post(&cb->sem); 339 return ret; 340 } 341 342 static int rping_accept(struct rping_cb *cb) 343 { 344 struct rdma_conn_param conn_param; 345 int ret; 346 347 DEBUG_LOG("accepting client connection request\n"); 348 349 memset(&conn_param, 0, sizeof conn_param); 350 conn_param.responder_resources = 1; 351 conn_param.initiator_depth = 1; 352 353 ret = rdma_accept(cb->child_cm_id, &conn_param); 354 if (ret) { 355 perror("rdma_accept"); 356 return ret; 357 } 358 359 sem_wait(&cb->sem); 360 if (cb->state == ERROR) { 361 fprintf(stderr, "wait for CONNECTED state %d\n", cb->state); 362 return -1; 363 } 364 return 0; 365 } 366 367 static void rping_setup_wr(struct rping_cb *cb) 368 { 369 cb->recv_sgl.addr = (uint64_t) (unsigned long) &cb->recv_buf; 370 cb->recv_sgl.length = sizeof cb->recv_buf; 371 cb->recv_sgl.lkey = cb->recv_mr->lkey; 372 cb->rq_wr.sg_list = &cb->recv_sgl; 373 cb->rq_wr.num_sge = 1; 374 375 cb->send_sgl.addr = (uint64_t) (unsigned long) &cb->send_buf; 376 cb->send_sgl.length = sizeof cb->send_buf; 377 cb->send_sgl.lkey = cb->send_mr->lkey; 378 379 cb->sq_wr.opcode = IBV_WR_SEND; 380 cb->sq_wr.send_flags = IBV_SEND_SIGNALED; 381 cb->sq_wr.sg_list = &cb->send_sgl; 382 cb->sq_wr.num_sge = 1; 383 384 cb->rdma_sgl.addr = (uint64_t) (unsigned long) cb->rdma_buf; 385 cb->rdma_sgl.lkey = cb->rdma_mr->lkey; 386 cb->rdma_sq_wr.send_flags = IBV_SEND_SIGNALED; 387 cb->rdma_sq_wr.sg_list = &cb->rdma_sgl; 388 cb->rdma_sq_wr.num_sge = 1; 389 } 390 391 static int rping_setup_buffers(struct rping_cb *cb) 392 { 393 int ret; 394 395 DEBUG_LOG("rping_setup_buffers called on cb %p\n", cb); 396 397 cb->recv_mr = ibv_reg_mr(cb->pd, &cb->recv_buf, sizeof cb->recv_buf, 398 IBV_ACCESS_LOCAL_WRITE); 399 if (!cb->recv_mr) { 400 fprintf(stderr, "recv_buf reg_mr failed\n"); 401 return errno; 402 } 403 404 cb->send_mr = ibv_reg_mr(cb->pd, &cb->send_buf, sizeof cb->send_buf, 0); 405 if (!cb->send_mr) { 406 fprintf(stderr, "send_buf reg_mr failed\n"); 407 ret = errno; 408 goto err1; 409 } 410 411 cb->rdma_buf = malloc(cb->size); 412 if (!cb->rdma_buf) { 413 fprintf(stderr, "rdma_buf malloc failed\n"); 414 ret = -ENOMEM; 415 goto err2; 416 } 417 418 cb->rdma_mr = ibv_reg_mr(cb->pd, cb->rdma_buf, cb->size, 419 IBV_ACCESS_LOCAL_WRITE | 420 IBV_ACCESS_REMOTE_READ | 421 IBV_ACCESS_REMOTE_WRITE); 422 if (!cb->rdma_mr) { 423 fprintf(stderr, "rdma_buf reg_mr failed\n"); 424 ret = errno; 425 goto err3; 426 } 427 428 if (!cb->server) { 429 cb->start_buf = malloc(cb->size); 430 if (!cb->start_buf) { 431 fprintf(stderr, "start_buf malloc failed\n"); 432 ret = -ENOMEM; 433 goto err4; 434 } 435 436 cb->start_mr = ibv_reg_mr(cb->pd, cb->start_buf, cb->size, 437 IBV_ACCESS_LOCAL_WRITE | 438 IBV_ACCESS_REMOTE_READ | 439 IBV_ACCESS_REMOTE_WRITE); 440 if (!cb->start_mr) { 441 fprintf(stderr, "start_buf reg_mr failed\n"); 442 ret = errno; 443 goto err5; 444 } 445 } 446 447 rping_setup_wr(cb); 448 DEBUG_LOG("allocated & registered buffers...\n"); 449 return 0; 450 451 err5: 452 free(cb->start_buf); 453 err4: 454 ibv_dereg_mr(cb->rdma_mr); 455 err3: 456 free(cb->rdma_buf); 457 err2: 458 ibv_dereg_mr(cb->send_mr); 459 err1: 460 ibv_dereg_mr(cb->recv_mr); 461 return ret; 462 } 463 464 static void rping_free_buffers(struct rping_cb *cb) 465 { 466 DEBUG_LOG("rping_free_buffers called on cb %p\n", cb); 467 ibv_dereg_mr(cb->recv_mr); 468 ibv_dereg_mr(cb->send_mr); 469 ibv_dereg_mr(cb->rdma_mr); 470 free(cb->rdma_buf); 471 if (!cb->server) { 472 ibv_dereg_mr(cb->start_mr); 473 free(cb->start_buf); 474 } 475 } 476 477 static int rping_create_qp(struct rping_cb *cb) 478 { 479 struct ibv_qp_init_attr init_attr; 480 int ret; 481 482 memset(&init_attr, 0, sizeof(init_attr)); 483 init_attr.cap.max_send_wr = RPING_SQ_DEPTH; 484 init_attr.cap.max_recv_wr = 2; 485 init_attr.cap.max_recv_sge = 1; 486 init_attr.cap.max_send_sge = 1; 487 init_attr.qp_type = IBV_QPT_RC; 488 init_attr.send_cq = cb->cq; 489 init_attr.recv_cq = cb->cq; 490 491 if (cb->server) { 492 ret = rdma_create_qp(cb->child_cm_id, cb->pd, &init_attr); 493 if (!ret) 494 cb->qp = cb->child_cm_id->qp; 495 } else { 496 ret = rdma_create_qp(cb->cm_id, cb->pd, &init_attr); 497 if (!ret) 498 cb->qp = cb->cm_id->qp; 499 } 500 501 return ret; 502 } 503 504 static void rping_free_qp(struct rping_cb *cb) 505 { 506 ibv_destroy_qp(cb->qp); 507 ibv_destroy_cq(cb->cq); 508 ibv_destroy_comp_channel(cb->channel); 509 ibv_dealloc_pd(cb->pd); 510 } 511 512 static int rping_setup_qp(struct rping_cb *cb, struct rdma_cm_id *cm_id) 513 { 514 int ret; 515 516 cb->pd = ibv_alloc_pd(cm_id->verbs); 517 if (!cb->pd) { 518 fprintf(stderr, "ibv_alloc_pd failed\n"); 519 return errno; 520 } 521 DEBUG_LOG("created pd %p\n", cb->pd); 522 523 cb->channel = ibv_create_comp_channel(cm_id->verbs); 524 if (!cb->channel) { 525 fprintf(stderr, "ibv_create_comp_channel failed\n"); 526 ret = errno; 527 goto err1; 528 } 529 DEBUG_LOG("created channel %p\n", cb->channel); 530 531 cb->cq = ibv_create_cq(cm_id->verbs, RPING_SQ_DEPTH * 2, cb, 532 cb->channel, 0); 533 if (!cb->cq) { 534 fprintf(stderr, "ibv_create_cq failed\n"); 535 ret = errno; 536 goto err2; 537 } 538 DEBUG_LOG("created cq %p\n", cb->cq); 539 540 ret = ibv_req_notify_cq(cb->cq, 0); 541 if (ret) { 542 fprintf(stderr, "ibv_create_cq failed\n"); 543 ret = errno; 544 goto err3; 545 } 546 547 ret = rping_create_qp(cb); 548 if (ret) { 549 perror("rdma_create_qp"); 550 goto err3; 551 } 552 DEBUG_LOG("created qp %p\n", cb->qp); 553 return 0; 554 555 err3: 556 ibv_destroy_cq(cb->cq); 557 err2: 558 ibv_destroy_comp_channel(cb->channel); 559 err1: 560 ibv_dealloc_pd(cb->pd); 561 return ret; 562 } 563 564 static void *cm_thread(void *arg) 565 { 566 struct rping_cb *cb = arg; 567 struct rdma_cm_event *event; 568 int ret; 569 570 while (1) { 571 ret = rdma_get_cm_event(cb->cm_channel, &event); 572 if (ret) { 573 perror("rdma_get_cm_event"); 574 exit(ret); 575 } 576 ret = rping_cma_event_handler(event->id, event); 577 rdma_ack_cm_event(event); 578 if (ret) 579 exit(ret); 580 } 581 } 582 583 static void *cq_thread(void *arg) 584 { 585 struct rping_cb *cb = arg; 586 struct ibv_cq *ev_cq; 587 void *ev_ctx; 588 int ret; 589 590 DEBUG_LOG("cq_thread started.\n"); 591 592 while (1) { 593 pthread_testcancel(); 594 595 ret = ibv_get_cq_event(cb->channel, &ev_cq, &ev_ctx); 596 if (ret) { 597 fprintf(stderr, "Failed to get cq event!\n"); 598 pthread_exit(NULL); 599 } 600 if (ev_cq != cb->cq) { 601 fprintf(stderr, "Unknown CQ!\n"); 602 pthread_exit(NULL); 603 } 604 ret = ibv_req_notify_cq(cb->cq, 0); 605 if (ret) { 606 fprintf(stderr, "Failed to set notify!\n"); 607 pthread_exit(NULL); 608 } 609 ret = rping_cq_event_handler(cb); 610 ibv_ack_cq_events(cb->cq, 1); 611 if (ret) 612 pthread_exit(NULL); 613 } 614 } 615 616 static void rping_format_send(struct rping_cb *cb, char *buf, struct ibv_mr *mr) 617 { 618 struct rping_rdma_info *info = &cb->send_buf; 619 620 info->buf = htonll((uint64_t) (unsigned long) buf); 621 info->rkey = htonl(mr->rkey); 622 info->size = htonl(cb->size); 623 624 DEBUG_LOG("RDMA addr %" PRIx64" rkey %x len %d\n", 625 ntohll(info->buf), ntohl(info->rkey), ntohl(info->size)); 626 } 627 628 static int rping_test_server(struct rping_cb *cb) 629 { 630 struct ibv_send_wr *bad_wr; 631 int ret; 632 633 while (1) { 634 /* Wait for client's Start STAG/TO/Len */ 635 sem_wait(&cb->sem); 636 if (cb->state != RDMA_READ_ADV) { 637 fprintf(stderr, "wait for RDMA_READ_ADV state %d\n", 638 cb->state); 639 ret = -1; 640 break; 641 } 642 643 DEBUG_LOG("server received sink adv\n"); 644 645 /* Issue RDMA Read. */ 646 cb->rdma_sq_wr.opcode = IBV_WR_RDMA_READ; 647 cb->rdma_sq_wr.wr.rdma.rkey = cb->remote_rkey; 648 cb->rdma_sq_wr.wr.rdma.remote_addr = cb->remote_addr; 649 cb->rdma_sq_wr.sg_list->length = cb->remote_len; 650 651 ret = ibv_post_send(cb->qp, &cb->rdma_sq_wr, &bad_wr); 652 if (ret) { 653 fprintf(stderr, "post send error %d\n", ret); 654 break; 655 } 656 DEBUG_LOG("server posted rdma read req \n"); 657 658 /* Wait for read completion */ 659 sem_wait(&cb->sem); 660 if (cb->state != RDMA_READ_COMPLETE) { 661 fprintf(stderr, "wait for RDMA_READ_COMPLETE state %d\n", 662 cb->state); 663 ret = -1; 664 break; 665 } 666 DEBUG_LOG("server received read complete\n"); 667 668 /* Display data in recv buf */ 669 if (cb->verbose) 670 printf("server ping data: %s\n", cb->rdma_buf); 671 672 /* Tell client to continue */ 673 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr); 674 if (ret) { 675 fprintf(stderr, "post send error %d\n", ret); 676 break; 677 } 678 DEBUG_LOG("server posted go ahead\n"); 679 680 /* Wait for client's RDMA STAG/TO/Len */ 681 sem_wait(&cb->sem); 682 if (cb->state != RDMA_WRITE_ADV) { 683 fprintf(stderr, "wait for RDMA_WRITE_ADV state %d\n", 684 cb->state); 685 ret = -1; 686 break; 687 } 688 DEBUG_LOG("server received sink adv\n"); 689 690 /* RDMA Write echo data */ 691 cb->rdma_sq_wr.opcode = IBV_WR_RDMA_WRITE; 692 cb->rdma_sq_wr.wr.rdma.rkey = cb->remote_rkey; 693 cb->rdma_sq_wr.wr.rdma.remote_addr = cb->remote_addr; 694 cb->rdma_sq_wr.sg_list->length = strlen(cb->rdma_buf) + 1; 695 DEBUG_LOG("rdma write from lkey %x laddr %" PRIx64 " len %d\n", 696 cb->rdma_sq_wr.sg_list->lkey, 697 cb->rdma_sq_wr.sg_list->addr, 698 cb->rdma_sq_wr.sg_list->length); 699 700 ret = ibv_post_send(cb->qp, &cb->rdma_sq_wr, &bad_wr); 701 if (ret) { 702 fprintf(stderr, "post send error %d\n", ret); 703 break; 704 } 705 706 /* Wait for completion */ 707 ret = sem_wait(&cb->sem); 708 if (cb->state != RDMA_WRITE_COMPLETE) { 709 fprintf(stderr, "wait for RDMA_WRITE_COMPLETE state %d\n", 710 cb->state); 711 ret = -1; 712 break; 713 } 714 DEBUG_LOG("server rdma write complete \n"); 715 716 /* Tell client to begin again */ 717 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr); 718 if (ret) { 719 fprintf(stderr, "post send error %d\n", ret); 720 break; 721 } 722 DEBUG_LOG("server posted go ahead\n"); 723 } 724 725 return ret; 726 } 727 728 static int rping_bind_server(struct rping_cb *cb) 729 { 730 int ret; 731 732 if (cb->sin.ss_family == AF_INET) 733 ((struct sockaddr_in *) &cb->sin)->sin_port = cb->port; 734 else 735 ((struct sockaddr_in6 *) &cb->sin)->sin6_port = cb->port; 736 737 ret = rdma_bind_addr(cb->cm_id, (struct sockaddr *) &cb->sin); 738 if (ret) { 739 perror("rdma_bind_addr"); 740 return ret; 741 } 742 DEBUG_LOG("rdma_bind_addr successful\n"); 743 744 DEBUG_LOG("rdma_listen\n"); 745 ret = rdma_listen(cb->cm_id, 3); 746 if (ret) { 747 perror("rdma_listen"); 748 return ret; 749 } 750 751 return 0; 752 } 753 754 static struct rping_cb *clone_cb(struct rping_cb *listening_cb) 755 { 756 struct rping_cb *cb = malloc(sizeof *cb); 757 if (!cb) 758 return NULL; 759 *cb = *listening_cb; 760 cb->child_cm_id->context = cb; 761 return cb; 762 } 763 764 static void free_cb(struct rping_cb *cb) 765 { 766 free(cb); 767 } 768 769 static void *rping_persistent_server_thread(void *arg) 770 { 771 struct rping_cb *cb = arg; 772 struct ibv_recv_wr *bad_wr; 773 int ret; 774 775 ret = rping_setup_qp(cb, cb->child_cm_id); 776 if (ret) { 777 fprintf(stderr, "setup_qp failed: %d\n", ret); 778 goto err0; 779 } 780 781 ret = rping_setup_buffers(cb); 782 if (ret) { 783 fprintf(stderr, "rping_setup_buffers failed: %d\n", ret); 784 goto err1; 785 } 786 787 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr); 788 if (ret) { 789 fprintf(stderr, "ibv_post_recv failed: %d\n", ret); 790 goto err2; 791 } 792 793 pthread_create(&cb->cqthread, NULL, cq_thread, cb); 794 795 ret = rping_accept(cb); 796 if (ret) { 797 fprintf(stderr, "connect error %d\n", ret); 798 goto err3; 799 } 800 801 rping_test_server(cb); 802 rdma_disconnect(cb->child_cm_id); 803 rping_free_buffers(cb); 804 rping_free_qp(cb); 805 pthread_cancel(cb->cqthread); 806 pthread_join(cb->cqthread, NULL); 807 rdma_destroy_id(cb->child_cm_id); 808 free_cb(cb); 809 return NULL; 810 err3: 811 pthread_cancel(cb->cqthread); 812 pthread_join(cb->cqthread, NULL); 813 err2: 814 rping_free_buffers(cb); 815 err1: 816 rping_free_qp(cb); 817 err0: 818 free_cb(cb); 819 return NULL; 820 } 821 822 static int rping_run_persistent_server(struct rping_cb *listening_cb) 823 { 824 int ret; 825 struct rping_cb *cb; 826 827 ret = rping_bind_server(listening_cb); 828 if (ret) 829 return ret; 830 831 while (1) { 832 sem_wait(&listening_cb->sem); 833 if (listening_cb->state != CONNECT_REQUEST) { 834 fprintf(stderr, "wait for CONNECT_REQUEST state %d\n", 835 listening_cb->state); 836 return -1; 837 } 838 839 cb = clone_cb(listening_cb); 840 if (!cb) 841 return -1; 842 pthread_create(&cb->persistent_server_thread, NULL, rping_persistent_server_thread, cb); 843 } 844 return 0; 845 } 846 847 static int rping_run_server(struct rping_cb *cb) 848 { 849 struct ibv_recv_wr *bad_wr; 850 int ret; 851 852 ret = rping_bind_server(cb); 853 if (ret) 854 return ret; 855 856 sem_wait(&cb->sem); 857 if (cb->state != CONNECT_REQUEST) { 858 fprintf(stderr, "wait for CONNECT_REQUEST state %d\n", 859 cb->state); 860 return -1; 861 } 862 863 ret = rping_setup_qp(cb, cb->child_cm_id); 864 if (ret) { 865 fprintf(stderr, "setup_qp failed: %d\n", ret); 866 return ret; 867 } 868 869 ret = rping_setup_buffers(cb); 870 if (ret) { 871 fprintf(stderr, "rping_setup_buffers failed: %d\n", ret); 872 goto err1; 873 } 874 875 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr); 876 if (ret) { 877 fprintf(stderr, "ibv_post_recv failed: %d\n", ret); 878 goto err2; 879 } 880 881 pthread_create(&cb->cqthread, NULL, cq_thread, cb); 882 883 ret = rping_accept(cb); 884 if (ret) { 885 fprintf(stderr, "connect error %d\n", ret); 886 goto err2; 887 } 888 889 rping_test_server(cb); 890 rdma_disconnect(cb->child_cm_id); 891 rdma_destroy_id(cb->child_cm_id); 892 err2: 893 rping_free_buffers(cb); 894 err1: 895 rping_free_qp(cb); 896 897 return ret; 898 } 899 900 static int rping_test_client(struct rping_cb *cb) 901 { 902 int ping, start, cc, i, ret = 0; 903 struct ibv_send_wr *bad_wr; 904 unsigned char c; 905 906 start = 65; 907 for (ping = 0; !cb->count || ping < cb->count; ping++) { 908 cb->state = RDMA_READ_ADV; 909 910 /* Put some ascii text in the buffer. */ 911 cc = sprintf(cb->start_buf, RPING_MSG_FMT, ping); 912 for (i = cc, c = start; i < cb->size; i++) { 913 cb->start_buf[i] = c; 914 c++; 915 if (c > 122) 916 c = 65; 917 } 918 start++; 919 if (start > 122) 920 start = 65; 921 cb->start_buf[cb->size - 1] = 0; 922 923 rping_format_send(cb, cb->start_buf, cb->start_mr); 924 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr); 925 if (ret) { 926 fprintf(stderr, "post send error %d\n", ret); 927 break; 928 } 929 930 /* Wait for server to ACK */ 931 sem_wait(&cb->sem); 932 if (cb->state != RDMA_WRITE_ADV) { 933 fprintf(stderr, "wait for RDMA_WRITE_ADV state %d\n", 934 cb->state); 935 ret = -1; 936 break; 937 } 938 939 rping_format_send(cb, cb->rdma_buf, cb->rdma_mr); 940 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr); 941 if (ret) { 942 fprintf(stderr, "post send error %d\n", ret); 943 break; 944 } 945 946 /* Wait for the server to say the RDMA Write is complete. */ 947 sem_wait(&cb->sem); 948 if (cb->state != RDMA_WRITE_COMPLETE) { 949 fprintf(stderr, "wait for RDMA_WRITE_COMPLETE state %d\n", 950 cb->state); 951 ret = -1; 952 break; 953 } 954 955 if (cb->validate) 956 if (memcmp(cb->start_buf, cb->rdma_buf, cb->size)) { 957 fprintf(stderr, "data mismatch!\n"); 958 ret = -1; 959 break; 960 } 961 962 if (cb->verbose) 963 printf("ping data: %s\n", cb->rdma_buf); 964 } 965 966 return ret; 967 } 968 969 static int rping_connect_client(struct rping_cb *cb) 970 { 971 struct rdma_conn_param conn_param; 972 int ret; 973 974 memset(&conn_param, 0, sizeof conn_param); 975 conn_param.responder_resources = 1; 976 conn_param.initiator_depth = 1; 977 conn_param.retry_count = 10; 978 979 ret = rdma_connect(cb->cm_id, &conn_param); 980 if (ret) { 981 perror("rdma_connect"); 982 return ret; 983 } 984 985 sem_wait(&cb->sem); 986 if (cb->state != CONNECTED) { 987 fprintf(stderr, "wait for CONNECTED state %d\n", cb->state); 988 return -1; 989 } 990 991 DEBUG_LOG("rmda_connect successful\n"); 992 return 0; 993 } 994 995 static int rping_bind_client(struct rping_cb *cb) 996 { 997 int ret; 998 999 if (cb->sin.ss_family == AF_INET) 1000 ((struct sockaddr_in *) &cb->sin)->sin_port = cb->port; 1001 else 1002 ((struct sockaddr_in6 *) &cb->sin)->sin6_port = cb->port; 1003 1004 ret = rdma_resolve_addr(cb->cm_id, NULL, (struct sockaddr *) &cb->sin, 2000); 1005 if (ret) { 1006 perror("rdma_resolve_addr"); 1007 return ret; 1008 } 1009 1010 sem_wait(&cb->sem); 1011 if (cb->state != ROUTE_RESOLVED) { 1012 fprintf(stderr, "waiting for addr/route resolution state %d\n", 1013 cb->state); 1014 return -1; 1015 } 1016 1017 DEBUG_LOG("rdma_resolve_addr - rdma_resolve_route successful\n"); 1018 return 0; 1019 } 1020 1021 static int rping_run_client(struct rping_cb *cb) 1022 { 1023 struct ibv_recv_wr *bad_wr; 1024 int ret; 1025 1026 ret = rping_bind_client(cb); 1027 if (ret) 1028 return ret; 1029 1030 ret = rping_setup_qp(cb, cb->cm_id); 1031 if (ret) { 1032 fprintf(stderr, "setup_qp failed: %d\n", ret); 1033 return ret; 1034 } 1035 1036 ret = rping_setup_buffers(cb); 1037 if (ret) { 1038 fprintf(stderr, "rping_setup_buffers failed: %d\n", ret); 1039 goto err1; 1040 } 1041 1042 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr); 1043 if (ret) { 1044 fprintf(stderr, "ibv_post_recv failed: %d\n", ret); 1045 goto err2; 1046 } 1047 1048 pthread_create(&cb->cqthread, NULL, cq_thread, cb); 1049 1050 ret = rping_connect_client(cb); 1051 if (ret) { 1052 fprintf(stderr, "connect error %d\n", ret); 1053 goto err2; 1054 } 1055 1056 rping_test_client(cb); 1057 rdma_disconnect(cb->cm_id); 1058 err2: 1059 rping_free_buffers(cb); 1060 err1: 1061 rping_free_qp(cb); 1062 1063 return ret; 1064 } 1065 1066 static int get_addr(char *dst, struct sockaddr *addr) 1067 { 1068 struct addrinfo *res; 1069 int ret; 1070 1071 ret = getaddrinfo(dst, NULL, NULL, &res); 1072 if (ret) { 1073 printf("getaddrinfo failed - invalid hostname or IP address\n"); 1074 return ret; 1075 } 1076 1077 if (res->ai_family == PF_INET) 1078 memcpy(addr, res->ai_addr, sizeof(struct sockaddr_in)); 1079 else if (res->ai_family == PF_INET6) 1080 memcpy(addr, res->ai_addr, sizeof(struct sockaddr_in6)); 1081 else 1082 ret = -1; 1083 1084 freeaddrinfo(res); 1085 return ret; 1086 } 1087 1088 static void usage(char *name) 1089 { 1090 printf("%s -s [-vVd] [-S size] [-C count] [-a addr] [-p port]\n", 1091 name); 1092 printf("%s -c [-vVd] [-S size] [-C count] -a addr [-p port]\n", 1093 name); 1094 printf("\t-c\t\tclient side\n"); 1095 printf("\t-s\t\tserver side. To bind to any address with IPv6 use -a ::0\n"); 1096 printf("\t-v\t\tdisplay ping data to stdout\n"); 1097 printf("\t-V\t\tvalidate ping data\n"); 1098 printf("\t-d\t\tdebug printfs\n"); 1099 printf("\t-S size \tping data size\n"); 1100 printf("\t-C count\tping count times\n"); 1101 printf("\t-a addr\t\taddress\n"); 1102 printf("\t-p port\t\tport\n"); 1103 printf("\t-P\t\tpersistent server mode allowing multiple connections\n"); 1104 } 1105 1106 int main(int argc, char *argv[]) 1107 { 1108 struct rping_cb *cb; 1109 int op; 1110 int ret = 0; 1111 int persistent_server = 0; 1112 1113 cb = malloc(sizeof(*cb)); 1114 if (!cb) 1115 return -ENOMEM; 1116 1117 memset(cb, 0, sizeof(*cb)); 1118 cb->server = -1; 1119 cb->state = IDLE; 1120 cb->size = 64; 1121 cb->sin.ss_family = PF_INET; 1122 cb->port = htons(7174); 1123 sem_init(&cb->sem, 0, 0); 1124 1125 opterr = 0; 1126 while ((op=getopt(argc, argv, "a:Pp:C:S:t:scvVd")) != -1) { 1127 switch (op) { 1128 case 'a': 1129 ret = get_addr(optarg, (struct sockaddr *) &cb->sin); 1130 break; 1131 case 'P': 1132 persistent_server = 1; 1133 break; 1134 case 'p': 1135 cb->port = htons(atoi(optarg)); 1136 DEBUG_LOG("port %d\n", (int) atoi(optarg)); 1137 break; 1138 case 's': 1139 cb->server = 1; 1140 DEBUG_LOG("server\n"); 1141 break; 1142 case 'c': 1143 cb->server = 0; 1144 DEBUG_LOG("client\n"); 1145 break; 1146 case 'S': 1147 cb->size = atoi(optarg); 1148 if ((cb->size < RPING_MIN_BUFSIZE) || 1149 (cb->size > (RPING_BUFSIZE - 1))) { 1150 fprintf(stderr, "Invalid size %d " 1151 "(valid range is %Zd to %d)\n", 1152 cb->size, RPING_MIN_BUFSIZE, RPING_BUFSIZE); 1153 ret = EINVAL; 1154 } else 1155 DEBUG_LOG("size %d\n", (int) atoi(optarg)); 1156 break; 1157 case 'C': 1158 cb->count = atoi(optarg); 1159 if (cb->count < 0) { 1160 fprintf(stderr, "Invalid count %d\n", 1161 cb->count); 1162 ret = EINVAL; 1163 } else 1164 DEBUG_LOG("count %d\n", (int) cb->count); 1165 break; 1166 case 'v': 1167 cb->verbose++; 1168 DEBUG_LOG("verbose\n"); 1169 break; 1170 case 'V': 1171 cb->validate++; 1172 DEBUG_LOG("validate data\n"); 1173 break; 1174 case 'd': 1175 debug++; 1176 break; 1177 default: 1178 usage("rping"); 1179 ret = EINVAL; 1180 goto out; 1181 } 1182 } 1183 if (ret) 1184 goto out; 1185 1186 if (cb->server == -1) { 1187 usage("rping"); 1188 ret = EINVAL; 1189 goto out; 1190 } 1191 1192 cb->cm_channel = rdma_create_event_channel(); 1193 if (!cb->cm_channel) { 1194 perror("rdma_create_event_channel"); 1195 goto out; 1196 } 1197 1198 ret = rdma_create_id(cb->cm_channel, &cb->cm_id, cb, RDMA_PS_TCP); 1199 if (ret) { 1200 perror("rdma_create_id"); 1201 goto out2; 1202 } 1203 DEBUG_LOG("created cm_id %p\n", cb->cm_id); 1204 1205 pthread_create(&cb->cmthread, NULL, cm_thread, cb); 1206 1207 if (cb->server) { 1208 if (persistent_server) 1209 ret = rping_run_persistent_server(cb); 1210 else 1211 ret = rping_run_server(cb); 1212 } else 1213 ret = rping_run_client(cb); 1214 1215 DEBUG_LOG("destroy cm_id %p\n", cb->cm_id); 1216 rdma_destroy_id(cb->cm_id); 1217 out2: 1218 rdma_destroy_event_channel(cb->cm_channel); 1219 out: 1220 free(cb); 1221 return ret; 1222 } 1223