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