2 * Copyright (c) 2005 Ammasso, Inc. All rights reserved.
3 * Copyright (c) 2006 Open Grid Computing, Inc. All rights reserved.
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:
11 * Redistribution and use in source and binary forms, with or
12 * without modification, are permitted provided that the following
15 * - Redistributions of source code must retain the above
16 * copyright notice, this list of conditions and the following
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.
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
34 #include <infiniband/endian.h>
40 #include <sys/types.h>
41 #include <sys/socket.h>
43 #include <semaphore.h>
47 #include <rdma/rdma_cma.h>
50 #define DEBUG_LOG if (debug) printf
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
66 * These states are used to signal events between the completion handler
67 * and the main client or server thread.
69 * Once CONNECTED, they cycle through RDMA_READ_ADV, RDMA_WRITE_ADV,
70 * and RDMA_WRITE_COMPLETE for each ping.
86 struct rping_rdma_info {
93 * Default max buffer size for IO...
95 #define RPING_BUFSIZE 64*1024
96 #define RPING_SQ_DEPTH 16
98 /* Default string for print data and
101 #define _stringify( _x ) # _x
102 #define stringify( _x ) _stringify( _x )
104 #define RPING_MSG_FMT "rdma-ping-%d: "
105 #define RPING_MIN_BUFSIZE sizeof(stringify(INT_MAX)) + sizeof(RPING_MSG_FMT)
108 * Control block struct.
111 int server; /* 0 iff client */
113 pthread_t persistent_server_thread;
114 struct ibv_comp_channel *channel;
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 */
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;
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;
134 uint32_t remote_rkey; /* remote guys RKEY */
135 uint64_t remote_addr; /* remote guys TO */
136 uint32_t remote_len; /* remote guys LEN */
138 char *start_buf; /* rdma read src */
139 struct ibv_mr *start_mr;
141 enum test_state state; /* used for cond/signalling */
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 */
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 */
160 static int rping_cma_event_handler(struct rdma_cm_id *cma_id,
161 struct rdma_cm_event *event)
164 struct rping_cb *cb = cma_id->context;
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");
170 switch (event->event) {
171 case RDMA_CM_EVENT_ADDR_RESOLVED:
172 cb->state = ADDR_RESOLVED;
173 ret = rdma_resolve_route(cma_id, 2000);
176 perror("rdma_resolve_route");
181 case RDMA_CM_EVENT_ROUTE_RESOLVED:
182 cb->state = ROUTE_RESOLVED;
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);
193 case RDMA_CM_EVENT_ESTABLISHED:
194 DEBUG_LOG("ESTABLISHED\n");
197 * Server will wake up when first RECV completes.
200 cb->state = CONNECTED;
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);
216 case RDMA_CM_EVENT_DISCONNECTED:
217 fprintf(stderr, "%s DISCONNECT EVENT...\n",
218 cb->server ? "server" : "client");
219 cb->state = DISCONNECTED;
223 case RDMA_CM_EVENT_DEVICE_REMOVAL:
224 fprintf(stderr, "cma detected device removal!!!!\n");
231 fprintf(stderr, "unhandled event: %s, ignoring\n",
232 rdma_event_str(event->event));
239 static int server_recv(struct rping_cb *cb, struct ibv_wc *wc)
241 if (wc->byte_len != sizeof(cb->recv_buf)) {
242 fprintf(stderr, "Received bogus data, size %d\n", wc->byte_len);
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);
252 if (cb->state <= CONNECTED || cb->state == RDMA_WRITE_COMPLETE)
253 cb->state = RDMA_READ_ADV;
255 cb->state = RDMA_WRITE_ADV;
260 static int client_recv(struct rping_cb *cb, struct ibv_wc *wc)
262 if (wc->byte_len != sizeof(cb->recv_buf)) {
263 fprintf(stderr, "Received bogus data, size %d\n", wc->byte_len);
267 if (cb->state == RDMA_READ_ADV)
268 cb->state = RDMA_WRITE_ADV;
270 cb->state = RDMA_WRITE_COMPLETE;
275 static int rping_cq_event_handler(struct rping_cb *cb)
278 struct ibv_recv_wr *bad_wr;
282 while ((ret = ibv_poll_cq(cb->cq, 1, &wc)) == 1) {
286 if (wc.status == IBV_WC_WR_FLUSH_ERR) {
292 "cq completion failed status %d\n",
300 DEBUG_LOG("send completion\n");
303 case IBV_WC_RDMA_WRITE:
304 DEBUG_LOG("rdma write completion\n");
305 cb->state = RDMA_WRITE_COMPLETE;
309 case IBV_WC_RDMA_READ:
310 DEBUG_LOG("rdma read completion\n");
311 cb->state = RDMA_READ_COMPLETE;
316 DEBUG_LOG("recv completion\n");
317 ret = cb->server ? server_recv(cb, &wc) :
318 client_recv(cb, &wc);
320 fprintf(stderr, "recv wc error: %d\n", ret);
324 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr);
326 fprintf(stderr, "post recv error: %d\n", ret);
333 DEBUG_LOG("unknown!!!!! completion\n");
339 fprintf(stderr, "poll error %d\n", ret);
350 static int rping_accept(struct rping_cb *cb)
354 DEBUG_LOG("accepting client connection request\n");
356 ret = rdma_accept(cb->child_cm_id, NULL);
358 perror("rdma_accept");
363 if (cb->state == ERROR) {
364 fprintf(stderr, "wait for CONNECTED state %d\n", cb->state);
370 static void rping_setup_wr(struct rping_cb *cb)
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;
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;
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;
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;
394 static int rping_setup_buffers(struct rping_cb *cb)
398 DEBUG_LOG("rping_setup_buffers called on cb %p\n", cb);
400 cb->recv_mr = ibv_reg_mr(cb->pd, &cb->recv_buf, sizeof cb->recv_buf,
401 IBV_ACCESS_LOCAL_WRITE);
403 fprintf(stderr, "recv_buf reg_mr failed\n");
407 cb->send_mr = ibv_reg_mr(cb->pd, &cb->send_buf, sizeof cb->send_buf, 0);
409 fprintf(stderr, "send_buf reg_mr failed\n");
414 cb->rdma_buf = malloc(cb->size);
416 fprintf(stderr, "rdma_buf malloc failed\n");
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);
426 fprintf(stderr, "rdma_buf reg_mr failed\n");
432 cb->start_buf = malloc(cb->size);
433 if (!cb->start_buf) {
434 fprintf(stderr, "start_buf malloc failed\n");
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);
444 fprintf(stderr, "start_buf reg_mr failed\n");
451 DEBUG_LOG("allocated & registered buffers...\n");
457 ibv_dereg_mr(cb->rdma_mr);
461 ibv_dereg_mr(cb->send_mr);
463 ibv_dereg_mr(cb->recv_mr);
467 static void rping_free_buffers(struct rping_cb *cb)
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);
475 ibv_dereg_mr(cb->start_mr);
480 static int rping_create_qp(struct rping_cb *cb)
482 struct ibv_qp_init_attr init_attr;
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;
495 ret = rdma_create_qp(cb->child_cm_id, cb->pd, &init_attr);
497 cb->qp = cb->child_cm_id->qp;
499 ret = rdma_create_qp(cb->cm_id, cb->pd, &init_attr);
501 cb->qp = cb->cm_id->qp;
507 static void rping_free_qp(struct rping_cb *cb)
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);
515 static int rping_setup_qp(struct rping_cb *cb, struct rdma_cm_id *cm_id)
519 cb->pd = ibv_alloc_pd(cm_id->verbs);
521 fprintf(stderr, "ibv_alloc_pd failed\n");
524 DEBUG_LOG("created pd %p\n", cb->pd);
526 cb->channel = ibv_create_comp_channel(cm_id->verbs);
528 fprintf(stderr, "ibv_create_comp_channel failed\n");
532 DEBUG_LOG("created channel %p\n", cb->channel);
534 cb->cq = ibv_create_cq(cm_id->verbs, RPING_SQ_DEPTH * 2, cb,
537 fprintf(stderr, "ibv_create_cq failed\n");
541 DEBUG_LOG("created cq %p\n", cb->cq);
543 ret = ibv_req_notify_cq(cb->cq, 0);
545 fprintf(stderr, "ibv_create_cq failed\n");
550 ret = rping_create_qp(cb);
552 perror("rdma_create_qp");
555 DEBUG_LOG("created qp %p\n", cb->qp);
559 ibv_destroy_cq(cb->cq);
561 ibv_destroy_comp_channel(cb->channel);
563 ibv_dealloc_pd(cb->pd);
567 static void *cm_thread(void *arg)
569 struct rping_cb *cb = arg;
570 struct rdma_cm_event *event;
574 ret = rdma_get_cm_event(cb->cm_channel, &event);
576 perror("rdma_get_cm_event");
579 ret = rping_cma_event_handler(event->id, event);
580 rdma_ack_cm_event(event);
586 static void *cq_thread(void *arg)
588 struct rping_cb *cb = arg;
589 struct ibv_cq *ev_cq;
593 DEBUG_LOG("cq_thread started.\n");
596 pthread_testcancel();
598 ret = ibv_get_cq_event(cb->channel, &ev_cq, &ev_ctx);
600 fprintf(stderr, "Failed to get cq event!\n");
603 if (ev_cq != cb->cq) {
604 fprintf(stderr, "Unknown CQ!\n");
607 ret = ibv_req_notify_cq(cb->cq, 0);
609 fprintf(stderr, "Failed to set notify!\n");
612 ret = rping_cq_event_handler(cb);
613 ibv_ack_cq_events(cb->cq, 1);
619 static void rping_format_send(struct rping_cb *cb, char *buf, struct ibv_mr *mr)
621 struct rping_rdma_info *info = &cb->send_buf;
623 info->buf = htobe64((uint64_t) (unsigned long) buf);
624 info->rkey = htobe32(mr->rkey);
625 info->size = htobe32(cb->size);
627 DEBUG_LOG("RDMA addr %" PRIx64" rkey %x len %d\n",
628 be64toh(info->buf), be32toh(info->rkey), be32toh(info->size));
631 static int rping_test_server(struct rping_cb *cb)
633 struct ibv_send_wr *bad_wr;
637 /* Wait for client's Start STAG/TO/Len */
639 if (cb->state != RDMA_READ_ADV) {
640 fprintf(stderr, "wait for RDMA_READ_ADV state %d\n",
646 DEBUG_LOG("server received sink adv\n");
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;
654 ret = ibv_post_send(cb->qp, &cb->rdma_sq_wr, &bad_wr);
656 fprintf(stderr, "post send error %d\n", ret);
659 DEBUG_LOG("server posted rdma read req \n");
661 /* Wait for read completion */
663 if (cb->state != RDMA_READ_COMPLETE) {
664 fprintf(stderr, "wait for RDMA_READ_COMPLETE state %d\n",
669 DEBUG_LOG("server received read complete\n");
671 /* Display data in recv buf */
673 printf("server ping data: %s\n", cb->rdma_buf);
675 /* Tell client to continue */
676 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr);
678 fprintf(stderr, "post send error %d\n", ret);
681 DEBUG_LOG("server posted go ahead\n");
683 /* Wait for client's RDMA STAG/TO/Len */
685 if (cb->state != RDMA_WRITE_ADV) {
686 fprintf(stderr, "wait for RDMA_WRITE_ADV state %d\n",
691 DEBUG_LOG("server received sink adv\n");
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);
703 ret = ibv_post_send(cb->qp, &cb->rdma_sq_wr, &bad_wr);
705 fprintf(stderr, "post send error %d\n", ret);
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",
717 DEBUG_LOG("server rdma write complete \n");
719 /* Tell client to begin again */
720 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr);
722 fprintf(stderr, "post send error %d\n", ret);
725 DEBUG_LOG("server posted go ahead\n");
728 return (cb->state == DISCONNECTED) ? 0 : ret;
731 static int rping_bind_server(struct rping_cb *cb)
735 if (cb->sin.ss_family == AF_INET)
736 ((struct sockaddr_in *) &cb->sin)->sin_port = cb->port;
738 ((struct sockaddr_in6 *) &cb->sin)->sin6_port = cb->port;
740 ret = rdma_bind_addr(cb->cm_id, (struct sockaddr *) &cb->sin);
742 perror("rdma_bind_addr");
745 DEBUG_LOG("rdma_bind_addr successful\n");
747 DEBUG_LOG("rdma_listen\n");
748 ret = rdma_listen(cb->cm_id, 3);
750 perror("rdma_listen");
757 static struct rping_cb *clone_cb(struct rping_cb *listening_cb)
759 struct rping_cb *cb = malloc(sizeof *cb);
762 memset(cb, 0, sizeof *cb);
764 cb->child_cm_id->context = cb;
768 static void free_cb(struct rping_cb *cb)
773 static void *rping_persistent_server_thread(void *arg)
775 struct rping_cb *cb = arg;
776 struct ibv_recv_wr *bad_wr;
779 ret = rping_setup_qp(cb, cb->child_cm_id);
781 fprintf(stderr, "setup_qp failed: %d\n", ret);
785 ret = rping_setup_buffers(cb);
787 fprintf(stderr, "rping_setup_buffers failed: %d\n", ret);
791 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr);
793 fprintf(stderr, "ibv_post_recv failed: %d\n", ret);
797 ret = pthread_create(&cb->cqthread, NULL, cq_thread, cb);
799 perror("pthread_create");
803 ret = rping_accept(cb);
805 fprintf(stderr, "connect error %d\n", ret);
809 rping_test_server(cb);
810 rdma_disconnect(cb->child_cm_id);
811 pthread_join(cb->cqthread, NULL);
812 rping_free_buffers(cb);
814 rdma_destroy_id(cb->child_cm_id);
818 pthread_cancel(cb->cqthread);
819 pthread_join(cb->cqthread, NULL);
821 rping_free_buffers(cb);
829 static int rping_run_persistent_server(struct rping_cb *listening_cb)
835 ret = rping_bind_server(listening_cb);
840 * Set persistent server threads to DEATCHED state so
841 * they release all their resources when they exit.
843 ret = pthread_attr_init(&attr);
845 perror("pthread_attr_init");
848 ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
850 perror("pthread_attr_setdetachstate");
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);
862 cb = clone_cb(listening_cb);
866 ret = pthread_create(&cb->persistent_server_thread, &attr, rping_persistent_server_thread, cb);
868 perror("pthread_create");
875 static int rping_run_server(struct rping_cb *cb)
877 struct ibv_recv_wr *bad_wr;
880 ret = rping_bind_server(cb);
885 if (cb->state != CONNECT_REQUEST) {
886 fprintf(stderr, "wait for CONNECT_REQUEST state %d\n",
891 ret = rping_setup_qp(cb, cb->child_cm_id);
893 fprintf(stderr, "setup_qp failed: %d\n", ret);
897 ret = rping_setup_buffers(cb);
899 fprintf(stderr, "rping_setup_buffers failed: %d\n", ret);
903 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr);
905 fprintf(stderr, "ibv_post_recv failed: %d\n", ret);
909 ret = pthread_create(&cb->cqthread, NULL, cq_thread, cb);
911 perror("pthread_create");
915 ret = rping_accept(cb);
917 fprintf(stderr, "connect error %d\n", ret);
921 ret = rping_test_server(cb);
923 fprintf(stderr, "rping server failed: %d\n", ret);
929 rdma_disconnect(cb->child_cm_id);
930 pthread_join(cb->cqthread, NULL);
931 rdma_destroy_id(cb->child_cm_id);
933 rping_free_buffers(cb);
940 static int rping_test_client(struct rping_cb *cb)
942 int ping, start, cc, i, ret = 0;
943 struct ibv_send_wr *bad_wr;
947 for (ping = 0; !cb->count || ping < cb->count; ping++) {
948 cb->state = RDMA_READ_ADV;
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;
961 cb->start_buf[cb->size - 1] = 0;
963 rping_format_send(cb, cb->start_buf, cb->start_mr);
964 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr);
966 fprintf(stderr, "post send error %d\n", ret);
970 /* Wait for server to ACK */
972 if (cb->state != RDMA_WRITE_ADV) {
973 fprintf(stderr, "wait for RDMA_WRITE_ADV state %d\n",
979 rping_format_send(cb, cb->rdma_buf, cb->rdma_mr);
980 ret = ibv_post_send(cb->qp, &cb->sq_wr, &bad_wr);
982 fprintf(stderr, "post send error %d\n", ret);
986 /* Wait for the server to say the RDMA Write is complete. */
988 if (cb->state != RDMA_WRITE_COMPLETE) {
989 fprintf(stderr, "wait for RDMA_WRITE_COMPLETE state %d\n",
996 if (memcmp(cb->start_buf, cb->rdma_buf, cb->size)) {
997 fprintf(stderr, "data mismatch!\n");
1003 printf("ping data: %s\n", cb->rdma_buf);
1006 return (cb->state == DISCONNECTED) ? 0 : ret;
1009 static int rping_connect_client(struct rping_cb *cb)
1011 struct rdma_conn_param conn_param;
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;
1019 ret = rdma_connect(cb->cm_id, &conn_param);
1021 perror("rdma_connect");
1026 if (cb->state != CONNECTED) {
1027 fprintf(stderr, "wait for CONNECTED state %d\n", cb->state);
1031 DEBUG_LOG("rmda_connect successful\n");
1035 static int rping_bind_client(struct rping_cb *cb)
1039 if (cb->sin.ss_family == AF_INET)
1040 ((struct sockaddr_in *) &cb->sin)->sin_port = cb->port;
1042 ((struct sockaddr_in6 *) &cb->sin)->sin6_port = cb->port;
1044 if (cb->ssource.ss_family)
1045 ret = rdma_resolve_addr(cb->cm_id, (struct sockaddr *) &cb->ssource,
1046 (struct sockaddr *) &cb->sin, 2000);
1048 ret = rdma_resolve_addr(cb->cm_id, NULL, (struct sockaddr *) &cb->sin, 2000);
1051 perror("rdma_resolve_addr");
1056 if (cb->state != ROUTE_RESOLVED) {
1057 fprintf(stderr, "waiting for addr/route resolution state %d\n",
1062 DEBUG_LOG("rdma_resolve_addr - rdma_resolve_route successful\n");
1066 static int rping_run_client(struct rping_cb *cb)
1068 struct ibv_recv_wr *bad_wr;
1071 ret = rping_bind_client(cb);
1075 ret = rping_setup_qp(cb, cb->cm_id);
1077 fprintf(stderr, "setup_qp failed: %d\n", ret);
1081 ret = rping_setup_buffers(cb);
1083 fprintf(stderr, "rping_setup_buffers failed: %d\n", ret);
1087 ret = ibv_post_recv(cb->qp, &cb->rq_wr, &bad_wr);
1089 fprintf(stderr, "ibv_post_recv failed: %d\n", ret);
1093 ret = pthread_create(&cb->cqthread, NULL, cq_thread, cb);
1095 perror("pthread_create");
1099 ret = rping_connect_client(cb);
1101 fprintf(stderr, "connect error %d\n", ret);
1105 ret = rping_test_client(cb);
1107 fprintf(stderr, "rping client failed: %d\n", ret);
1113 rdma_disconnect(cb->cm_id);
1115 pthread_join(cb->cqthread, NULL);
1117 rping_free_buffers(cb);
1124 static int get_addr(char *dst, struct sockaddr *addr)
1126 struct addrinfo *res;
1129 ret = getaddrinfo(dst, NULL, NULL, &res);
1131 printf("getaddrinfo failed (%s) - invalid hostname or IP address\n", gai_strerror(ret));
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));
1146 static void usage(const char *name)
1148 printf("%s -s [-vVd] [-S size] [-C count] [-a addr] [-p port]\n",
1150 printf("%s -c [-vVd] [-S size] [-C count] [-I addr] -a addr [-p port]\n",
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");
1165 int main(int argc, char *argv[])
1167 struct rping_cb *cb;
1170 int persistent_server = 0;
1172 cb = malloc(sizeof(*cb));
1176 memset(cb, 0, sizeof(*cb));
1180 cb->port = htobe16(7174);
1181 sem_init(&cb->sem, 0, 0);
1183 /* initialize sockaddr structure with defaults */
1184 ret = get_addr("0.0.0.0", (struct sockaddr *) &cb->sin);
1189 while ((op=getopt(argc, argv, "a:I:Pp:C:S:t:scvVd")) != -1) {
1192 ret = get_addr(optarg, (struct sockaddr *) &cb->sin);
1195 ret = get_addr(optarg, (struct sockaddr *) &cb->ssource);
1198 persistent_server = 1;
1201 cb->port = htobe16(atoi(optarg));
1202 DEBUG_LOG("port %d\n", (int) atoi(optarg));
1206 DEBUG_LOG("server\n");
1210 DEBUG_LOG("client\n");
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);
1221 DEBUG_LOG("size %d\n", (int) atoi(optarg));
1224 cb->count = atoi(optarg);
1225 if (cb->count < 0) {
1226 fprintf(stderr, "Invalid count %d\n",
1230 DEBUG_LOG("count %d\n", (int) cb->count);
1234 DEBUG_LOG("verbose\n");
1238 DEBUG_LOG("validate data\n");
1252 if (cb->server == -1) {
1258 cb->cm_channel = rdma_create_event_channel();
1259 if (!cb->cm_channel) {
1260 perror("rdma_create_event_channel");
1265 ret = rdma_create_id(cb->cm_channel, &cb->cm_id, cb, RDMA_PS_TCP);
1267 perror("rdma_create_id");
1270 DEBUG_LOG("created cm_id %p\n", cb->cm_id);
1272 ret = pthread_create(&cb->cmthread, NULL, cm_thread, cb);
1274 perror("pthread_create");
1279 if (persistent_server)
1280 ret = rping_run_persistent_server(cb);
1282 ret = rping_run_server(cb);
1284 ret = rping_run_client(cb);
1287 DEBUG_LOG("destroy cm_id %p\n", cb->cm_id);
1288 rdma_destroy_id(cb->cm_id);
1290 rdma_destroy_event_channel(cb->cm_channel);