2 * Copyright (c) 2008-2014 Intel Corporation. All rights reserved.
4 * This software is available to you under a choice of one of two
5 * licenses. You may choose to be licensed under the terms of the GNU
6 * General Public License (GPL) Version 2, available from the file
7 * COPYING in the main directory of this source tree, or the
8 * OpenIB.org BSD license below:
10 * Redistribution and use in source and binary forms, with or
11 * without modification, are permitted provided that the following
14 * - Redistributions of source code must retain the above
15 * copyright notice, this list of conditions and the following
18 * - Redistributions in binary form must reproduce the above
19 * copyright notice, this list of conditions and the following
20 * disclaimer in the documentation and/or other materials
21 * provided with the distribution.
23 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
24 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
25 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
26 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
27 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
28 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
29 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
36 #include <sys/types.h>
37 #include <sys/socket.h>
39 #include <infiniband/endian.h>
47 #include <netinet/tcp.h>
48 #include <sys/epoll.h>
51 #include <util/compiler.h>
53 #include <rdma/rdma_cma.h>
54 #include <rdma/rdma_verbs.h>
55 #include <rdma/rsocket.h>
59 #define RS_OLAP_START_SIZE 2048
60 #define RS_MAX_TRANSFER 65536
61 #define RS_SNDLOWAT 2048
62 #define RS_QP_MIN_SIZE 16
63 #define RS_QP_MAX_SIZE 0xFFFE
64 #define RS_QP_CTRL_SIZE 4 /* must be power of 2 */
65 #define RS_CONN_RETRIES 6
67 static struct index_map idm;
68 static pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER;
93 void *(*run)(void *svc);
98 static struct pollfd *udp_svc_fds;
99 static void *udp_svc_run(void *arg);
100 static struct rs_svc udp_svc = {
101 .context_size = sizeof(*udp_svc_fds),
104 static uint32_t *tcp_svc_timeouts;
105 static void *tcp_svc_run(void *arg);
106 static struct rs_svc tcp_svc = {
107 .context_size = sizeof(*tcp_svc_timeouts),
111 static uint16_t def_iomap_size = 0;
112 static uint16_t def_inline = 64;
113 static uint16_t def_sqsize = 384;
114 static uint16_t def_rqsize = 384;
115 static uint32_t def_mem = (1 << 17);
116 static uint32_t def_wmem = (1 << 17);
117 static uint32_t polling_time = 10;
120 * Immediate data format is determined by the upper bits
121 * bit 31: message type, 0 - data, 1 - control
122 * bit 30: buffers updated, 0 - target, 1 - direct-receive
123 * bit 29: more data, 0 - end of transfer, 1 - more data available
125 * for data transfers:
126 * bits [28:0]: bytes transferred
127 * for control messages:
129 * bits [28-0]: receive credits granted
131 * bits [28-16]: reserved, bits [15-0]: index
136 RS_OP_RSVD_DATA_MORE,
137 RS_OP_WRITE, /* opcode is not transmitted over the network */
144 #define rs_msg_set(op, data) ((op << 29) | (uint32_t) (data))
145 #define rs_msg_op(imm_data) (imm_data >> 29)
146 #define rs_msg_data(imm_data) (imm_data & 0x1FFFFFFF)
147 #define RS_MSG_SIZE sizeof(uint32_t)
149 #define RS_WR_ID_FLAG_RECV (((uint64_t) 1) << 63)
150 #define RS_WR_ID_FLAG_MSG_SEND (((uint64_t) 1) << 62) /* See RS_OPT_MSG_SEND */
151 #define rs_send_wr_id(data) ((uint64_t) data)
152 #define rs_recv_wr_id(data) (RS_WR_ID_FLAG_RECV | (uint64_t) data)
153 #define rs_wr_is_recv(wr_id) (wr_id & RS_WR_ID_FLAG_RECV)
154 #define rs_wr_is_msg_send(wr_id) (wr_id & RS_WR_ID_FLAG_MSG_SEND)
155 #define rs_wr_data(wr_id) ((uint32_t) wr_id)
177 struct ds_smsg *next;
196 int index; /* -1 if mapping is local and not in iomap_list */
199 #define RS_MAX_CTRL_MSG (sizeof(struct rs_sge))
200 #define rs_host_is_net() (__BYTE_ORDER == __BIG_ENDIAN)
201 #define RS_CONN_FLAG_NET (1 << 0)
202 #define RS_CONN_FLAG_IOMAP (1 << 1)
204 struct rs_conn_data {
209 uint8_t target_iomap_size;
210 struct rs_sge target_sgl;
211 struct rs_sge data_buf;
214 struct rs_conn_private_data {
216 struct rs_conn_data conn_data;
218 struct ib_connect_hdr ib_hdr;
219 struct rs_conn_data conn_data;
225 * rsocket states are ordered as passive, connecting, connected, disconnected.
230 rs_listening = 0x0002,
232 rs_resolving_addr = rs_opening | 0x0010,
233 rs_resolving_route = rs_opening | 0x0020,
234 rs_connecting = rs_opening | 0x0040,
235 rs_accepting = rs_opening | 0x0080,
236 rs_connected = 0x0100,
237 rs_writable = 0x0200,
238 rs_readable = 0x0400,
239 rs_connect_rdwr = rs_connected | rs_readable | rs_writable,
240 rs_connect_error = 0x0800,
241 rs_disconnected = 0x1000,
245 #define RS_OPT_SWAP_SGL (1 << 0)
247 * iWarp does not support RDMA write with immediate data. For iWarp, we
248 * transfer rsocket messages as inline sends.
250 #define RS_OPT_MSG_SEND (1 << 1)
251 #define RS_OPT_SVC_ACTIVE (1 << 2)
255 struct sockaddr_in sin;
256 struct sockaddr_in6 sin6;
272 #define DS_IPV4_HDR_LEN 8
273 #define DS_IPV6_HDR_LEN 24
276 union socket_addr addr; /* must be first */
285 struct rdma_cm_id *cm_id;
286 struct ds_header hdr;
302 fastlock_t cq_wait_lock;
303 fastlock_t map_lock; /* acquire slock first if needed */
308 struct rdma_cm_id *cm_id;
310 unsigned int keepalive_time;
312 unsigned int ctrl_seqno;
313 unsigned int ctrl_max_seqno;
320 struct rs_sge remote_sgl;
321 struct rs_sge remote_iomap;
323 struct ibv_mr *target_mr;
325 int target_iomap_size;
326 void *target_buffer_list;
327 volatile struct rs_sge *target_sgl;
328 struct rs_iomap *target_iomap;
331 int rbuf_bytes_avail;
332 int rbuf_free_offset;
337 int sbuf_bytes_avail;
339 struct ibv_sge ssgl[2];
343 struct ds_qp *qp_list;
345 struct ds_dest *conn_dest;
350 struct ds_smsg *smsg_free;
376 struct ds_rmsg *dmsg;
380 struct rs_iomap_mr *remote_iomappings;
381 dlist_entry iomap_list;
382 dlist_entry iomap_queue;
387 #define DS_UDP_TAG 0x55555555
389 struct ds_udp_header {
395 __be32 qpn; /* lower 8-bits reserved */
402 #define DS_UDP_IPV4_HDR_LEN 16
403 #define DS_UDP_IPV6_HDR_LEN 28
405 #define ds_next_qp(qp) container_of((qp)->list.next, struct ds_qp, list)
407 static void write_all(int fd, const void *msg, size_t len)
409 // FIXME: if fd is a socket this really needs to handle EINTR and other conditions.
410 ssize_t rc = write(fd, msg, len);
414 static void read_all(int fd, void *msg, size_t len)
416 // FIXME: if fd is a socket this really needs to handle EINTR and other conditions.
417 ssize_t rc = read(fd, msg, len);
421 static void ds_insert_qp(struct rsocket *rs, struct ds_qp *qp)
424 dlist_init(&qp->list);
426 dlist_insert_head(&qp->list, &rs->qp_list->list);
430 static void ds_remove_qp(struct rsocket *rs, struct ds_qp *qp)
432 if (qp->list.next != &qp->list) {
433 rs->qp_list = ds_next_qp(qp);
434 dlist_remove(&qp->list);
440 static int rs_notify_svc(struct rs_svc *svc, struct rsocket *rs, int cmd)
442 struct rs_svc_msg msg;
445 pthread_mutex_lock(&mut);
447 ret = socketpair(AF_UNIX, SOCK_STREAM, 0, svc->sock);
451 ret = pthread_create(&svc->id, NULL, svc->run, svc);
461 write_all(svc->sock[0], &msg, sizeof msg);
462 read_all(svc->sock[0], &msg, sizeof msg);
463 ret = rdma_seterrno(msg.status);
467 pthread_join(svc->id, NULL);
472 pthread_mutex_unlock(&mut);
476 static int ds_compare_addr(const void *dst1, const void *dst2)
478 const struct sockaddr *sa1, *sa2;
481 sa1 = (const struct sockaddr *) dst1;
482 sa2 = (const struct sockaddr *) dst2;
484 len = (sa1->sa_family == AF_INET6 && sa2->sa_family == AF_INET6) ?
485 sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in);
486 return memcmp(dst1, dst2, len);
489 static int rs_value_to_scale(int value, int bits)
491 return value <= (1 << (bits - 1)) ?
492 value : (1 << (bits - 1)) | (value >> bits);
495 static int rs_scale_to_value(int value, int bits)
497 return value <= (1 << (bits - 1)) ?
498 value : (value & ~(1 << (bits - 1))) << bits;
501 /* gcc > ~5 will not allow (void)fscanf to suppress -Wunused-result, but this
502 will do it. In this case ignoring the result is OK (but horribly
503 unfriendly to user) since the library has a sane default. */
504 #define failable_fscanf(f, fmt, ...) \
506 int rc = fscanf(f, fmt, __VA_ARGS__); \
510 static void rs_configure(void)
518 pthread_mutex_lock(&mut);
526 if ((f = fopen(RS_CONF_DIR "/polling_time", "r"))) {
527 failable_fscanf(f, "%u", &polling_time);
531 if ((f = fopen(RS_CONF_DIR "/inline_default", "r"))) {
532 failable_fscanf(f, "%hu", &def_inline);
536 if ((f = fopen(RS_CONF_DIR "/sqsize_default", "r"))) {
537 failable_fscanf(f, "%hu", &def_sqsize);
541 if ((f = fopen(RS_CONF_DIR "/rqsize_default", "r"))) {
542 failable_fscanf(f, "%hu", &def_rqsize);
546 if ((f = fopen(RS_CONF_DIR "/mem_default", "r"))) {
547 failable_fscanf(f, "%u", &def_mem);
554 if ((f = fopen(RS_CONF_DIR "/wmem_default", "r"))) {
555 failable_fscanf(f, "%u", &def_wmem);
557 if (def_wmem < RS_SNDLOWAT)
558 def_wmem = RS_SNDLOWAT << 1;
561 if ((f = fopen(RS_CONF_DIR "/iomap_size", "r"))) {
562 failable_fscanf(f, "%hu", &def_iomap_size);
565 /* round to supported values */
566 def_iomap_size = (uint8_t) rs_value_to_scale(
567 (uint16_t) rs_scale_to_value(def_iomap_size, 8), 8);
571 pthread_mutex_unlock(&mut);
574 static int rs_insert(struct rsocket *rs, int index)
576 pthread_mutex_lock(&mut);
577 rs->index = idm_set(&idm, index, rs);
578 pthread_mutex_unlock(&mut);
582 static void rs_remove(struct rsocket *rs)
584 pthread_mutex_lock(&mut);
585 idm_clear(&idm, rs->index);
586 pthread_mutex_unlock(&mut);
589 /* We only inherit from listening sockets */
590 static struct rsocket *rs_alloc(struct rsocket *inherited_rs, int type)
594 rs = calloc(1, sizeof(*rs));
600 if (type == SOCK_DGRAM) {
606 rs->sbuf_size = inherited_rs->sbuf_size;
607 rs->rbuf_size = inherited_rs->rbuf_size;
608 rs->sq_inline = inherited_rs->sq_inline;
609 rs->sq_size = inherited_rs->sq_size;
610 rs->rq_size = inherited_rs->rq_size;
611 if (type == SOCK_STREAM) {
612 rs->ctrl_max_seqno = inherited_rs->ctrl_max_seqno;
613 rs->target_iomap_size = inherited_rs->target_iomap_size;
616 rs->sbuf_size = def_wmem;
617 rs->rbuf_size = def_mem;
618 rs->sq_inline = def_inline;
619 rs->sq_size = def_sqsize;
620 rs->rq_size = def_rqsize;
621 if (type == SOCK_STREAM) {
622 rs->ctrl_max_seqno = RS_QP_CTRL_SIZE;
623 rs->target_iomap_size = def_iomap_size;
626 fastlock_init(&rs->slock);
627 fastlock_init(&rs->rlock);
628 fastlock_init(&rs->cq_lock);
629 fastlock_init(&rs->cq_wait_lock);
630 fastlock_init(&rs->map_lock);
631 dlist_init(&rs->iomap_list);
632 dlist_init(&rs->iomap_queue);
636 static int rs_set_nonblocking(struct rsocket *rs, int arg)
641 if (rs->type == SOCK_STREAM) {
642 if (rs->cm_id->recv_cq_channel)
643 ret = fcntl(rs->cm_id->recv_cq_channel->fd, F_SETFL, arg);
645 if (!ret && rs->state < rs_connected)
646 ret = fcntl(rs->cm_id->channel->fd, F_SETFL, arg);
648 ret = fcntl(rs->epfd, F_SETFL, arg);
649 if (!ret && rs->qp_list) {
652 ret = fcntl(qp->cm_id->recv_cq_channel->fd,
655 } while (qp != rs->qp_list && !ret);
662 static void rs_set_qp_size(struct rsocket *rs)
666 max_size = min(ucma_max_qpsize(rs->cm_id), RS_QP_MAX_SIZE);
668 if (rs->sq_size > max_size)
669 rs->sq_size = max_size;
670 else if (rs->sq_size < RS_QP_MIN_SIZE)
671 rs->sq_size = RS_QP_MIN_SIZE;
673 if (rs->rq_size > max_size)
674 rs->rq_size = max_size;
675 else if (rs->rq_size < RS_QP_MIN_SIZE)
676 rs->rq_size = RS_QP_MIN_SIZE;
679 static void ds_set_qp_size(struct rsocket *rs)
683 max_size = min(ucma_max_qpsize(NULL), RS_QP_MAX_SIZE);
685 if (rs->sq_size > max_size)
686 rs->sq_size = max_size;
687 if (rs->rq_size > max_size)
688 rs->rq_size = max_size;
690 if (rs->rq_size > (rs->rbuf_size / RS_SNDLOWAT))
691 rs->rq_size = rs->rbuf_size / RS_SNDLOWAT;
693 rs->rbuf_size = rs->rq_size * RS_SNDLOWAT;
695 if (rs->sq_size > (rs->sbuf_size / RS_SNDLOWAT))
696 rs->sq_size = rs->sbuf_size / RS_SNDLOWAT;
698 rs->sbuf_size = rs->sq_size * RS_SNDLOWAT;
701 static int rs_init_bufs(struct rsocket *rs)
703 uint32_t total_rbuf_size, total_sbuf_size;
706 rs->rmsg = calloc(rs->rq_size + 1, sizeof(*rs->rmsg));
710 total_sbuf_size = rs->sbuf_size;
711 if (rs->sq_inline < RS_MAX_CTRL_MSG)
712 total_sbuf_size += RS_MAX_CTRL_MSG * RS_QP_CTRL_SIZE;
713 rs->sbuf = calloc(total_sbuf_size, 1);
717 rs->smr = rdma_reg_msgs(rs->cm_id, rs->sbuf, total_sbuf_size);
721 len = sizeof(*rs->target_sgl) * RS_SGL_SIZE +
722 sizeof(*rs->target_iomap) * rs->target_iomap_size;
723 rs->target_buffer_list = malloc(len);
724 if (!rs->target_buffer_list)
727 rs->target_mr = rdma_reg_write(rs->cm_id, rs->target_buffer_list, len);
731 memset(rs->target_buffer_list, 0, len);
732 rs->target_sgl = rs->target_buffer_list;
733 if (rs->target_iomap_size)
734 rs->target_iomap = (struct rs_iomap *) (rs->target_sgl + RS_SGL_SIZE);
736 total_rbuf_size = rs->rbuf_size;
737 if (rs->opts & RS_OPT_MSG_SEND)
738 total_rbuf_size += rs->rq_size * RS_MSG_SIZE;
739 rs->rbuf = calloc(total_rbuf_size, 1);
743 rs->rmr = rdma_reg_write(rs->cm_id, rs->rbuf, total_rbuf_size);
747 rs->ssgl[0].addr = rs->ssgl[1].addr = (uintptr_t) rs->sbuf;
748 rs->sbuf_bytes_avail = rs->sbuf_size;
749 rs->ssgl[0].lkey = rs->ssgl[1].lkey = rs->smr->lkey;
751 rs->rbuf_free_offset = rs->rbuf_size >> 1;
752 rs->rbuf_bytes_avail = rs->rbuf_size >> 1;
753 rs->sqe_avail = rs->sq_size - rs->ctrl_max_seqno;
754 rs->rseq_comp = rs->rq_size >> 1;
758 static int ds_init_bufs(struct ds_qp *qp)
760 qp->rbuf = calloc(qp->rs->rbuf_size + sizeof(struct ibv_grh), 1);
764 qp->smr = rdma_reg_msgs(qp->cm_id, qp->rs->sbuf, qp->rs->sbuf_size);
768 qp->rmr = rdma_reg_msgs(qp->cm_id, qp->rbuf, qp->rs->rbuf_size +
769 sizeof(struct ibv_grh));
777 * If a user is waiting on a datagram rsocket through poll or select, then
778 * we need the first completion to generate an event on the related epoll fd
779 * in order to signal the user. We arm the CQ on creation for this purpose
781 static int rs_create_cq(struct rsocket *rs, struct rdma_cm_id *cm_id)
783 cm_id->recv_cq_channel = ibv_create_comp_channel(cm_id->verbs);
784 if (!cm_id->recv_cq_channel)
787 cm_id->recv_cq = ibv_create_cq(cm_id->verbs, rs->sq_size + rs->rq_size,
788 cm_id, cm_id->recv_cq_channel, 0);
792 if (rs->fd_flags & O_NONBLOCK) {
793 if (fcntl(cm_id->recv_cq_channel->fd, F_SETFL, O_NONBLOCK))
797 ibv_req_notify_cq(cm_id->recv_cq, 0);
798 cm_id->send_cq_channel = cm_id->recv_cq_channel;
799 cm_id->send_cq = cm_id->recv_cq;
803 ibv_destroy_cq(cm_id->recv_cq);
804 cm_id->recv_cq = NULL;
806 ibv_destroy_comp_channel(cm_id->recv_cq_channel);
807 cm_id->recv_cq_channel = NULL;
811 static inline int rs_post_recv(struct rsocket *rs)
813 struct ibv_recv_wr wr, *bad;
817 if (!(rs->opts & RS_OPT_MSG_SEND)) {
818 wr.wr_id = rs_recv_wr_id(0);
822 wr.wr_id = rs_recv_wr_id(rs->rbuf_msg_index);
823 sge.addr = (uintptr_t) rs->rbuf + rs->rbuf_size +
824 (rs->rbuf_msg_index * RS_MSG_SIZE);
825 sge.length = RS_MSG_SIZE;
826 sge.lkey = rs->rmr->lkey;
830 if(++rs->rbuf_msg_index == rs->rq_size)
831 rs->rbuf_msg_index = 0;
834 return rdma_seterrno(ibv_post_recv(rs->cm_id->qp, &wr, &bad));
837 static inline int ds_post_recv(struct rsocket *rs, struct ds_qp *qp, uint32_t offset)
839 struct ibv_recv_wr wr, *bad;
840 struct ibv_sge sge[2];
842 sge[0].addr = (uintptr_t) qp->rbuf + rs->rbuf_size;
843 sge[0].length = sizeof(struct ibv_grh);
844 sge[0].lkey = qp->rmr->lkey;
845 sge[1].addr = (uintptr_t) qp->rbuf + offset;
846 sge[1].length = RS_SNDLOWAT;
847 sge[1].lkey = qp->rmr->lkey;
849 wr.wr_id = rs_recv_wr_id(offset);
854 return rdma_seterrno(ibv_post_recv(qp->cm_id->qp, &wr, &bad));
857 static int rs_create_ep(struct rsocket *rs)
859 struct ibv_qp_init_attr qp_attr;
863 if (rs->cm_id->verbs->device->transport_type == IBV_TRANSPORT_IWARP)
864 rs->opts |= RS_OPT_MSG_SEND;
865 ret = rs_create_cq(rs, rs->cm_id);
869 memset(&qp_attr, 0, sizeof qp_attr);
870 qp_attr.qp_context = rs;
871 qp_attr.send_cq = rs->cm_id->send_cq;
872 qp_attr.recv_cq = rs->cm_id->recv_cq;
873 qp_attr.qp_type = IBV_QPT_RC;
874 qp_attr.sq_sig_all = 1;
875 qp_attr.cap.max_send_wr = rs->sq_size;
876 qp_attr.cap.max_recv_wr = rs->rq_size;
877 qp_attr.cap.max_send_sge = 2;
878 qp_attr.cap.max_recv_sge = 1;
879 qp_attr.cap.max_inline_data = rs->sq_inline;
881 ret = rdma_create_qp(rs->cm_id, NULL, &qp_attr);
885 rs->sq_inline = qp_attr.cap.max_inline_data;
886 if ((rs->opts & RS_OPT_MSG_SEND) && (rs->sq_inline < RS_MSG_SIZE))
889 ret = rs_init_bufs(rs);
893 for (i = 0; i < rs->rq_size; i++) {
894 ret = rs_post_recv(rs);
901 static void rs_release_iomap_mr(struct rs_iomap_mr *iomr)
903 if (atomic_fetch_sub(&iomr->refcnt, 1) != 1)
906 dlist_remove(&iomr->entry);
907 ibv_dereg_mr(iomr->mr);
908 if (iomr->index >= 0)
914 static void rs_free_iomappings(struct rsocket *rs)
916 struct rs_iomap_mr *iomr;
918 while (!dlist_empty(&rs->iomap_list)) {
919 iomr = container_of(rs->iomap_list.next,
920 struct rs_iomap_mr, entry);
921 riounmap(rs->index, iomr->mr->addr, iomr->mr->length);
923 while (!dlist_empty(&rs->iomap_queue)) {
924 iomr = container_of(rs->iomap_queue.next,
925 struct rs_iomap_mr, entry);
926 riounmap(rs->index, iomr->mr->addr, iomr->mr->length);
930 static void ds_free_qp(struct ds_qp *qp)
933 rdma_dereg_mr(qp->smr);
937 rdma_dereg_mr(qp->rmr);
943 tdelete(&qp->dest.addr, &qp->rs->dest_map, ds_compare_addr);
944 epoll_ctl(qp->rs->epfd, EPOLL_CTL_DEL,
945 qp->cm_id->recv_cq_channel->fd, NULL);
946 rdma_destroy_qp(qp->cm_id);
948 rdma_destroy_id(qp->cm_id);
954 static void ds_free(struct rsocket *rs)
958 if (rs->udp_sock >= 0)
967 while ((qp = rs->qp_list)) {
968 ds_remove_qp(rs, qp);
978 tdestroy(rs->dest_map, free);
979 fastlock_destroy(&rs->map_lock);
980 fastlock_destroy(&rs->cq_wait_lock);
981 fastlock_destroy(&rs->cq_lock);
982 fastlock_destroy(&rs->rlock);
983 fastlock_destroy(&rs->slock);
987 static void rs_free(struct rsocket *rs)
989 if (rs->type == SOCK_DGRAM) {
999 rdma_dereg_mr(rs->smr);
1005 rdma_dereg_mr(rs->rmr);
1009 if (rs->target_buffer_list) {
1011 rdma_dereg_mr(rs->target_mr);
1012 free(rs->target_buffer_list);
1016 rs_free_iomappings(rs);
1017 if (rs->cm_id->qp) {
1018 ibv_ack_cq_events(rs->cm_id->recv_cq, rs->unack_cqe);
1019 rdma_destroy_qp(rs->cm_id);
1021 rdma_destroy_id(rs->cm_id);
1027 fastlock_destroy(&rs->map_lock);
1028 fastlock_destroy(&rs->cq_wait_lock);
1029 fastlock_destroy(&rs->cq_lock);
1030 fastlock_destroy(&rs->rlock);
1031 fastlock_destroy(&rs->slock);
1035 static size_t rs_conn_data_offset(struct rsocket *rs)
1037 return (rs->cm_id->route.addr.src_addr.sa_family == AF_IB) ?
1038 sizeof(struct ib_connect_hdr) : 0;
1041 static void rs_format_conn_data(struct rsocket *rs, struct rs_conn_data *conn)
1044 conn->flags = RS_CONN_FLAG_IOMAP |
1045 (rs_host_is_net() ? RS_CONN_FLAG_NET : 0);
1046 conn->credits = htobe16(rs->rq_size);
1047 memset(conn->reserved, 0, sizeof conn->reserved);
1048 conn->target_iomap_size = (uint8_t) rs_value_to_scale(rs->target_iomap_size, 8);
1050 conn->target_sgl.addr = (__force uint64_t)htobe64((uintptr_t) rs->target_sgl);
1051 conn->target_sgl.length = (__force uint32_t)htobe32(RS_SGL_SIZE);
1052 conn->target_sgl.key = (__force uint32_t)htobe32(rs->target_mr->rkey);
1054 conn->data_buf.addr = (__force uint64_t)htobe64((uintptr_t) rs->rbuf);
1055 conn->data_buf.length = (__force uint32_t)htobe32(rs->rbuf_size >> 1);
1056 conn->data_buf.key = (__force uint32_t)htobe32(rs->rmr->rkey);
1059 static void rs_save_conn_data(struct rsocket *rs, struct rs_conn_data *conn)
1061 rs->remote_sgl.addr = be64toh((__force __be64)conn->target_sgl.addr);
1062 rs->remote_sgl.length = be32toh((__force __be32)conn->target_sgl.length);
1063 rs->remote_sgl.key = be32toh((__force __be32)conn->target_sgl.key);
1065 if ((rs_host_is_net() && !(conn->flags & RS_CONN_FLAG_NET)) ||
1066 (!rs_host_is_net() && (conn->flags & RS_CONN_FLAG_NET)))
1067 rs->opts = RS_OPT_SWAP_SGL;
1069 if (conn->flags & RS_CONN_FLAG_IOMAP) {
1070 rs->remote_iomap.addr = rs->remote_sgl.addr +
1071 sizeof(rs->remote_sgl) * rs->remote_sgl.length;
1072 rs->remote_iomap.length = rs_scale_to_value(conn->target_iomap_size, 8);
1073 rs->remote_iomap.key = rs->remote_sgl.key;
1076 rs->target_sgl[0].addr = be64toh((__force __be64)conn->data_buf.addr);
1077 rs->target_sgl[0].length = be32toh((__force __be32)conn->data_buf.length);
1078 rs->target_sgl[0].key = be32toh((__force __be32)conn->data_buf.key);
1080 rs->sseq_comp = be16toh(conn->credits);
1083 static int ds_init(struct rsocket *rs, int domain)
1085 rs->udp_sock = socket(domain, SOCK_DGRAM, 0);
1086 if (rs->udp_sock < 0)
1087 return rs->udp_sock;
1089 rs->epfd = epoll_create(2);
1096 static int ds_init_ep(struct rsocket *rs)
1098 struct ds_smsg *msg;
1103 rs->sbuf = calloc(rs->sq_size, RS_SNDLOWAT);
1107 rs->dmsg = calloc(rs->rq_size + 1, sizeof(*rs->dmsg));
1111 rs->sqe_avail = rs->sq_size;
1112 rs->rqe_avail = rs->rq_size;
1114 rs->smsg_free = (struct ds_smsg *) rs->sbuf;
1115 msg = rs->smsg_free;
1116 for (i = 0; i < rs->sq_size - 1; i++) {
1117 msg->next = (void *) msg + RS_SNDLOWAT;
1122 ret = rs_notify_svc(&udp_svc, rs, RS_SVC_ADD_DGRAM);
1126 rs->state = rs_readable | rs_writable;
1130 int rsocket(int domain, int type, int protocol)
1135 if ((domain != AF_INET && domain != AF_INET6 && domain != AF_IB) ||
1136 ((type != SOCK_STREAM) && (type != SOCK_DGRAM)) ||
1137 (type == SOCK_STREAM && protocol && protocol != IPPROTO_TCP) ||
1138 (type == SOCK_DGRAM && protocol && protocol != IPPROTO_UDP))
1139 return ERR(ENOTSUP);
1142 rs = rs_alloc(NULL, type);
1146 if (type == SOCK_STREAM) {
1147 ret = rdma_create_id(NULL, &rs->cm_id, rs, RDMA_PS_TCP);
1151 rs->cm_id->route.addr.src_addr.sa_family = domain;
1152 index = rs->cm_id->channel->fd;
1154 ret = ds_init(rs, domain);
1158 index = rs->udp_sock;
1161 ret = rs_insert(rs, index);
1172 int rbind(int socket, const struct sockaddr *addr, socklen_t addrlen)
1177 rs = idm_lookup(&idm, socket);
1180 if (rs->type == SOCK_STREAM) {
1181 ret = rdma_bind_addr(rs->cm_id, (struct sockaddr *) addr);
1183 rs->state = rs_bound;
1185 if (rs->state == rs_init) {
1186 ret = ds_init_ep(rs);
1190 ret = bind(rs->udp_sock, addr, addrlen);
1195 int rlisten(int socket, int backlog)
1200 rs = idm_lookup(&idm, socket);
1204 if (rs->state != rs_listening) {
1205 ret = rdma_listen(rs->cm_id, backlog);
1207 rs->state = rs_listening;
1215 * Nonblocking is usually not inherited between sockets, but we need to
1216 * inherit it here to establish the connection only. This is needed to
1217 * prevent rdma_accept from blocking until the remote side finishes
1218 * establishing the connection. If we were to allow rdma_accept to block,
1219 * then a single thread cannot establish a connection with itself, or
1220 * two threads which try to connect to each other can deadlock trying to
1221 * form a connection.
1223 * Data transfers on the new socket remain blocking unless the user
1224 * specifies otherwise through rfcntl.
1226 int raccept(int socket, struct sockaddr *addr, socklen_t *addrlen)
1228 struct rsocket *rs, *new_rs;
1229 struct rdma_conn_param param;
1230 struct rs_conn_data *creq, cresp;
1233 rs = idm_lookup(&idm, socket);
1236 new_rs = rs_alloc(rs, rs->type);
1240 ret = rdma_get_request(rs->cm_id, &new_rs->cm_id);
1244 ret = rs_insert(new_rs, new_rs->cm_id->channel->fd);
1248 creq = (struct rs_conn_data *)
1249 (new_rs->cm_id->event->param.conn.private_data + rs_conn_data_offset(rs));
1250 if (creq->version != 1) {
1255 if (rs->fd_flags & O_NONBLOCK)
1256 fcntl(new_rs->cm_id->channel->fd, F_SETFL, O_NONBLOCK);
1258 ret = rs_create_ep(new_rs);
1262 rs_save_conn_data(new_rs, creq);
1263 param = new_rs->cm_id->event->param.conn;
1264 rs_format_conn_data(new_rs, &cresp);
1265 param.private_data = &cresp;
1266 param.private_data_len = sizeof cresp;
1267 ret = rdma_accept(new_rs->cm_id, ¶m);
1269 new_rs->state = rs_connect_rdwr;
1270 else if (errno == EAGAIN || errno == EWOULDBLOCK)
1271 new_rs->state = rs_accepting;
1275 if (addr && addrlen)
1276 rgetpeername(new_rs->index, addr, addrlen);
1277 return new_rs->index;
1284 static int rs_do_connect(struct rsocket *rs)
1286 struct rdma_conn_param param;
1287 struct rs_conn_private_data cdata;
1288 struct rs_conn_data *creq, *cresp;
1291 switch (rs->state) {
1295 to = 1000 << rs->retries++;
1296 ret = rdma_resolve_addr(rs->cm_id, NULL,
1297 &rs->cm_id->route.addr.dst_addr, to);
1300 if (errno == EAGAIN || errno == EWOULDBLOCK)
1301 rs->state = rs_resolving_addr;
1303 case rs_resolving_addr:
1304 ret = ucma_complete(rs->cm_id);
1306 if (errno == ETIMEDOUT && rs->retries <= RS_CONN_RETRIES)
1313 to = 1000 << rs->retries++;
1315 ret = rdma_set_option(rs->cm_id, RDMA_OPTION_IB,
1316 RDMA_OPTION_IB_PATH, rs->optval,
1321 rs->state = rs_resolving_route;
1322 goto resolving_route;
1325 ret = rdma_resolve_route(rs->cm_id, to);
1329 if (errno == EAGAIN || errno == EWOULDBLOCK)
1330 rs->state = rs_resolving_route;
1332 case rs_resolving_route:
1334 ret = ucma_complete(rs->cm_id);
1336 if (errno == ETIMEDOUT && rs->retries <= RS_CONN_RETRIES)
1341 ret = rs_create_ep(rs);
1345 memset(¶m, 0, sizeof param);
1346 creq = (void *) &cdata + rs_conn_data_offset(rs);
1347 rs_format_conn_data(rs, creq);
1348 param.private_data = (void *) creq - rs_conn_data_offset(rs);
1349 param.private_data_len = sizeof(*creq) + rs_conn_data_offset(rs);
1350 param.flow_control = 1;
1351 param.retry_count = 7;
1352 param.rnr_retry_count = 7;
1353 /* work-around: iWarp issues RDMA read during connection */
1354 if (rs->opts & RS_OPT_MSG_SEND)
1355 param.initiator_depth = 1;
1358 ret = rdma_connect(rs->cm_id, ¶m);
1361 if (errno == EAGAIN || errno == EWOULDBLOCK)
1362 rs->state = rs_connecting;
1365 ret = ucma_complete(rs->cm_id);
1369 cresp = (struct rs_conn_data *) rs->cm_id->event->param.conn.private_data;
1370 if (cresp->version != 1) {
1375 rs_save_conn_data(rs, cresp);
1376 rs->state = rs_connect_rdwr;
1379 if (!(rs->fd_flags & O_NONBLOCK))
1380 fcntl(rs->cm_id->channel->fd, F_SETFL, 0);
1382 ret = ucma_complete(rs->cm_id);
1386 rs->state = rs_connect_rdwr;
1394 if (errno == EAGAIN || errno == EWOULDBLOCK) {
1395 errno = EINPROGRESS;
1397 rs->state = rs_connect_error;
1404 static int rs_any_addr(const union socket_addr *addr)
1406 if (addr->sa.sa_family == AF_INET) {
1407 return (addr->sin.sin_addr.s_addr == htobe32(INADDR_ANY) ||
1408 addr->sin.sin_addr.s_addr == htobe32(INADDR_LOOPBACK));
1410 return (!memcmp(&addr->sin6.sin6_addr, &in6addr_any, 16) ||
1411 !memcmp(&addr->sin6.sin6_addr, &in6addr_loopback, 16));
1415 static int ds_get_src_addr(struct rsocket *rs,
1416 const struct sockaddr *dest_addr, socklen_t dest_len,
1417 union socket_addr *src_addr, socklen_t *src_len)
1422 *src_len = sizeof(*src_addr);
1423 ret = getsockname(rs->udp_sock, &src_addr->sa, src_len);
1424 if (ret || !rs_any_addr(src_addr))
1427 port = src_addr->sin.sin_port;
1428 sock = socket(dest_addr->sa_family, SOCK_DGRAM, 0);
1432 ret = connect(sock, dest_addr, dest_len);
1436 *src_len = sizeof(*src_addr);
1437 ret = getsockname(sock, &src_addr->sa, src_len);
1438 src_addr->sin.sin_port = port;
1444 static void ds_format_hdr(struct ds_header *hdr, union socket_addr *addr)
1446 if (addr->sa.sa_family == AF_INET) {
1448 hdr->length = DS_IPV4_HDR_LEN;
1449 hdr->port = addr->sin.sin_port;
1450 hdr->addr.ipv4 = addr->sin.sin_addr.s_addr;
1453 hdr->length = DS_IPV6_HDR_LEN;
1454 hdr->port = addr->sin6.sin6_port;
1455 hdr->addr.ipv6.flowinfo= addr->sin6.sin6_flowinfo;
1456 memcpy(&hdr->addr.ipv6.addr, &addr->sin6.sin6_addr, 16);
1460 static int ds_add_qp_dest(struct ds_qp *qp, union socket_addr *addr,
1463 struct ibv_port_attr port_attr;
1464 struct ibv_ah_attr attr;
1467 memcpy(&qp->dest.addr, addr, addrlen);
1469 qp->dest.qpn = qp->cm_id->qp->qp_num;
1471 ret = ibv_query_port(qp->cm_id->verbs, qp->cm_id->port_num, &port_attr);
1475 memset(&attr, 0, sizeof attr);
1476 attr.dlid = port_attr.lid;
1477 attr.port_num = qp->cm_id->port_num;
1478 qp->dest.ah = ibv_create_ah(qp->cm_id->pd, &attr);
1482 tsearch(&qp->dest.addr, &qp->rs->dest_map, ds_compare_addr);
1486 static int ds_create_qp(struct rsocket *rs, union socket_addr *src_addr,
1487 socklen_t addrlen, struct ds_qp **new_qp)
1490 struct ibv_qp_init_attr qp_attr;
1491 struct epoll_event event;
1494 qp = calloc(1, sizeof(*qp));
1499 ret = rdma_create_id(NULL, &qp->cm_id, qp, RDMA_PS_UDP);
1503 ds_format_hdr(&qp->hdr, src_addr);
1504 ret = rdma_bind_addr(qp->cm_id, &src_addr->sa);
1508 ret = ds_init_bufs(qp);
1512 ret = rs_create_cq(rs, qp->cm_id);
1516 memset(&qp_attr, 0, sizeof qp_attr);
1517 qp_attr.qp_context = qp;
1518 qp_attr.send_cq = qp->cm_id->send_cq;
1519 qp_attr.recv_cq = qp->cm_id->recv_cq;
1520 qp_attr.qp_type = IBV_QPT_UD;
1521 qp_attr.sq_sig_all = 1;
1522 qp_attr.cap.max_send_wr = rs->sq_size;
1523 qp_attr.cap.max_recv_wr = rs->rq_size;
1524 qp_attr.cap.max_send_sge = 1;
1525 qp_attr.cap.max_recv_sge = 2;
1526 qp_attr.cap.max_inline_data = rs->sq_inline;
1527 ret = rdma_create_qp(qp->cm_id, NULL, &qp_attr);
1531 rs->sq_inline = qp_attr.cap.max_inline_data;
1532 ret = ds_add_qp_dest(qp, src_addr, addrlen);
1536 event.events = EPOLLIN;
1537 event.data.ptr = qp;
1538 ret = epoll_ctl(rs->epfd, EPOLL_CTL_ADD,
1539 qp->cm_id->recv_cq_channel->fd, &event);
1543 for (i = 0; i < rs->rq_size; i++) {
1544 ret = ds_post_recv(rs, qp, i * RS_SNDLOWAT);
1549 ds_insert_qp(rs, qp);
1557 static int ds_get_qp(struct rsocket *rs, union socket_addr *src_addr,
1558 socklen_t addrlen, struct ds_qp **qp)
1563 if (!ds_compare_addr(rdma_get_local_addr((*qp)->cm_id),
1567 *qp = ds_next_qp(*qp);
1568 } while (*qp != rs->qp_list);
1571 return ds_create_qp(rs, src_addr, addrlen, qp);
1574 static int ds_get_dest(struct rsocket *rs, const struct sockaddr *addr,
1575 socklen_t addrlen, struct ds_dest **dest)
1577 union socket_addr src_addr;
1580 struct ds_dest **tdest, *new_dest;
1583 fastlock_acquire(&rs->map_lock);
1584 tdest = tfind(addr, &rs->dest_map, ds_compare_addr);
1588 ret = ds_get_src_addr(rs, addr, addrlen, &src_addr, &src_len);
1592 ret = ds_get_qp(rs, &src_addr, src_len, &qp);
1596 tdest = tfind(addr, &rs->dest_map, ds_compare_addr);
1598 new_dest = calloc(1, sizeof(*new_dest));
1604 memcpy(&new_dest->addr, addr, addrlen);
1606 tdest = tsearch(&new_dest->addr, &rs->dest_map, ds_compare_addr);
1612 fastlock_release(&rs->map_lock);
1616 int rconnect(int socket, const struct sockaddr *addr, socklen_t addrlen)
1621 rs = idm_lookup(&idm, socket);
1624 if (rs->type == SOCK_STREAM) {
1625 memcpy(&rs->cm_id->route.addr.dst_addr, addr, addrlen);
1626 ret = rs_do_connect(rs);
1628 if (rs->state == rs_init) {
1629 ret = ds_init_ep(rs);
1634 fastlock_acquire(&rs->slock);
1635 ret = connect(rs->udp_sock, addr, addrlen);
1637 ret = ds_get_dest(rs, addr, addrlen, &rs->conn_dest);
1638 fastlock_release(&rs->slock);
1643 static void *rs_get_ctrl_buf(struct rsocket *rs)
1645 return rs->sbuf + rs->sbuf_size +
1646 RS_MAX_CTRL_MSG * (rs->ctrl_seqno & (RS_QP_CTRL_SIZE - 1));
1649 static int rs_post_msg(struct rsocket *rs, uint32_t msg)
1651 struct ibv_send_wr wr, *bad;
1654 wr.wr_id = rs_send_wr_id(msg);
1656 if (!(rs->opts & RS_OPT_MSG_SEND)) {
1659 wr.opcode = IBV_WR_RDMA_WRITE_WITH_IMM;
1661 wr.imm_data = htobe32(msg);
1663 sge.addr = (uintptr_t) &msg;
1665 sge.length = sizeof msg;
1668 wr.opcode = IBV_WR_SEND;
1669 wr.send_flags = IBV_SEND_INLINE;
1672 return rdma_seterrno(ibv_post_send(rs->cm_id->qp, &wr, &bad));
1675 static int rs_post_write(struct rsocket *rs,
1676 struct ibv_sge *sgl, int nsge,
1677 uint32_t wr_data, int flags,
1678 uint64_t addr, uint32_t rkey)
1680 struct ibv_send_wr wr, *bad;
1682 wr.wr_id = rs_send_wr_id(wr_data);
1686 wr.opcode = IBV_WR_RDMA_WRITE;
1687 wr.send_flags = flags;
1688 wr.wr.rdma.remote_addr = addr;
1689 wr.wr.rdma.rkey = rkey;
1691 return rdma_seterrno(ibv_post_send(rs->cm_id->qp, &wr, &bad));
1694 static int rs_post_write_msg(struct rsocket *rs,
1695 struct ibv_sge *sgl, int nsge,
1696 uint32_t msg, int flags,
1697 uint64_t addr, uint32_t rkey)
1699 struct ibv_send_wr wr, *bad;
1704 if (!(rs->opts & RS_OPT_MSG_SEND)) {
1705 wr.wr_id = rs_send_wr_id(msg);
1708 wr.opcode = IBV_WR_RDMA_WRITE_WITH_IMM;
1709 wr.send_flags = flags;
1710 wr.imm_data = htobe32(msg);
1711 wr.wr.rdma.remote_addr = addr;
1712 wr.wr.rdma.rkey = rkey;
1714 return rdma_seterrno(ibv_post_send(rs->cm_id->qp, &wr, &bad));
1716 ret = rs_post_write(rs, sgl, nsge, msg, flags, addr, rkey);
1718 wr.wr_id = rs_send_wr_id(rs_msg_set(rs_msg_op(msg), 0)) |
1719 RS_WR_ID_FLAG_MSG_SEND;
1720 sge.addr = (uintptr_t) &msg;
1722 sge.length = sizeof msg;
1725 wr.opcode = IBV_WR_SEND;
1726 wr.send_flags = IBV_SEND_INLINE;
1728 ret = rdma_seterrno(ibv_post_send(rs->cm_id->qp, &wr, &bad));
1734 static int ds_post_send(struct rsocket *rs, struct ibv_sge *sge,
1737 struct ibv_send_wr wr, *bad;
1739 wr.wr_id = rs_send_wr_id(wr_data);
1743 wr.opcode = IBV_WR_SEND;
1744 wr.send_flags = (sge->length <= rs->sq_inline) ? IBV_SEND_INLINE : 0;
1745 wr.wr.ud.ah = rs->conn_dest->ah;
1746 wr.wr.ud.remote_qpn = rs->conn_dest->qpn;
1747 wr.wr.ud.remote_qkey = RDMA_UDP_QKEY;
1749 return rdma_seterrno(ibv_post_send(rs->conn_dest->qp->cm_id->qp, &wr, &bad));
1753 * Update target SGE before sending data. Otherwise the remote side may
1754 * update the entry before we do.
1756 static int rs_write_data(struct rsocket *rs,
1757 struct ibv_sge *sgl, int nsge,
1758 uint32_t length, int flags)
1765 if (rs->opts & RS_OPT_MSG_SEND)
1767 rs->sbuf_bytes_avail -= length;
1769 addr = rs->target_sgl[rs->target_sge].addr;
1770 rkey = rs->target_sgl[rs->target_sge].key;
1772 rs->target_sgl[rs->target_sge].addr += length;
1773 rs->target_sgl[rs->target_sge].length -= length;
1775 if (!rs->target_sgl[rs->target_sge].length) {
1776 if (++rs->target_sge == RS_SGL_SIZE)
1780 return rs_post_write_msg(rs, sgl, nsge, rs_msg_set(RS_OP_DATA, length),
1784 static int rs_write_direct(struct rsocket *rs, struct rs_iomap *iom, uint64_t offset,
1785 struct ibv_sge *sgl, int nsge, uint32_t length, int flags)
1790 rs->sbuf_bytes_avail -= length;
1792 addr = iom->sge.addr + offset - iom->offset;
1793 return rs_post_write(rs, sgl, nsge, rs_msg_set(RS_OP_WRITE, length),
1794 flags, addr, iom->sge.key);
1797 static int rs_write_iomap(struct rsocket *rs, struct rs_iomap_mr *iomr,
1798 struct ibv_sge *sgl, int nsge, int flags)
1804 if (rs->opts & RS_OPT_MSG_SEND)
1806 rs->sbuf_bytes_avail -= sizeof(struct rs_iomap);
1808 addr = rs->remote_iomap.addr + iomr->index * sizeof(struct rs_iomap);
1809 return rs_post_write_msg(rs, sgl, nsge, rs_msg_set(RS_OP_IOMAP_SGL, iomr->index),
1810 flags, addr, rs->remote_iomap.key);
1813 static uint32_t rs_sbuf_left(struct rsocket *rs)
1815 return (uint32_t) (((uint64_t) (uintptr_t) &rs->sbuf[rs->sbuf_size]) -
1819 static void rs_send_credits(struct rsocket *rs)
1821 struct ibv_sge ibsge;
1822 struct rs_sge sge, *sge_buf;
1826 rs->rseq_comp = rs->rseq_no + (rs->rq_size >> 1);
1827 if (rs->rbuf_bytes_avail >= (rs->rbuf_size >> 1)) {
1828 if (rs->opts & RS_OPT_MSG_SEND)
1831 if (!(rs->opts & RS_OPT_SWAP_SGL)) {
1832 sge.addr = (uintptr_t) &rs->rbuf[rs->rbuf_free_offset];
1833 sge.key = rs->rmr->rkey;
1834 sge.length = rs->rbuf_size >> 1;
1836 sge.addr = bswap_64((uintptr_t) &rs->rbuf[rs->rbuf_free_offset]);
1837 sge.key = bswap_32(rs->rmr->rkey);
1838 sge.length = bswap_32(rs->rbuf_size >> 1);
1841 if (rs->sq_inline < sizeof sge) {
1842 sge_buf = rs_get_ctrl_buf(rs);
1843 memcpy(sge_buf, &sge, sizeof sge);
1844 ibsge.addr = (uintptr_t) sge_buf;
1845 ibsge.lkey = rs->smr->lkey;
1848 ibsge.addr = (uintptr_t) &sge;
1850 flags = IBV_SEND_INLINE;
1852 ibsge.length = sizeof(sge);
1854 rs_post_write_msg(rs, &ibsge, 1,
1855 rs_msg_set(RS_OP_SGL, rs->rseq_no + rs->rq_size), flags,
1856 rs->remote_sgl.addr + rs->remote_sge * sizeof(struct rs_sge),
1857 rs->remote_sgl.key);
1859 rs->rbuf_bytes_avail -= rs->rbuf_size >> 1;
1860 rs->rbuf_free_offset += rs->rbuf_size >> 1;
1861 if (rs->rbuf_free_offset >= rs->rbuf_size)
1862 rs->rbuf_free_offset = 0;
1863 if (++rs->remote_sge == rs->remote_sgl.length)
1866 rs_post_msg(rs, rs_msg_set(RS_OP_SGL, rs->rseq_no + rs->rq_size));
1870 static inline int rs_ctrl_avail(struct rsocket *rs)
1872 return rs->ctrl_seqno != rs->ctrl_max_seqno;
1875 /* Protocols that do not support RDMA write with immediate may require 2 msgs */
1876 static inline int rs_2ctrl_avail(struct rsocket *rs)
1878 return (int)((rs->ctrl_seqno + 1) - rs->ctrl_max_seqno) < 0;
1881 static int rs_give_credits(struct rsocket *rs)
1883 if (!(rs->opts & RS_OPT_MSG_SEND)) {
1884 return ((rs->rbuf_bytes_avail >= (rs->rbuf_size >> 1)) ||
1885 ((short) ((short) rs->rseq_no - (short) rs->rseq_comp) >= 0)) &&
1886 rs_ctrl_avail(rs) && (rs->state & rs_connected);
1888 return ((rs->rbuf_bytes_avail >= (rs->rbuf_size >> 1)) ||
1889 ((short) ((short) rs->rseq_no - (short) rs->rseq_comp) >= 0)) &&
1890 rs_2ctrl_avail(rs) && (rs->state & rs_connected);
1894 static void rs_update_credits(struct rsocket *rs)
1896 if (rs_give_credits(rs))
1897 rs_send_credits(rs);
1900 static int rs_poll_cq(struct rsocket *rs)
1906 while ((ret = ibv_poll_cq(rs->cm_id->recv_cq, 1, &wc)) > 0) {
1907 if (rs_wr_is_recv(wc.wr_id)) {
1908 if (wc.status != IBV_WC_SUCCESS)
1912 if (wc.wc_flags & IBV_WC_WITH_IMM) {
1913 msg = be32toh(wc.imm_data);
1915 msg = ((uint32_t *) (rs->rbuf + rs->rbuf_size))
1916 [rs_wr_data(wc.wr_id)];
1919 switch (rs_msg_op(msg)) {
1921 rs->sseq_comp = (uint16_t) rs_msg_data(msg);
1923 case RS_OP_IOMAP_SGL:
1924 /* The iomap was updated, that's nice to know. */
1927 if (rs_msg_data(msg) == RS_CTRL_DISCONNECT) {
1928 rs->state = rs_disconnected;
1930 } else if (rs_msg_data(msg) == RS_CTRL_SHUTDOWN) {
1931 if (rs->state & rs_writable) {
1932 rs->state &= ~rs_readable;
1934 rs->state = rs_disconnected;
1940 /* We really shouldn't be here. */
1943 rs->rmsg[rs->rmsg_tail].op = rs_msg_op(msg);
1944 rs->rmsg[rs->rmsg_tail].data = rs_msg_data(msg);
1945 if (++rs->rmsg_tail == rs->rq_size + 1)
1950 switch (rs_msg_op(rs_wr_data(wc.wr_id))) {
1952 rs->ctrl_max_seqno++;
1955 rs->ctrl_max_seqno++;
1956 if (rs_msg_data(rs_wr_data(wc.wr_id)) == RS_CTRL_DISCONNECT)
1957 rs->state = rs_disconnected;
1959 case RS_OP_IOMAP_SGL:
1961 if (!rs_wr_is_msg_send(wc.wr_id))
1962 rs->sbuf_bytes_avail += sizeof(struct rs_iomap);
1966 rs->sbuf_bytes_avail += rs_msg_data(rs_wr_data(wc.wr_id));
1969 if (wc.status != IBV_WC_SUCCESS && (rs->state & rs_connected)) {
1970 rs->state = rs_error;
1976 if (rs->state & rs_connected) {
1977 while (!ret && rcnt--)
1978 ret = rs_post_recv(rs);
1981 rs->state = rs_error;
1988 static int rs_get_cq_event(struct rsocket *rs)
1997 ret = ibv_get_cq_event(rs->cm_id->recv_cq_channel, &cq, &context);
1999 if (++rs->unack_cqe >= rs->sq_size + rs->rq_size) {
2000 ibv_ack_cq_events(rs->cm_id->recv_cq, rs->unack_cqe);
2004 } else if (!(errno == EAGAIN || errno == EINTR)) {
2005 rs->state = rs_error;
2012 * Although we serialize rsend and rrecv calls with respect to themselves,
2013 * both calls may run simultaneously and need to poll the CQ for completions.
2014 * We need to serialize access to the CQ, but rsend and rrecv need to
2015 * allow each other to make forward progress.
2017 * For example, rsend may need to wait for credits from the remote side,
2018 * which could be stalled until the remote process calls rrecv. This should
2019 * not block rrecv from receiving data from the remote side however.
2021 * We handle this by using two locks. The cq_lock protects against polling
2022 * the CQ and processing completions. The cq_wait_lock serializes access to
2023 * waiting on the CQ.
2025 static int rs_process_cq(struct rsocket *rs, int nonblock, int (*test)(struct rsocket *rs))
2029 fastlock_acquire(&rs->cq_lock);
2031 rs_update_credits(rs);
2032 ret = rs_poll_cq(rs);
2038 } else if (nonblock) {
2039 ret = ERR(EWOULDBLOCK);
2040 } else if (!rs->cq_armed) {
2041 ibv_req_notify_cq(rs->cm_id->recv_cq, 0);
2044 rs_update_credits(rs);
2045 fastlock_acquire(&rs->cq_wait_lock);
2046 fastlock_release(&rs->cq_lock);
2048 ret = rs_get_cq_event(rs);
2049 fastlock_release(&rs->cq_wait_lock);
2050 fastlock_acquire(&rs->cq_lock);
2054 rs_update_credits(rs);
2055 fastlock_release(&rs->cq_lock);
2059 static int rs_get_comp(struct rsocket *rs, int nonblock, int (*test)(struct rsocket *rs))
2061 struct timeval s, e;
2062 uint32_t poll_time = 0;
2066 ret = rs_process_cq(rs, 1, test);
2067 if (!ret || nonblock || errno != EWOULDBLOCK)
2071 gettimeofday(&s, NULL);
2073 gettimeofday(&e, NULL);
2074 poll_time = (e.tv_sec - s.tv_sec) * 1000000 +
2075 (e.tv_usec - s.tv_usec) + 1;
2076 } while (poll_time <= polling_time);
2078 ret = rs_process_cq(rs, 0, test);
2082 static int ds_valid_recv(struct ds_qp *qp, struct ibv_wc *wc)
2084 struct ds_header *hdr;
2086 hdr = (struct ds_header *) (qp->rbuf + rs_wr_data(wc->wr_id));
2087 return ((wc->byte_len >= sizeof(struct ibv_grh) + DS_IPV4_HDR_LEN) &&
2088 ((hdr->version == 4 && hdr->length == DS_IPV4_HDR_LEN) ||
2089 (hdr->version == 6 && hdr->length == DS_IPV6_HDR_LEN)));
2093 * Poll all CQs associated with a datagram rsocket. We need to drop any
2094 * received messages that we do not have room to store. To limit drops,
2095 * we only poll if we have room to store the receive or we need a send
2096 * buffer. To ensure fairness, we poll the CQs round robin, remembering
2097 * where we left off.
2099 static void ds_poll_cqs(struct rsocket *rs)
2102 struct ds_smsg *smsg;
2103 struct ds_rmsg *rmsg;
2107 if (!(qp = rs->qp_list))
2113 ret = ibv_poll_cq(qp->cm_id->recv_cq, 1, &wc);
2115 qp = ds_next_qp(qp);
2119 if (rs_wr_is_recv(wc.wr_id)) {
2120 if (rs->rqe_avail && wc.status == IBV_WC_SUCCESS &&
2121 ds_valid_recv(qp, &wc)) {
2123 rmsg = &rs->dmsg[rs->rmsg_tail];
2125 rmsg->offset = rs_wr_data(wc.wr_id);
2126 rmsg->length = wc.byte_len - sizeof(struct ibv_grh);
2127 if (++rs->rmsg_tail == rs->rq_size + 1)
2130 ds_post_recv(rs, qp, rs_wr_data(wc.wr_id));
2133 smsg = (struct ds_smsg *) (rs->sbuf + rs_wr_data(wc.wr_id));
2134 smsg->next = rs->smsg_free;
2135 rs->smsg_free = smsg;
2139 qp = ds_next_qp(qp);
2140 if (!rs->rqe_avail && rs->sqe_avail) {
2145 } while (qp != rs->qp_list);
2149 static void ds_req_notify_cqs(struct rsocket *rs)
2153 if (!(qp = rs->qp_list))
2157 if (!qp->cq_armed) {
2158 ibv_req_notify_cq(qp->cm_id->recv_cq, 0);
2161 qp = ds_next_qp(qp);
2162 } while (qp != rs->qp_list);
2165 static int ds_get_cq_event(struct rsocket *rs)
2167 struct epoll_event event;
2176 ret = epoll_wait(rs->epfd, &event, 1, -1);
2180 qp = event.data.ptr;
2181 ret = ibv_get_cq_event(qp->cm_id->recv_cq_channel, &cq, &context);
2183 ibv_ack_cq_events(qp->cm_id->recv_cq, 1);
2191 static int ds_process_cqs(struct rsocket *rs, int nonblock, int (*test)(struct rsocket *rs))
2195 fastlock_acquire(&rs->cq_lock);
2201 } else if (nonblock) {
2202 ret = ERR(EWOULDBLOCK);
2203 } else if (!rs->cq_armed) {
2204 ds_req_notify_cqs(rs);
2207 fastlock_acquire(&rs->cq_wait_lock);
2208 fastlock_release(&rs->cq_lock);
2210 ret = ds_get_cq_event(rs);
2211 fastlock_release(&rs->cq_wait_lock);
2212 fastlock_acquire(&rs->cq_lock);
2216 fastlock_release(&rs->cq_lock);
2220 static int ds_get_comp(struct rsocket *rs, int nonblock, int (*test)(struct rsocket *rs))
2222 struct timeval s, e;
2223 uint32_t poll_time = 0;
2227 ret = ds_process_cqs(rs, 1, test);
2228 if (!ret || nonblock || errno != EWOULDBLOCK)
2232 gettimeofday(&s, NULL);
2234 gettimeofday(&e, NULL);
2235 poll_time = (e.tv_sec - s.tv_sec) * 1000000 +
2236 (e.tv_usec - s.tv_usec) + 1;
2237 } while (poll_time <= polling_time);
2239 ret = ds_process_cqs(rs, 0, test);
2243 static int rs_nonblocking(struct rsocket *rs, int flags)
2245 return (rs->fd_flags & O_NONBLOCK) || (flags & MSG_DONTWAIT);
2248 static int rs_is_cq_armed(struct rsocket *rs)
2250 return rs->cq_armed;
2253 static int rs_poll_all(struct rsocket *rs)
2259 * We use hardware flow control to prevent over running the remote
2260 * receive queue. However, data transfers still require space in
2261 * the remote rmsg queue, or we risk losing notification that data
2262 * has been transfered.
2264 * Be careful with race conditions in the check below. The target SGL
2265 * may be updated by a remote RDMA write.
2267 static int rs_can_send(struct rsocket *rs)
2269 if (!(rs->opts & RS_OPT_MSG_SEND)) {
2270 return rs->sqe_avail && (rs->sbuf_bytes_avail >= RS_SNDLOWAT) &&
2271 (rs->sseq_no != rs->sseq_comp) &&
2272 (rs->target_sgl[rs->target_sge].length != 0);
2274 return (rs->sqe_avail >= 2) && (rs->sbuf_bytes_avail >= RS_SNDLOWAT) &&
2275 (rs->sseq_no != rs->sseq_comp) &&
2276 (rs->target_sgl[rs->target_sge].length != 0);
2280 static int ds_can_send(struct rsocket *rs)
2282 return rs->sqe_avail;
2285 static int ds_all_sends_done(struct rsocket *rs)
2287 return rs->sqe_avail == rs->sq_size;
2290 static int rs_conn_can_send(struct rsocket *rs)
2292 return rs_can_send(rs) || !(rs->state & rs_writable);
2295 static int rs_conn_can_send_ctrl(struct rsocket *rs)
2297 return rs_ctrl_avail(rs) || !(rs->state & rs_connected);
2300 static int rs_have_rdata(struct rsocket *rs)
2302 return (rs->rmsg_head != rs->rmsg_tail);
2305 static int rs_conn_have_rdata(struct rsocket *rs)
2307 return rs_have_rdata(rs) || !(rs->state & rs_readable);
2310 static int rs_conn_all_sends_done(struct rsocket *rs)
2312 return ((((int) rs->ctrl_max_seqno) - ((int) rs->ctrl_seqno)) +
2313 rs->sqe_avail == rs->sq_size) ||
2314 !(rs->state & rs_connected);
2317 static void ds_set_src(struct sockaddr *addr, socklen_t *addrlen,
2318 struct ds_header *hdr)
2320 union socket_addr sa;
2322 memset(&sa, 0, sizeof sa);
2323 if (hdr->version == 4) {
2324 if (*addrlen > sizeof(sa.sin))
2325 *addrlen = sizeof(sa.sin);
2327 sa.sin.sin_family = AF_INET;
2328 sa.sin.sin_port = hdr->port;
2329 sa.sin.sin_addr.s_addr = hdr->addr.ipv4;
2331 if (*addrlen > sizeof(sa.sin6))
2332 *addrlen = sizeof(sa.sin6);
2334 sa.sin6.sin6_family = AF_INET6;
2335 sa.sin6.sin6_port = hdr->port;
2336 sa.sin6.sin6_flowinfo = hdr->addr.ipv6.flowinfo;
2337 memcpy(&sa.sin6.sin6_addr, &hdr->addr.ipv6.addr, 16);
2339 memcpy(addr, &sa, *addrlen);
2342 static ssize_t ds_recvfrom(struct rsocket *rs, void *buf, size_t len, int flags,
2343 struct sockaddr *src_addr, socklen_t *addrlen)
2345 struct ds_rmsg *rmsg;
2346 struct ds_header *hdr;
2349 if (!(rs->state & rs_readable))
2352 if (!rs_have_rdata(rs)) {
2353 ret = ds_get_comp(rs, rs_nonblocking(rs, flags),
2359 rmsg = &rs->dmsg[rs->rmsg_head];
2360 hdr = (struct ds_header *) (rmsg->qp->rbuf + rmsg->offset);
2361 if (len > rmsg->length - hdr->length)
2362 len = rmsg->length - hdr->length;
2364 memcpy(buf, (void *) hdr + hdr->length, len);
2366 ds_set_src(src_addr, addrlen, hdr);
2368 if (!(flags & MSG_PEEK)) {
2369 ds_post_recv(rs, rmsg->qp, rmsg->offset);
2370 if (++rs->rmsg_head == rs->rq_size + 1)
2378 static ssize_t rs_peek(struct rsocket *rs, void *buf, size_t len)
2381 uint32_t end_size, rsize;
2382 int rmsg_head, rbuf_offset;
2384 rmsg_head = rs->rmsg_head;
2385 rbuf_offset = rs->rbuf_offset;
2387 for (; left && (rmsg_head != rs->rmsg_tail); left -= rsize) {
2388 if (left < rs->rmsg[rmsg_head].data) {
2391 rsize = rs->rmsg[rmsg_head].data;
2392 if (++rmsg_head == rs->rq_size + 1)
2396 end_size = rs->rbuf_size - rbuf_offset;
2397 if (rsize > end_size) {
2398 memcpy(buf, &rs->rbuf[rbuf_offset], end_size);
2404 memcpy(buf, &rs->rbuf[rbuf_offset], rsize);
2405 rbuf_offset += rsize;
2413 * Continue to receive any queued data even if the remote side has disconnected.
2415 ssize_t rrecv(int socket, void *buf, size_t len, int flags)
2419 uint32_t end_size, rsize;
2422 rs = idm_at(&idm, socket);
2423 if (rs->type == SOCK_DGRAM) {
2424 fastlock_acquire(&rs->rlock);
2425 ret = ds_recvfrom(rs, buf, len, flags, NULL, NULL);
2426 fastlock_release(&rs->rlock);
2430 if (rs->state & rs_opening) {
2431 ret = rs_do_connect(rs);
2433 if (errno == EINPROGRESS)
2438 fastlock_acquire(&rs->rlock);
2440 if (!rs_have_rdata(rs)) {
2441 ret = rs_get_comp(rs, rs_nonblocking(rs, flags),
2442 rs_conn_have_rdata);
2447 if (flags & MSG_PEEK) {
2448 left = len - rs_peek(rs, buf, left);
2452 for (; left && rs_have_rdata(rs); left -= rsize) {
2453 if (left < rs->rmsg[rs->rmsg_head].data) {
2455 rs->rmsg[rs->rmsg_head].data -= left;
2458 rsize = rs->rmsg[rs->rmsg_head].data;
2459 if (++rs->rmsg_head == rs->rq_size + 1)
2463 end_size = rs->rbuf_size - rs->rbuf_offset;
2464 if (rsize > end_size) {
2465 memcpy(buf, &rs->rbuf[rs->rbuf_offset], end_size);
2466 rs->rbuf_offset = 0;
2470 rs->rbuf_bytes_avail += end_size;
2472 memcpy(buf, &rs->rbuf[rs->rbuf_offset], rsize);
2473 rs->rbuf_offset += rsize;
2475 rs->rbuf_bytes_avail += rsize;
2478 } while (left && (flags & MSG_WAITALL) && (rs->state & rs_readable));
2480 fastlock_release(&rs->rlock);
2481 return (ret && left == len) ? ret : len - left;
2484 ssize_t rrecvfrom(int socket, void *buf, size_t len, int flags,
2485 struct sockaddr *src_addr, socklen_t *addrlen)
2490 rs = idm_at(&idm, socket);
2491 if (rs->type == SOCK_DGRAM) {
2492 fastlock_acquire(&rs->rlock);
2493 ret = ds_recvfrom(rs, buf, len, flags, src_addr, addrlen);
2494 fastlock_release(&rs->rlock);
2498 ret = rrecv(socket, buf, len, flags);
2499 if (ret > 0 && src_addr)
2500 rgetpeername(socket, src_addr, addrlen);
2506 * Simple, straightforward implementation for now that only tries to fill
2507 * in the first vector.
2509 static ssize_t rrecvv(int socket, const struct iovec *iov, int iovcnt, int flags)
2511 return rrecv(socket, iov[0].iov_base, iov[0].iov_len, flags);
2514 ssize_t rrecvmsg(int socket, struct msghdr *msg, int flags)
2516 if (msg->msg_control && msg->msg_controllen)
2517 return ERR(ENOTSUP);
2519 return rrecvv(socket, msg->msg_iov, (int) msg->msg_iovlen, msg->msg_flags);
2522 ssize_t rread(int socket, void *buf, size_t count)
2524 return rrecv(socket, buf, count, 0);
2527 ssize_t rreadv(int socket, const struct iovec *iov, int iovcnt)
2529 return rrecvv(socket, iov, iovcnt, 0);
2532 static int rs_send_iomaps(struct rsocket *rs, int flags)
2534 struct rs_iomap_mr *iomr;
2536 struct rs_iomap iom;
2539 fastlock_acquire(&rs->map_lock);
2540 while (!dlist_empty(&rs->iomap_queue)) {
2541 if (!rs_can_send(rs)) {
2542 ret = rs_get_comp(rs, rs_nonblocking(rs, flags),
2546 if (!(rs->state & rs_writable)) {
2547 ret = ERR(ECONNRESET);
2552 iomr = container_of(rs->iomap_queue.next, struct rs_iomap_mr, entry);
2553 if (!(rs->opts & RS_OPT_SWAP_SGL)) {
2554 iom.offset = iomr->offset;
2555 iom.sge.addr = (uintptr_t) iomr->mr->addr;
2556 iom.sge.length = iomr->mr->length;
2557 iom.sge.key = iomr->mr->rkey;
2559 iom.offset = bswap_64(iomr->offset);
2560 iom.sge.addr = bswap_64((uintptr_t) iomr->mr->addr);
2561 iom.sge.length = bswap_32(iomr->mr->length);
2562 iom.sge.key = bswap_32(iomr->mr->rkey);
2565 if (rs->sq_inline >= sizeof iom) {
2566 sge.addr = (uintptr_t) &iom;
2567 sge.length = sizeof iom;
2569 ret = rs_write_iomap(rs, iomr, &sge, 1, IBV_SEND_INLINE);
2570 } else if (rs_sbuf_left(rs) >= sizeof iom) {
2571 memcpy((void *) (uintptr_t) rs->ssgl[0].addr, &iom, sizeof iom);
2572 rs->ssgl[0].length = sizeof iom;
2573 ret = rs_write_iomap(rs, iomr, rs->ssgl, 1, 0);
2574 if (rs_sbuf_left(rs) > sizeof iom)
2575 rs->ssgl[0].addr += sizeof iom;
2577 rs->ssgl[0].addr = (uintptr_t) rs->sbuf;
2579 rs->ssgl[0].length = rs_sbuf_left(rs);
2580 memcpy((void *) (uintptr_t) rs->ssgl[0].addr, &iom,
2581 rs->ssgl[0].length);
2582 rs->ssgl[1].length = sizeof iom - rs->ssgl[0].length;
2583 memcpy(rs->sbuf, ((void *) &iom) + rs->ssgl[0].length,
2584 rs->ssgl[1].length);
2585 ret = rs_write_iomap(rs, iomr, rs->ssgl, 2, 0);
2586 rs->ssgl[0].addr = (uintptr_t) rs->sbuf + rs->ssgl[1].length;
2588 dlist_remove(&iomr->entry);
2589 dlist_insert_tail(&iomr->entry, &rs->iomap_list);
2594 rs->iomap_pending = !dlist_empty(&rs->iomap_queue);
2595 fastlock_release(&rs->map_lock);
2599 static ssize_t ds_sendv_udp(struct rsocket *rs, const struct iovec *iov,
2600 int iovcnt, int flags, uint8_t op)
2602 struct ds_udp_header hdr;
2604 struct iovec miov[8];
2608 return ERR(ENOTSUP);
2610 hdr.tag = htobe32(DS_UDP_TAG);
2611 hdr.version = rs->conn_dest->qp->hdr.version;
2614 hdr.qpn = htobe32(rs->conn_dest->qp->cm_id->qp->qp_num & 0xFFFFFF);
2615 if (rs->conn_dest->qp->hdr.version == 4) {
2616 hdr.length = DS_UDP_IPV4_HDR_LEN;
2617 hdr.addr.ipv4 = rs->conn_dest->qp->hdr.addr.ipv4;
2619 hdr.length = DS_UDP_IPV6_HDR_LEN;
2620 memcpy(hdr.addr.ipv6, &rs->conn_dest->qp->hdr.addr.ipv6, 16);
2623 miov[0].iov_base = &hdr;
2624 miov[0].iov_len = hdr.length;
2626 memcpy(&miov[1], iov, sizeof(*iov) * iovcnt);
2628 memset(&msg, 0, sizeof msg);
2629 msg.msg_name = &rs->conn_dest->addr;
2630 msg.msg_namelen = ucma_addrlen(&rs->conn_dest->addr.sa);
2632 msg.msg_iovlen = iovcnt + 1;
2633 ret = sendmsg(rs->udp_sock, &msg, flags);
2634 return ret > 0 ? ret - hdr.length : ret;
2637 static ssize_t ds_send_udp(struct rsocket *rs, const void *buf, size_t len,
2638 int flags, uint8_t op)
2642 iov.iov_base = (void *) buf;
2644 return ds_sendv_udp(rs, &iov, 1, flags, op);
2646 return ds_sendv_udp(rs, NULL, 0, flags, op);
2650 static ssize_t dsend(struct rsocket *rs, const void *buf, size_t len, int flags)
2652 struct ds_smsg *msg;
2657 if (!rs->conn_dest->ah)
2658 return ds_send_udp(rs, buf, len, flags, RS_OP_DATA);
2660 if (!ds_can_send(rs)) {
2661 ret = ds_get_comp(rs, rs_nonblocking(rs, flags), ds_can_send);
2666 msg = rs->smsg_free;
2667 rs->smsg_free = msg->next;
2670 memcpy((void *) msg, &rs->conn_dest->qp->hdr, rs->conn_dest->qp->hdr.length);
2671 memcpy((void *) msg + rs->conn_dest->qp->hdr.length, buf, len);
2672 sge.addr = (uintptr_t) msg;
2673 sge.length = rs->conn_dest->qp->hdr.length + len;
2674 sge.lkey = rs->conn_dest->qp->smr->lkey;
2675 offset = (uint8_t *) msg - rs->sbuf;
2677 ret = ds_post_send(rs, &sge, offset);
2678 return ret ? ret : len;
2682 * We overlap sending the data, by posting a small work request immediately,
2683 * then increasing the size of the send on each iteration.
2685 ssize_t rsend(int socket, const void *buf, size_t len, int flags)
2690 uint32_t xfer_size, olen = RS_OLAP_START_SIZE;
2693 rs = idm_at(&idm, socket);
2694 if (rs->type == SOCK_DGRAM) {
2695 fastlock_acquire(&rs->slock);
2696 ret = dsend(rs, buf, len, flags);
2697 fastlock_release(&rs->slock);
2701 if (rs->state & rs_opening) {
2702 ret = rs_do_connect(rs);
2704 if (errno == EINPROGRESS)
2710 fastlock_acquire(&rs->slock);
2711 if (rs->iomap_pending) {
2712 ret = rs_send_iomaps(rs, flags);
2716 for (; left; left -= xfer_size, buf += xfer_size) {
2717 if (!rs_can_send(rs)) {
2718 ret = rs_get_comp(rs, rs_nonblocking(rs, flags),
2722 if (!(rs->state & rs_writable)) {
2723 ret = ERR(ECONNRESET);
2730 if (olen < RS_MAX_TRANSFER)
2736 if (xfer_size > rs->sbuf_bytes_avail)
2737 xfer_size = rs->sbuf_bytes_avail;
2738 if (xfer_size > rs->target_sgl[rs->target_sge].length)
2739 xfer_size = rs->target_sgl[rs->target_sge].length;
2741 if (xfer_size <= rs->sq_inline) {
2742 sge.addr = (uintptr_t) buf;
2743 sge.length = xfer_size;
2745 ret = rs_write_data(rs, &sge, 1, xfer_size, IBV_SEND_INLINE);
2746 } else if (xfer_size <= rs_sbuf_left(rs)) {
2747 memcpy((void *) (uintptr_t) rs->ssgl[0].addr, buf, xfer_size);
2748 rs->ssgl[0].length = xfer_size;
2749 ret = rs_write_data(rs, rs->ssgl, 1, xfer_size, 0);
2750 if (xfer_size < rs_sbuf_left(rs))
2751 rs->ssgl[0].addr += xfer_size;
2753 rs->ssgl[0].addr = (uintptr_t) rs->sbuf;
2755 rs->ssgl[0].length = rs_sbuf_left(rs);
2756 memcpy((void *) (uintptr_t) rs->ssgl[0].addr, buf,
2757 rs->ssgl[0].length);
2758 rs->ssgl[1].length = xfer_size - rs->ssgl[0].length;
2759 memcpy(rs->sbuf, buf + rs->ssgl[0].length, rs->ssgl[1].length);
2760 ret = rs_write_data(rs, rs->ssgl, 2, xfer_size, 0);
2761 rs->ssgl[0].addr = (uintptr_t) rs->sbuf + rs->ssgl[1].length;
2767 fastlock_release(&rs->slock);
2769 return (ret && left == len) ? ret : len - left;
2772 ssize_t rsendto(int socket, const void *buf, size_t len, int flags,
2773 const struct sockaddr *dest_addr, socklen_t addrlen)
2778 rs = idm_at(&idm, socket);
2779 if (rs->type == SOCK_STREAM) {
2780 if (dest_addr || addrlen)
2781 return ERR(EISCONN);
2783 return rsend(socket, buf, len, flags);
2786 if (rs->state == rs_init) {
2787 ret = ds_init_ep(rs);
2792 fastlock_acquire(&rs->slock);
2793 if (!rs->conn_dest || ds_compare_addr(dest_addr, &rs->conn_dest->addr)) {
2794 ret = ds_get_dest(rs, dest_addr, addrlen, &rs->conn_dest);
2799 ret = dsend(rs, buf, len, flags);
2801 fastlock_release(&rs->slock);
2805 static void rs_copy_iov(void *dst, const struct iovec **iov, size_t *offset, size_t len)
2810 size = (*iov)->iov_len - *offset;
2812 memcpy (dst, (*iov)->iov_base + *offset, len);
2817 memcpy(dst, (*iov)->iov_base + *offset, size);
2825 static ssize_t rsendv(int socket, const struct iovec *iov, int iovcnt, int flags)
2828 const struct iovec *cur_iov;
2829 size_t left, len, offset = 0;
2830 uint32_t xfer_size, olen = RS_OLAP_START_SIZE;
2833 rs = idm_at(&idm, socket);
2834 if (rs->state & rs_opening) {
2835 ret = rs_do_connect(rs);
2837 if (errno == EINPROGRESS)
2844 len = iov[0].iov_len;
2845 for (i = 1; i < iovcnt; i++)
2846 len += iov[i].iov_len;
2849 fastlock_acquire(&rs->slock);
2850 if (rs->iomap_pending) {
2851 ret = rs_send_iomaps(rs, flags);
2855 for (; left; left -= xfer_size) {
2856 if (!rs_can_send(rs)) {
2857 ret = rs_get_comp(rs, rs_nonblocking(rs, flags),
2861 if (!(rs->state & rs_writable)) {
2862 ret = ERR(ECONNRESET);
2869 if (olen < RS_MAX_TRANSFER)
2875 if (xfer_size > rs->sbuf_bytes_avail)
2876 xfer_size = rs->sbuf_bytes_avail;
2877 if (xfer_size > rs->target_sgl[rs->target_sge].length)
2878 xfer_size = rs->target_sgl[rs->target_sge].length;
2880 if (xfer_size <= rs_sbuf_left(rs)) {
2881 rs_copy_iov((void *) (uintptr_t) rs->ssgl[0].addr,
2882 &cur_iov, &offset, xfer_size);
2883 rs->ssgl[0].length = xfer_size;
2884 ret = rs_write_data(rs, rs->ssgl, 1, xfer_size,
2885 xfer_size <= rs->sq_inline ? IBV_SEND_INLINE : 0);
2886 if (xfer_size < rs_sbuf_left(rs))
2887 rs->ssgl[0].addr += xfer_size;
2889 rs->ssgl[0].addr = (uintptr_t) rs->sbuf;
2891 rs->ssgl[0].length = rs_sbuf_left(rs);
2892 rs_copy_iov((void *) (uintptr_t) rs->ssgl[0].addr, &cur_iov,
2893 &offset, rs->ssgl[0].length);
2894 rs->ssgl[1].length = xfer_size - rs->ssgl[0].length;
2895 rs_copy_iov(rs->sbuf, &cur_iov, &offset, rs->ssgl[1].length);
2896 ret = rs_write_data(rs, rs->ssgl, 2, xfer_size,
2897 xfer_size <= rs->sq_inline ? IBV_SEND_INLINE : 0);
2898 rs->ssgl[0].addr = (uintptr_t) rs->sbuf + rs->ssgl[1].length;
2904 fastlock_release(&rs->slock);
2906 return (ret && left == len) ? ret : len - left;
2909 ssize_t rsendmsg(int socket, const struct msghdr *msg, int flags)
2911 if (msg->msg_control && msg->msg_controllen)
2912 return ERR(ENOTSUP);
2914 return rsendv(socket, msg->msg_iov, (int) msg->msg_iovlen, flags);
2917 ssize_t rwrite(int socket, const void *buf, size_t count)
2919 return rsend(socket, buf, count, 0);
2922 ssize_t rwritev(int socket, const struct iovec *iov, int iovcnt)
2924 return rsendv(socket, iov, iovcnt, 0);
2927 static struct pollfd *rs_fds_alloc(nfds_t nfds)
2929 static __thread struct pollfd *rfds;
2930 static __thread nfds_t rnfds;
2936 rfds = malloc(sizeof(*rfds) * nfds);
2937 rnfds = rfds ? nfds : 0;
2943 static int rs_poll_rs(struct rsocket *rs, int events,
2944 int nonblock, int (*test)(struct rsocket *rs))
2951 if ((rs->type == SOCK_STREAM) && ((rs->state & rs_connected) ||
2952 (rs->state == rs_disconnected) || (rs->state & rs_error))) {
2953 rs_process_cq(rs, nonblock, test);
2956 if ((events & POLLIN) && rs_conn_have_rdata(rs))
2958 if ((events & POLLOUT) && rs_can_send(rs))
2960 if (!(rs->state & rs_connected)) {
2961 if (rs->state == rs_disconnected)
2968 } else if (rs->type == SOCK_DGRAM) {
2969 ds_process_cqs(rs, nonblock, test);
2972 if ((events & POLLIN) && rs_have_rdata(rs))
2974 if ((events & POLLOUT) && ds_can_send(rs))
2980 if (rs->state == rs_listening) {
2981 fds.fd = rs->cm_id->channel->fd;
2982 fds.events = events;
2988 if (rs->state & rs_opening) {
2989 ret = rs_do_connect(rs);
2990 if (ret && (errno == EINPROGRESS)) {
2997 if (rs->state == rs_connect_error) {
2999 if (events & POLLOUT)
3001 if (events & POLLIN)
3010 static int rs_poll_check(struct pollfd *fds, nfds_t nfds)
3015 for (i = 0; i < nfds; i++) {
3016 rs = idm_lookup(&idm, fds[i].fd);
3018 fds[i].revents = rs_poll_rs(rs, fds[i].events, 1, rs_poll_all);
3020 poll(&fds[i], 1, 0);
3028 static int rs_poll_arm(struct pollfd *rfds, struct pollfd *fds, nfds_t nfds)
3033 for (i = 0; i < nfds; i++) {
3034 rs = idm_lookup(&idm, fds[i].fd);
3036 fds[i].revents = rs_poll_rs(rs, fds[i].events, 0, rs_is_cq_armed);
3040 if (rs->type == SOCK_STREAM) {
3041 if (rs->state >= rs_connected)
3042 rfds[i].fd = rs->cm_id->recv_cq_channel->fd;
3044 rfds[i].fd = rs->cm_id->channel->fd;
3046 rfds[i].fd = rs->epfd;
3048 rfds[i].events = POLLIN;
3050 rfds[i].fd = fds[i].fd;
3051 rfds[i].events = fds[i].events;
3053 rfds[i].revents = 0;
3058 static int rs_poll_events(struct pollfd *rfds, struct pollfd *fds, nfds_t nfds)
3063 for (i = 0; i < nfds; i++) {
3064 if (!rfds[i].revents)
3067 rs = idm_lookup(&idm, fds[i].fd);
3069 fastlock_acquire(&rs->cq_wait_lock);
3070 if (rs->type == SOCK_STREAM)
3071 rs_get_cq_event(rs);
3073 ds_get_cq_event(rs);
3074 fastlock_release(&rs->cq_wait_lock);
3075 fds[i].revents = rs_poll_rs(rs, fds[i].events, 1, rs_poll_all);
3077 fds[i].revents = rfds[i].revents;
3086 * We need to poll *all* fd's that the user specifies at least once.
3087 * Note that we may receive events on an rsocket that may not be reported
3088 * to the user (e.g. connection events or credit updates). Process those
3089 * events, then return to polling until we find ones of interest.
3091 int rpoll(struct pollfd *fds, nfds_t nfds, int timeout)
3093 struct timeval s, e;
3094 struct pollfd *rfds;
3095 uint32_t poll_time = 0;
3099 ret = rs_poll_check(fds, nfds);
3100 if (ret || !timeout)
3104 gettimeofday(&s, NULL);
3106 gettimeofday(&e, NULL);
3107 poll_time = (e.tv_sec - s.tv_sec) * 1000000 +
3108 (e.tv_usec - s.tv_usec) + 1;
3109 } while (poll_time <= polling_time);
3111 rfds = rs_fds_alloc(nfds);
3116 ret = rs_poll_arm(rfds, fds, nfds);
3120 ret = poll(rfds, nfds, timeout);
3124 ret = rs_poll_events(rfds, fds, nfds);
3130 static struct pollfd *
3131 rs_select_to_poll(int *nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds)
3136 fds = calloc(*nfds, sizeof(*fds));
3140 for (fd = 0; fd < *nfds; fd++) {
3141 if (readfds && FD_ISSET(fd, readfds)) {
3143 fds[i].events = POLLIN;
3146 if (writefds && FD_ISSET(fd, writefds)) {
3148 fds[i].events |= POLLOUT;
3151 if (exceptfds && FD_ISSET(fd, exceptfds))
3163 rs_poll_to_select(int nfds, struct pollfd *fds, fd_set *readfds,
3164 fd_set *writefds, fd_set *exceptfds)
3168 for (i = 0; i < nfds; i++) {
3169 if (readfds && (fds[i].revents & (POLLIN | POLLHUP))) {
3170 FD_SET(fds[i].fd, readfds);
3174 if (writefds && (fds[i].revents & POLLOUT)) {
3175 FD_SET(fds[i].fd, writefds);
3179 if (exceptfds && (fds[i].revents & ~(POLLIN | POLLOUT))) {
3180 FD_SET(fds[i].fd, exceptfds);
3187 static int rs_convert_timeout(struct timeval *timeout)
3189 return !timeout ? -1 :
3190 timeout->tv_sec * 1000 + timeout->tv_usec / 1000;
3193 int rselect(int nfds, fd_set *readfds, fd_set *writefds,
3194 fd_set *exceptfds, struct timeval *timeout)
3199 fds = rs_select_to_poll(&nfds, readfds, writefds, exceptfds);
3203 ret = rpoll(fds, nfds, rs_convert_timeout(timeout));
3213 ret = rs_poll_to_select(nfds, fds, readfds, writefds, exceptfds);
3220 * For graceful disconnect, notify the remote side that we're
3221 * disconnecting and wait until all outstanding sends complete, provided
3222 * that the remote side has not sent a disconnect message.
3224 int rshutdown(int socket, int how)
3229 rs = idm_lookup(&idm, socket);
3232 if (rs->opts & RS_OPT_SVC_ACTIVE)
3233 rs_notify_svc(&tcp_svc, rs, RS_SVC_REM_KEEPALIVE);
3235 if (rs->fd_flags & O_NONBLOCK)
3236 rs_set_nonblocking(rs, 0);
3238 if (rs->state & rs_connected) {
3239 if (how == SHUT_RDWR) {
3240 ctrl = RS_CTRL_DISCONNECT;
3241 rs->state &= ~(rs_readable | rs_writable);
3242 } else if (how == SHUT_WR) {
3243 rs->state &= ~rs_writable;
3244 ctrl = (rs->state & rs_readable) ?
3245 RS_CTRL_SHUTDOWN : RS_CTRL_DISCONNECT;
3247 rs->state &= ~rs_readable;
3248 if (rs->state & rs_writable)
3250 ctrl = RS_CTRL_DISCONNECT;
3252 if (!rs_ctrl_avail(rs)) {
3253 ret = rs_process_cq(rs, 0, rs_conn_can_send_ctrl);
3258 if ((rs->state & rs_connected) && rs_ctrl_avail(rs)) {
3260 ret = rs_post_msg(rs, rs_msg_set(RS_OP_CTRL, ctrl));
3264 if (rs->state & rs_connected)
3265 rs_process_cq(rs, 0, rs_conn_all_sends_done);
3268 if ((rs->fd_flags & O_NONBLOCK) && (rs->state & rs_connected))
3269 rs_set_nonblocking(rs, rs->fd_flags);
3271 if (rs->state & rs_disconnected) {
3272 /* Generate event by flushing receives to unblock rpoll */
3273 ibv_req_notify_cq(rs->cm_id->recv_cq, 0);
3274 ucma_shutdown(rs->cm_id);
3280 static void ds_shutdown(struct rsocket *rs)
3282 if (rs->opts & RS_OPT_SVC_ACTIVE)
3283 rs_notify_svc(&udp_svc, rs, RS_SVC_REM_DGRAM);
3285 if (rs->fd_flags & O_NONBLOCK)
3286 rs_set_nonblocking(rs, 0);
3288 rs->state &= ~(rs_readable | rs_writable);
3289 ds_process_cqs(rs, 0, ds_all_sends_done);
3291 if (rs->fd_flags & O_NONBLOCK)
3292 rs_set_nonblocking(rs, rs->fd_flags);
3295 int rclose(int socket)
3299 rs = idm_lookup(&idm, socket);
3302 if (rs->type == SOCK_STREAM) {
3303 if (rs->state & rs_connected)
3304 rshutdown(socket, SHUT_RDWR);
3305 else if (rs->opts & RS_OPT_SVC_ACTIVE)
3306 rs_notify_svc(&tcp_svc, rs, RS_SVC_REM_KEEPALIVE);
3315 static void rs_copy_addr(struct sockaddr *dst, struct sockaddr *src, socklen_t *len)
3319 if (src->sa_family == AF_INET) {
3320 size = min_t(socklen_t, *len, sizeof(struct sockaddr_in));
3321 *len = sizeof(struct sockaddr_in);
3323 size = min_t(socklen_t, *len, sizeof(struct sockaddr_in6));
3324 *len = sizeof(struct sockaddr_in6);
3326 memcpy(dst, src, size);
3329 int rgetpeername(int socket, struct sockaddr *addr, socklen_t *addrlen)
3333 rs = idm_lookup(&idm, socket);
3336 if (rs->type == SOCK_STREAM) {
3337 rs_copy_addr(addr, rdma_get_peer_addr(rs->cm_id), addrlen);
3340 return getpeername(rs->udp_sock, addr, addrlen);
3344 int rgetsockname(int socket, struct sockaddr *addr, socklen_t *addrlen)
3348 rs = idm_lookup(&idm, socket);
3351 if (rs->type == SOCK_STREAM) {
3352 rs_copy_addr(addr, rdma_get_local_addr(rs->cm_id), addrlen);
3355 return getsockname(rs->udp_sock, addr, addrlen);
3359 static int rs_set_keepalive(struct rsocket *rs, int on)
3364 if ((on && (rs->opts & RS_OPT_SVC_ACTIVE)) ||
3365 (!on && !(rs->opts & RS_OPT_SVC_ACTIVE)))
3369 if (!rs->keepalive_time) {
3370 if ((f = fopen("/proc/sys/net/ipv4/tcp_keepalive_time", "r"))) {
3371 if (fscanf(f, "%u", &rs->keepalive_time) != 1)
3372 rs->keepalive_time = 7200;
3375 rs->keepalive_time = 7200;
3378 ret = rs_notify_svc(&tcp_svc, rs, RS_SVC_ADD_KEEPALIVE);
3380 ret = rs_notify_svc(&tcp_svc, rs, RS_SVC_REM_KEEPALIVE);
3386 int rsetsockopt(int socket, int level, int optname,
3387 const void *optval, socklen_t optlen)
3390 int ret, opt_on = 0;
3391 uint64_t *opts = NULL;
3394 rs = idm_lookup(&idm, socket);
3397 if (rs->type == SOCK_DGRAM && level != SOL_RDMA) {
3398 ret = setsockopt(rs->udp_sock, level, optname, optval, optlen);
3405 opts = &rs->so_opts;
3408 if (rs->type == SOCK_STREAM) {
3409 ret = rdma_set_option(rs->cm_id, RDMA_OPTION_ID,
3410 RDMA_OPTION_ID_REUSEADDR,
3411 (void *) optval, optlen);
3412 if (ret && ((errno == ENOSYS) || ((rs->state != rs_init) &&
3413 rs->cm_id->context &&
3414 (rs->cm_id->verbs->device->transport_type == IBV_TRANSPORT_IB))))
3417 opt_on = *(int *) optval;
3420 if ((rs->type == SOCK_STREAM && !rs->rbuf) ||
3421 (rs->type == SOCK_DGRAM && !rs->qp_list))
3422 rs->rbuf_size = (*(uint32_t *) optval) << 1;
3427 rs->sbuf_size = (*(uint32_t *) optval) << 1;
3428 if (rs->sbuf_size < RS_SNDLOWAT)
3429 rs->sbuf_size = RS_SNDLOWAT << 1;
3433 /* Invert value so default so_opt = 0 is on */
3434 opt_on = !((struct linger *) optval)->l_onoff;
3438 ret = rs_set_keepalive(rs, *(int *) optval);
3439 opt_on = rs->opts & RS_OPT_SVC_ACTIVE;
3442 opt_on = *(int *) optval;
3450 opts = &rs->tcp_opts;
3454 ret = 0; /* N/A - we're using a reliable connection */
3457 if (*(int *) optval <= 0) {
3461 rs->keepalive_time = *(int *) optval;
3462 ret = (rs->opts & RS_OPT_SVC_ACTIVE) ?
3463 rs_notify_svc(&tcp_svc, rs, RS_SVC_MOD_KEEPALIVE) : 0;
3466 opt_on = *(int *) optval;
3477 opts = &rs->ipv6_opts;
3480 if (rs->type == SOCK_STREAM) {
3481 ret = rdma_set_option(rs->cm_id, RDMA_OPTION_ID,
3482 RDMA_OPTION_ID_AFONLY,
3483 (void *) optval, optlen);
3485 opt_on = *(int *) optval;
3492 if (rs->state >= rs_opening) {
3499 rs->sq_size = min_t(uint32_t, (*(uint32_t *)optval),
3504 rs->rq_size = min_t(uint32_t, (*(uint32_t *)optval),
3509 rs->sq_inline = min_t(uint32_t, *(uint32_t *)optval,
3513 case RDMA_IOMAPSIZE:
3514 rs->target_iomap_size = (uint16_t) rs_scale_to_value(
3515 (uint8_t) rs_value_to_scale(*(int *) optval, 8), 8);
3519 if ((rs->optval = malloc(optlen))) {
3520 memcpy(rs->optval, optval, optlen);
3521 rs->optlen = optlen;
3537 *opts |= (1 << optname);
3539 *opts &= ~(1 << optname);
3545 static void rs_convert_sa_path(struct ibv_sa_path_rec *sa_path,
3546 struct ibv_path_data *path_data)
3550 memset(path_data, 0, sizeof(*path_data));
3551 path_data->path.dgid = sa_path->dgid;
3552 path_data->path.sgid = sa_path->sgid;
3553 path_data->path.dlid = sa_path->dlid;
3554 path_data->path.slid = sa_path->slid;
3555 fl_hop = be32toh(sa_path->flow_label) << 8;
3556 path_data->path.flowlabel_hoplimit = htobe32(fl_hop | sa_path->hop_limit);
3557 path_data->path.tclass = sa_path->traffic_class;
3558 path_data->path.reversible_numpath = sa_path->reversible << 7 | 1;
3559 path_data->path.pkey = sa_path->pkey;
3560 path_data->path.qosclass_sl = htobe16(sa_path->sl);
3561 path_data->path.mtu = sa_path->mtu | 2 << 6; /* exactly */
3562 path_data->path.rate = sa_path->rate | 2 << 6;
3563 path_data->path.packetlifetime = sa_path->packet_life_time | 2 << 6;
3564 path_data->flags= sa_path->preference;
3567 int rgetsockopt(int socket, int level, int optname,
3568 void *optval, socklen_t *optlen)
3572 struct ibv_sa_path_rec *path_rec;
3573 struct ibv_path_data path_data;
3578 rs = idm_lookup(&idm, socket);
3587 *((int *) optval) = !!(rs->so_opts & (1 << optname));
3588 *optlen = sizeof(int);
3591 *((int *) optval) = rs->rbuf_size;
3592 *optlen = sizeof(int);
3595 *((int *) optval) = rs->sbuf_size;
3596 *optlen = sizeof(int);
3599 /* Value is inverted so default so_opt = 0 is on */
3600 ((struct linger *) optval)->l_onoff =
3601 !(rs->so_opts & (1 << optname));
3602 ((struct linger *) optval)->l_linger = 0;
3603 *optlen = sizeof(struct linger);
3606 *((int *) optval) = rs->err;
3607 *optlen = sizeof(int);
3619 *((int *) optval) = 1; /* N/A */
3622 *((int *) optval) = (int) rs->keepalive_time;
3623 *optlen = sizeof(int);
3626 *((int *) optval) = !!(rs->tcp_opts & (1 << optname));
3627 *optlen = sizeof(int);
3630 *((int *) optval) = (rs->cm_id && rs->cm_id->route.num_paths) ?
3631 1 << (7 + rs->cm_id->route.path_rec->mtu) :
3633 *optlen = sizeof(int);
3643 *((int *) optval) = !!(rs->ipv6_opts & (1 << optname));
3644 *optlen = sizeof(int);
3654 *((int *) optval) = rs->sq_size;
3655 *optlen = sizeof(int);
3658 *((int *) optval) = rs->rq_size;
3659 *optlen = sizeof(int);
3662 *((int *) optval) = rs->sq_inline;
3663 *optlen = sizeof(int);
3665 case RDMA_IOMAPSIZE:
3666 *((int *) optval) = rs->target_iomap_size;
3667 *optlen = sizeof(int);
3671 if (*optlen < rs->optlen) {
3674 memcpy(rs->optval, optval, rs->optlen);
3675 *optlen = rs->optlen;
3678 if (*optlen < sizeof(path_data)) {
3683 path_rec = rs->cm_id->route.path_rec;
3685 while (len + sizeof(path_data) <= *optlen &&
3686 num_paths < rs->cm_id->route.num_paths) {
3687 rs_convert_sa_path(path_rec, &path_data);
3688 memcpy(opt, &path_data, sizeof(path_data));
3689 len += sizeof(path_data);
3690 opt += sizeof(path_data);
3709 return rdma_seterrno(ret);
3712 int rfcntl(int socket, int cmd, ... /* arg */ )
3719 rs = idm_lookup(&idm, socket);
3722 va_start(args, cmd);
3728 param = va_arg(args, int);
3729 if ((rs->fd_flags & O_NONBLOCK) != (param & O_NONBLOCK))
3730 ret = rs_set_nonblocking(rs, param & O_NONBLOCK);
3733 rs->fd_flags = param;
3743 static struct rs_iomap_mr *rs_get_iomap_mr(struct rsocket *rs)
3747 if (!rs->remote_iomappings) {
3748 rs->remote_iomappings = calloc(rs->remote_iomap.length,
3749 sizeof(*rs->remote_iomappings));
3750 if (!rs->remote_iomappings)
3753 for (i = 0; i < rs->remote_iomap.length; i++)
3754 rs->remote_iomappings[i].index = i;
3757 for (i = 0; i < rs->remote_iomap.length; i++) {
3758 if (!rs->remote_iomappings[i].mr)
3759 return &rs->remote_iomappings[i];
3765 * If an offset is given, we map to it. If offset is -1, then we map the
3766 * offset to the address of buf. We do not check for conflicts, which must
3767 * be fixed at some point.
3769 off_t riomap(int socket, void *buf, size_t len, int prot, int flags, off_t offset)
3772 struct rs_iomap_mr *iomr;
3773 int access = IBV_ACCESS_LOCAL_WRITE;
3775 rs = idm_at(&idm, socket);
3776 if (!rs->cm_id->pd || (prot & ~(PROT_WRITE | PROT_NONE)))
3779 fastlock_acquire(&rs->map_lock);
3780 if (prot & PROT_WRITE) {
3781 iomr = rs_get_iomap_mr(rs);
3782 access |= IBV_ACCESS_REMOTE_WRITE;
3784 iomr = calloc(1, sizeof(*iomr));
3788 offset = ERR(ENOMEM);
3792 iomr->mr = ibv_reg_mr(rs->cm_id->pd, buf, len, access);
3794 if (iomr->index < 0)
3801 offset = (uintptr_t) buf;
3802 iomr->offset = offset;
3803 atomic_store(&iomr->refcnt, 1);
3805 if (iomr->index >= 0) {
3806 dlist_insert_tail(&iomr->entry, &rs->iomap_queue);
3807 rs->iomap_pending = 1;
3809 dlist_insert_tail(&iomr->entry, &rs->iomap_list);
3812 fastlock_release(&rs->map_lock);
3816 int riounmap(int socket, void *buf, size_t len)
3819 struct rs_iomap_mr *iomr;
3823 rs = idm_at(&idm, socket);
3824 fastlock_acquire(&rs->map_lock);
3826 for (entry = rs->iomap_list.next; entry != &rs->iomap_list;
3827 entry = entry->next) {
3828 iomr = container_of(entry, struct rs_iomap_mr, entry);
3829 if (iomr->mr->addr == buf && iomr->mr->length == len) {
3830 rs_release_iomap_mr(iomr);
3835 for (entry = rs->iomap_queue.next; entry != &rs->iomap_queue;
3836 entry = entry->next) {
3837 iomr = container_of(entry, struct rs_iomap_mr, entry);
3838 if (iomr->mr->addr == buf && iomr->mr->length == len) {
3839 rs_release_iomap_mr(iomr);
3845 fastlock_release(&rs->map_lock);
3849 static struct rs_iomap *rs_find_iomap(struct rsocket *rs, off_t offset)
3853 for (i = 0; i < rs->target_iomap_size; i++) {
3854 if (offset >= rs->target_iomap[i].offset &&
3855 offset < rs->target_iomap[i].offset + rs->target_iomap[i].sge.length)
3856 return &rs->target_iomap[i];
3861 size_t riowrite(int socket, const void *buf, size_t count, off_t offset, int flags)
3864 struct rs_iomap *iom = NULL;
3866 size_t left = count;
3867 uint32_t xfer_size, olen = RS_OLAP_START_SIZE;
3870 rs = idm_at(&idm, socket);
3871 fastlock_acquire(&rs->slock);
3872 if (rs->iomap_pending) {
3873 ret = rs_send_iomaps(rs, flags);
3877 for (; left; left -= xfer_size, buf += xfer_size, offset += xfer_size) {
3878 if (!iom || offset > iom->offset + iom->sge.length) {
3879 iom = rs_find_iomap(rs, offset);
3884 if (!rs_can_send(rs)) {
3885 ret = rs_get_comp(rs, rs_nonblocking(rs, flags),
3889 if (!(rs->state & rs_writable)) {
3890 ret = ERR(ECONNRESET);
3897 if (olen < RS_MAX_TRANSFER)
3903 if (xfer_size > rs->sbuf_bytes_avail)
3904 xfer_size = rs->sbuf_bytes_avail;
3905 if (xfer_size > iom->offset + iom->sge.length - offset)
3906 xfer_size = iom->offset + iom->sge.length - offset;
3908 if (xfer_size <= rs->sq_inline) {
3909 sge.addr = (uintptr_t) buf;
3910 sge.length = xfer_size;
3912 ret = rs_write_direct(rs, iom, offset, &sge, 1,
3913 xfer_size, IBV_SEND_INLINE);
3914 } else if (xfer_size <= rs_sbuf_left(rs)) {
3915 memcpy((void *) (uintptr_t) rs->ssgl[0].addr, buf, xfer_size);
3916 rs->ssgl[0].length = xfer_size;
3917 ret = rs_write_direct(rs, iom, offset, rs->ssgl, 1, xfer_size, 0);
3918 if (xfer_size < rs_sbuf_left(rs))
3919 rs->ssgl[0].addr += xfer_size;
3921 rs->ssgl[0].addr = (uintptr_t) rs->sbuf;
3923 rs->ssgl[0].length = rs_sbuf_left(rs);
3924 memcpy((void *) (uintptr_t) rs->ssgl[0].addr, buf,
3925 rs->ssgl[0].length);
3926 rs->ssgl[1].length = xfer_size - rs->ssgl[0].length;
3927 memcpy(rs->sbuf, buf + rs->ssgl[0].length, rs->ssgl[1].length);
3928 ret = rs_write_direct(rs, iom, offset, rs->ssgl, 2, xfer_size, 0);
3929 rs->ssgl[0].addr = (uintptr_t) rs->sbuf + rs->ssgl[1].length;
3935 fastlock_release(&rs->slock);
3937 return (ret && left == count) ? ret : count - left;
3940 /****************************************************************************
3941 * Service Processing Threads
3942 ****************************************************************************/
3944 static int rs_svc_grow_sets(struct rs_svc *svc, int grow_size)
3946 struct rsocket **rss;
3947 void *set, *contexts;
3949 set = calloc(svc->size + grow_size, sizeof(*rss) + svc->context_size);
3953 svc->size += grow_size;
3955 contexts = set + sizeof(*rss) * svc->size;
3957 memcpy(rss, svc->rss, sizeof(*rss) * (svc->cnt + 1));
3958 memcpy(contexts, svc->contexts, svc->context_size * (svc->cnt + 1));
3963 svc->contexts = contexts;
3968 * Index 0 is reserved for the service's communication socket.
3970 static int rs_svc_add_rs(struct rs_svc *svc, struct rsocket *rs)
3974 if (svc->cnt >= svc->size - 1) {
3975 ret = rs_svc_grow_sets(svc, 4);
3980 svc->rss[++svc->cnt] = rs;
3984 static int rs_svc_index(struct rs_svc *svc, struct rsocket *rs)
3988 for (i = 1; i <= svc->cnt; i++) {
3989 if (svc->rss[i] == rs)
3995 static int rs_svc_rm_rs(struct rs_svc *svc, struct rsocket *rs)
3999 if ((i = rs_svc_index(svc, rs)) >= 0) {
4000 svc->rss[i] = svc->rss[svc->cnt];
4001 memcpy(svc->contexts + i * svc->context_size,
4002 svc->contexts + svc->cnt * svc->context_size,
4010 static void udp_svc_process_sock(struct rs_svc *svc)
4012 struct rs_svc_msg msg;
4014 read_all(svc->sock[1], &msg, sizeof msg);
4016 case RS_SVC_ADD_DGRAM:
4017 msg.status = rs_svc_add_rs(svc, msg.rs);
4019 msg.rs->opts |= RS_OPT_SVC_ACTIVE;
4020 udp_svc_fds = svc->contexts;
4021 udp_svc_fds[svc->cnt].fd = msg.rs->udp_sock;
4022 udp_svc_fds[svc->cnt].events = POLLIN;
4023 udp_svc_fds[svc->cnt].revents = 0;
4026 case RS_SVC_REM_DGRAM:
4027 msg.status = rs_svc_rm_rs(svc, msg.rs);
4029 msg.rs->opts &= ~RS_OPT_SVC_ACTIVE;
4038 write_all(svc->sock[1], &msg, sizeof msg);
4041 static uint8_t udp_svc_sgid_index(struct ds_dest *dest, union ibv_gid *sgid)
4046 for (i = 0; i < 16; i++) {
4047 ibv_query_gid(dest->qp->cm_id->verbs, dest->qp->cm_id->port_num,
4049 if (!memcmp(sgid, &gid, sizeof gid))
4055 static uint8_t udp_svc_path_bits(struct ds_dest *dest)
4057 struct ibv_port_attr attr;
4059 if (!ibv_query_port(dest->qp->cm_id->verbs, dest->qp->cm_id->port_num, &attr))
4060 return (uint8_t) ((1 << attr.lmc) - 1);
4064 static void udp_svc_create_ah(struct rsocket *rs, struct ds_dest *dest, uint32_t qpn)
4066 union socket_addr saddr;
4067 struct rdma_cm_id *id;
4068 struct ibv_ah_attr attr;
4072 fastlock_acquire(&rs->slock);
4073 ibv_destroy_ah(dest->ah);
4075 fastlock_release(&rs->slock);
4078 ret = rdma_create_id(NULL, &id, NULL, dest->qp->cm_id->ps);
4082 memcpy(&saddr, rdma_get_local_addr(dest->qp->cm_id),
4083 ucma_addrlen(rdma_get_local_addr(dest->qp->cm_id)));
4084 if (saddr.sa.sa_family == AF_INET)
4085 saddr.sin.sin_port = 0;
4087 saddr.sin6.sin6_port = 0;
4088 ret = rdma_resolve_addr(id, &saddr.sa, &dest->addr.sa, 2000);
4092 ret = rdma_resolve_route(id, 2000);
4096 memset(&attr, 0, sizeof attr);
4097 if (id->route.path_rec->hop_limit > 1) {
4099 attr.grh.dgid = id->route.path_rec->dgid;
4100 attr.grh.flow_label = be32toh(id->route.path_rec->flow_label);
4101 attr.grh.sgid_index = udp_svc_sgid_index(dest, &id->route.path_rec->sgid);
4102 attr.grh.hop_limit = id->route.path_rec->hop_limit;
4103 attr.grh.traffic_class = id->route.path_rec->traffic_class;
4105 attr.dlid = be16toh(id->route.path_rec->dlid);
4106 attr.sl = id->route.path_rec->sl;
4107 attr.src_path_bits = be16toh(id->route.path_rec->slid) & udp_svc_path_bits(dest);
4108 attr.static_rate = id->route.path_rec->rate;
4109 attr.port_num = id->port_num;
4111 fastlock_acquire(&rs->slock);
4113 dest->ah = ibv_create_ah(dest->qp->cm_id->pd, &attr);
4114 fastlock_release(&rs->slock);
4116 rdma_destroy_id(id);
4119 static int udp_svc_valid_udp_hdr(struct ds_udp_header *udp_hdr,
4120 union socket_addr *addr)
4122 return (udp_hdr->tag == htobe32(DS_UDP_TAG)) &&
4123 ((udp_hdr->version == 4 && addr->sa.sa_family == AF_INET &&
4124 udp_hdr->length == DS_UDP_IPV4_HDR_LEN) ||
4125 (udp_hdr->version == 6 && addr->sa.sa_family == AF_INET6 &&
4126 udp_hdr->length == DS_UDP_IPV6_HDR_LEN));
4129 static void udp_svc_forward(struct rsocket *rs, void *buf, size_t len,
4130 union socket_addr *src)
4132 struct ds_header hdr;
4133 struct ds_smsg *msg;
4137 if (!ds_can_send(rs)) {
4138 if (ds_get_comp(rs, 0, ds_can_send))
4142 msg = rs->smsg_free;
4143 rs->smsg_free = msg->next;
4146 ds_format_hdr(&hdr, src);
4147 memcpy((void *) msg, &hdr, hdr.length);
4148 memcpy((void *) msg + hdr.length, buf, len);
4149 sge.addr = (uintptr_t) msg;
4150 sge.length = hdr.length + len;
4151 sge.lkey = rs->conn_dest->qp->smr->lkey;
4152 offset = (uint8_t *) msg - rs->sbuf;
4154 ds_post_send(rs, &sge, offset);
4157 static void udp_svc_process_rs(struct rsocket *rs)
4159 static uint8_t buf[RS_SNDLOWAT];
4160 struct ds_dest *dest, *cur_dest;
4161 struct ds_udp_header *udp_hdr;
4162 union socket_addr addr;
4163 socklen_t addrlen = sizeof addr;
4167 ret = recvfrom(rs->udp_sock, buf, sizeof buf, 0, &addr.sa, &addrlen);
4168 if (ret < DS_UDP_IPV4_HDR_LEN)
4171 udp_hdr = (struct ds_udp_header *) buf;
4172 if (!udp_svc_valid_udp_hdr(udp_hdr, &addr))
4175 len = ret - udp_hdr->length;
4176 qpn = be32toh(udp_hdr->qpn) & 0xFFFFFF;
4178 udp_hdr->tag = (__force __be32)be32toh(udp_hdr->tag);
4179 udp_hdr->qpn = (__force __be32)qpn;
4181 ret = ds_get_dest(rs, &addr.sa, addrlen, &dest);
4185 if (udp_hdr->op == RS_OP_DATA) {
4186 fastlock_acquire(&rs->slock);
4187 cur_dest = rs->conn_dest;
4188 rs->conn_dest = dest;
4189 ds_send_udp(rs, NULL, 0, 0, RS_OP_CTRL);
4190 rs->conn_dest = cur_dest;
4191 fastlock_release(&rs->slock);
4194 if (!dest->ah || (dest->qpn != qpn))
4195 udp_svc_create_ah(rs, dest, qpn);
4197 /* to do: handle when dest local ip address doesn't match udp ip */
4198 if (udp_hdr->op == RS_OP_DATA) {
4199 fastlock_acquire(&rs->slock);
4200 cur_dest = rs->conn_dest;
4201 rs->conn_dest = &dest->qp->dest;
4202 udp_svc_forward(rs, buf + udp_hdr->length, len, &addr);
4203 rs->conn_dest = cur_dest;
4204 fastlock_release(&rs->slock);
4208 static void *udp_svc_run(void *arg)
4210 struct rs_svc *svc = arg;
4211 struct rs_svc_msg msg;
4214 ret = rs_svc_grow_sets(svc, 4);
4217 write_all(svc->sock[1], &msg, sizeof msg);
4218 return (void *) (uintptr_t) ret;
4221 udp_svc_fds = svc->contexts;
4222 udp_svc_fds[0].fd = svc->sock[1];
4223 udp_svc_fds[0].events = POLLIN;
4225 for (i = 0; i <= svc->cnt; i++)
4226 udp_svc_fds[i].revents = 0;
4228 poll(udp_svc_fds, svc->cnt + 1, -1);
4229 if (udp_svc_fds[0].revents)
4230 udp_svc_process_sock(svc);
4232 for (i = 1; i <= svc->cnt; i++) {
4233 if (udp_svc_fds[i].revents)
4234 udp_svc_process_rs(svc->rss[i]);
4236 } while (svc->cnt >= 1);
4241 static uint32_t rs_get_time(void)
4245 memset(&now, 0, sizeof now);
4246 gettimeofday(&now, NULL);
4247 return (uint32_t) now.tv_sec;
4250 static void tcp_svc_process_sock(struct rs_svc *svc)
4252 struct rs_svc_msg msg;
4255 read_all(svc->sock[1], &msg, sizeof msg);
4257 case RS_SVC_ADD_KEEPALIVE:
4258 msg.status = rs_svc_add_rs(svc, msg.rs);
4260 msg.rs->opts |= RS_OPT_SVC_ACTIVE;
4261 tcp_svc_timeouts = svc->contexts;
4262 tcp_svc_timeouts[svc->cnt] = rs_get_time() +
4263 msg.rs->keepalive_time;
4266 case RS_SVC_REM_KEEPALIVE:
4267 msg.status = rs_svc_rm_rs(svc, msg.rs);
4269 msg.rs->opts &= ~RS_OPT_SVC_ACTIVE;
4271 case RS_SVC_MOD_KEEPALIVE:
4272 i = rs_svc_index(svc, msg.rs);
4274 tcp_svc_timeouts[i] = rs_get_time() + msg.rs->keepalive_time;
4286 write_all(svc->sock[1], &msg, sizeof msg);
4290 * Send a 0 byte RDMA write with immediate as keep-alive message.
4291 * This avoids the need for the receive side to do any acknowledgment.
4293 static void tcp_svc_send_keepalive(struct rsocket *rs)
4295 fastlock_acquire(&rs->cq_lock);
4296 if (rs_ctrl_avail(rs) && (rs->state & rs_connected)) {
4298 rs_post_write(rs, NULL, 0, rs_msg_set(RS_OP_CTRL, RS_CTRL_KEEPALIVE),
4299 0, (uintptr_t) NULL, (uintptr_t) NULL);
4301 fastlock_release(&rs->cq_lock);
4304 static void *tcp_svc_run(void *arg)
4306 struct rs_svc *svc = arg;
4307 struct rs_svc_msg msg;
4309 uint32_t now, next_timeout;
4310 int i, ret, timeout;
4312 ret = rs_svc_grow_sets(svc, 16);
4315 write_all(svc->sock[1], &msg, sizeof msg);
4316 return (void *) (uintptr_t) ret;
4319 tcp_svc_timeouts = svc->contexts;
4320 fds.fd = svc->sock[1];
4321 fds.events = POLLIN;
4324 poll(&fds, 1, timeout * 1000);
4326 tcp_svc_process_sock(svc);
4328 now = rs_get_time();
4330 for (i = 1; i <= svc->cnt; i++) {
4331 if (tcp_svc_timeouts[i] <= now) {
4332 tcp_svc_send_keepalive(svc->rss[i]);
4333 tcp_svc_timeouts[i] =
4334 now + svc->rss[i]->keepalive_time;
4336 if (tcp_svc_timeouts[i] < next_timeout)
4337 next_timeout = tcp_svc_timeouts[i];
4339 timeout = (int) (next_timeout - now);
4340 } while (svc->cnt >= 1);