Bottom: 1fa07c62817ac4b6cb8d9c5e327ea2cdc75dbd21
-Top: ac38ccc22b5116e46418ba99f1924f1c35d49ffc
+Top: ae4df7b46b3d1c519383a31cb018c70f3e4e8201
Author: Sean Hefty <sean.hefty@intel.com>
Date: 2012-11-09 10:26:38 -0800
{
errno = err;
diff --git a/src/rsocket.c b/src/rsocket.c
-index a060f66..018af90 100644
+index a060f66..9996d33 100644
--- a/src/rsocket.c
+++ b/src/rsocket.c
-@@ -47,6 +47,7 @@
+@@ -47,6 +47,8 @@
#include <string.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
+#include <sys/epoll.h>
++#include <search.h>
#include <rdma/rdma_cma.h>
#include <rdma/rdma_verbs.h>
-@@ -56,7 +57,7 @@
+@@ -56,7 +58,7 @@
#define RS_OLAP_START_SIZE 2048
#define RS_MAX_TRANSFER 65536
#define RS_QP_MAX_SIZE 0xFFFE
#define RS_QP_CTRL_SIZE 4
#define RS_CONN_RETRIES 6
-@@ -64,6 +65,28 @@
+@@ -64,6 +66,28 @@
static struct index_map idm;
static pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER;
+static struct rsocket **svc_rss;
+static struct pollfd *svc_fds;
+static uint8_t svc_buf[RS_SNDLOWAT];
-+static int rs_svc_run(void *arg);
++static void *rs_svc_run(void *arg);
+
static uint16_t def_iomap_size = 0;
static uint16_t def_inline = 64;
static uint16_t def_sqsize = 384;
-@@ -100,6 +123,14 @@ enum {
+@@ -100,6 +124,14 @@ enum {
#define rs_msg_set(op, data) ((op << 29) | (uint32_t) (data))
#define rs_msg_op(imm_data) (imm_data >> 29)
#define rs_msg_data(imm_data) (imm_data & 0x1FFFFFFF)
enum {
RS_CTRL_DISCONNECT,
-@@ -111,6 +142,18 @@ struct rs_msg {
+@@ -111,6 +143,18 @@ struct rs_msg {
uint32_t data;
};
struct rs_sge {
uint64_t addr;
uint32_t key;
-@@ -145,8 +188,6 @@ struct rs_conn_data {
+@@ -145,8 +189,6 @@ struct rs_conn_data {
struct rs_sge data_buf;
};
/*
* rsocket states are ordered as passive, connecting, connected, disconnected.
*/
-@@ -160,9 +201,9 @@ enum rs_state {
+@@ -160,9 +202,9 @@ enum rs_state {
rs_connecting = rs_opening | 0x0040,
rs_accepting = rs_opening | 0x0080,
rs_connected = 0x0100,
rs_connect_error = 0x0800,
rs_disconnected = 0x1000,
rs_error = 0x2000,
-@@ -170,68 +211,235 @@ enum rs_state {
+@@ -170,68 +212,248 @@ enum rs_state {
#define RS_OPT_SWAP_SGL 1
+};
+
+struct ds_qp {
-+ dlist_t list;
++ dlist_entry list;
+ struct rsocket *rs;
struct rdma_cm_id *cm_id;
+ struct ds_header hdr;
+static void ds_insert_qp(struct rsocket *rs, struct ds_qp *qp)
+{
+ if (!rs->qp_list)
-+ list_init(&qp->list);
++ dlist_init(&qp->list);
+ else
-+ list_insert_head(&qp->list, &rs->qp_list->list);
-+ rs->qp_list = *qp;
++ dlist_insert_head(&qp->list, &rs->qp_list->list);
++ rs->qp_list = qp;
+}
+
+static void ds_remove_qp(struct rsocket *rs, struct ds_qp *qp)
+{
-+ if (qp->list.next != qp->list) {
++ if (qp->list.next != &qp->list) {
+ rs->qp_list = ds_next_qp(qp);
+ dlist_remove(&qp->list);
+ } else {
+
+ pthread_mutex_lock(&mut);
+ if (!svc_cnt) {
-+ ret = socketpair(AF_INET, SOCK_STREAM, 0, &svc_sock);
++ ret = socketpair(AF_INET, SOCK_STREAM, 0, svc_sock);
+ if (ret)
+ goto err1;
+
+ write(svc_sock[0], &msg, sizeof msg);
+ read(svc_sock[0], &msg, sizeof msg);
+ ret = ERR(msg.status);
-+ if (ret && !svn_cnt)
++ if (ret && !svc_cnt)
+ goto err3;
+
+ pthread_mutex_unlock(&mut);
+ write(svc_sock[0], &msg, sizeof msg);
+ read(svc_sock[0], &msg, sizeof msg);
+ ret = ERR(msg.status);
-+ if (!svn_cnt) {
++ if (!svc_cnt) {
+ pthread_join(svc_id, NULL);
+ close(svc_sock[0]);
+ close(svc_sock[1]);
+ pthread_mutex_unlock(&mut);
+ return ret;
+}
++
++static int ds_compare_addr(const void *dst1, const void *dst2)
++{
++ const struct sockaddr *sa1, *sa2;
++ size_t len;
++
++ sa1 = (const struct sockaddr *) dst1;
++ sa2 = (const struct sockaddr *) dst2;
++
++ len = (sa1->sa_family == AF_INET6 && sa2->sa_family == AF_INET6) ?
++ sizeof(struct sockaddr_in6) : sizeof(struct sockaddr);
++ return memcmp(dst1, dst2, len);
++}
+
static int rs_value_to_scale(int value, int bits)
{
return value <= (1 << (bits - 1)) ?
-@@ -307,10 +515,10 @@ out:
+@@ -307,10 +529,10 @@ out:
pthread_mutex_unlock(&mut);
}
-static int rs_insert(struct rsocket *rs)
-+static int rs_insert(struct rsocket *rs, index)
++static int rs_insert(struct rsocket *rs, int index)
{
pthread_mutex_lock(&mut);
- rs->index = idm_set(&idm, rs->cm_id->channel->fd, rs);
pthread_mutex_unlock(&mut);
return rs->index;
}
-@@ -322,7 +530,7 @@ static void rs_remove(struct rsocket *rs)
+@@ -322,7 +544,7 @@ static void rs_remove(struct rsocket *rs)
pthread_mutex_unlock(&mut);
}
{
struct rsocket *rs;
-@@ -330,7 +538,11 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs)
+@@ -330,7 +552,11 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs)
if (!rs)
return NULL;
if (inherited_rs) {
rs->sbuf_size = inherited_rs->sbuf_size;
rs->rbuf_size = inherited_rs->rbuf_size;
-@@ -352,7 +564,7 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs)
+@@ -352,7 +578,7 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs)
fastlock_init(&rs->rlock);
fastlock_init(&rs->cq_lock);
fastlock_init(&rs->cq_wait_lock);
dlist_init(&rs->iomap_list);
dlist_init(&rs->iomap_queue);
return rs;
-@@ -360,13 +572,27 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs)
+@@ -360,13 +586,27 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs)
static int rs_set_nonblocking(struct rsocket *rs, long arg)
{
return ret;
}
-@@ -390,17 +616,39 @@ static void rs_set_qp_size(struct rsocket *rs)
+@@ -390,17 +630,39 @@ static void rs_set_qp_size(struct rsocket *rs)
rs->rq_size = 2;
}
rs->rmsg = calloc(rs->rq_size + 1, sizeof(*rs->rmsg));
if (!rs->rmsg)
- return -1;
-+ return ERR(ENOEMEM);
++ return ERR(ENOMEM);
rs->sbuf = calloc(rs->sbuf_size, sizeof(*rs->sbuf));
if (!rs->sbuf)
- return -1;
-+ return ERR(ENOEMEM);
++ return ERR(ENOMEM);
rs->smr = rdma_reg_msgs(rs->cm_id, rs->sbuf, rs->sbuf_size);
if (!rs->smr)
-@@ -410,7 +658,7 @@ static int rs_init_bufs(struct rsocket *rs)
+@@ -410,7 +672,7 @@ static int rs_init_bufs(struct rsocket *rs)
sizeof(*rs->target_iomap) * rs->target_iomap_size;
rs->target_buffer_list = malloc(len);
if (!rs->target_buffer_list)
- return -1;
-+ return ERR(ENOEMEM);
++ return ERR(ENOMEM);
rs->target_mr = rdma_reg_write(rs->cm_id, rs->target_buffer_list, len);
if (!rs->target_mr)
-@@ -423,7 +671,7 @@ static int rs_init_bufs(struct rsocket *rs)
+@@ -423,7 +685,7 @@ static int rs_init_bufs(struct rsocket *rs)
rs->rbuf = calloc(rs->rbuf_size, sizeof(*rs->rbuf));
if (!rs->rbuf)
- return -1;
-+ return ERR(ENOEMEM);
++ return ERR(ENOMEM);
rs->rmr = rdma_reg_write(rs->cm_id, rs->rbuf, rs->rbuf_size);
if (!rs->rmr)
-@@ -440,15 +688,32 @@ static int rs_init_bufs(struct rsocket *rs)
+@@ -440,15 +702,32 @@ static int rs_init_bufs(struct rsocket *rs)
return 0;
}
-static int rs_create_cq(struct rsocket *rs)
+static int ds_init_bufs(struct ds_qp *qp)
-+{
+ {
+- rs->cm_id->recv_cq_channel = ibv_create_comp_channel(rs->cm_id->verbs);
+- if (!rs->cm_id->recv_cq_channel)
+ qp->rbuf = calloc(qp->rs->rbuf_size, sizeof(*qp->rbuf));
+ if (!qp->rbuf)
+ return ERR(ENOMEM);
+
+ qp->smr = rdma_reg_msgs(qp->cm_id, qp->rs->sbuf, qp->rs->sbuf_size);
+ if (!qp->smr)
-+ return -1;
-+
+ return -1;
+
+- rs->cm_id->recv_cq = ibv_create_cq(rs->cm_id->verbs, rs->sq_size + rs->rq_size,
+- rs->cm_id, rs->cm_id->recv_cq_channel, 0);
+- if (!rs->cm_id->recv_cq)
+ qp->rmr = rdma_reg_msgs(qp->cm_id, qp->rbuf, qp->rs->rbuf_size);
+ if (!qp->rmr)
+ return -1;
+}
+
+static int rs_create_cq(struct rsocket *rs, struct rdma_cm_id *cm_id)
- {
-- rs->cm_id->recv_cq_channel = ibv_create_comp_channel(rs->cm_id->verbs);
-- if (!rs->cm_id->recv_cq_channel)
++{
+ cm_id->recv_cq_channel = ibv_create_comp_channel(cm_id->verbs);
+ if (!cm_id->recv_cq_channel)
- return -1;
-
-- rs->cm_id->recv_cq = ibv_create_cq(rs->cm_id->verbs, rs->sq_size + rs->rq_size,
-- rs->cm_id, rs->cm_id->recv_cq_channel, 0);
-- if (!rs->cm_id->recv_cq)
++ return -1;
++
+ cm_id->recv_cq = ibv_create_cq(cm_id->verbs, rs->sq_size + rs->rq_size,
+ cm_id, cm_id->recv_cq_channel, 0);
+ if (!cm_id->recv_cq)
goto err1;
if (rs->fd_flags & O_NONBLOCK) {
-@@ -456,21 +721,20 @@ static int rs_create_cq(struct rsocket *rs)
+@@ -456,21 +735,20 @@ static int rs_create_cq(struct rsocket *rs)
goto err2;
}
{
struct ibv_recv_wr wr, *bad;
-@@ -482,6 +746,23 @@ rs_post_recv(struct rsocket *rs)
+@@ -482,6 +760,23 @@ rs_post_recv(struct rsocket *rs)
return rdma_seterrno(ibv_post_recv(rs->cm_id->qp, &wr, &bad));
}
+
+ sge.addr = (uintptr_t) buf;
+ sge.length = RS_SNDLOWAT;
-+ sge.lkey = qp->rmr;
++ sge.lkey = qp->rmr->lkey;
+
-+ wr.wr_id = ds_recv_wr_id((uint32_t) (buf - rs->rbuf));
++ wr.wr_id = ds_recv_wr_id((uint32_t) ((uint8_t *) buf - rs->rbuf));
+ wr.next = NULL;
+ wr.sg_list = &sge;
+ wr.num_sge = 1;
static int rs_create_ep(struct rsocket *rs)
{
struct ibv_qp_init_attr qp_attr;
-@@ -492,7 +773,7 @@ static int rs_create_ep(struct rsocket *rs)
+@@ -492,7 +787,7 @@ static int rs_create_ep(struct rsocket *rs)
if (ret)
return ret;
if (ret)
return ret;
-@@ -549,8 +830,76 @@ static void rs_free_iomappings(struct rsocket *rs)
+@@ -549,8 +844,74 @@ static void rs_free_iomappings(struct rsocket *rs)
}
}
+
+ if (qp->cm_id) {
+ if (qp->cm_id->qp) {
-+ tdelete(&qp->dest.addr, &qp->rs->dest_map, ds_compare_dest);
++ tdelete(&qp->dest.addr, &qp->rs->dest_map, ds_compare_addr);
+ epoll_ctl(qp->rs->epfd, EPOLL_CTL_DEL,
+ qp->cm_id->recv_cq_channel->fd, NULL);
+ rdma_destroy_qp(qp->cm_id);
+
+static void ds_free(struct rsocket *rs)
+{
-+ struct ds_qp *qp;
-+
+ if (rs->state & (rs_readable | rs_writable))
+ rs_remove_from_svc(rs);
+
+ if (rs->dmsg)
+ free(rs->dmsg);
+
-+ if (rs->smsg)
-+ free(rs->smsg);
++ if (rs->smsg_free)
++ free(rs->smsg_free);
+
+ while (rs->qp_list) {
+ ds_remove_qp(rs, rs->qp_list);
if (rs->index >= 0)
rs_remove(rs);
-@@ -582,7 +931,7 @@ static void rs_free(struct rsocket *rs)
+@@ -582,7 +943,7 @@ static void rs_free(struct rsocket *rs)
rdma_destroy_id(rs->cm_id);
}
fastlock_destroy(&rs->cq_wait_lock);
fastlock_destroy(&rs->cq_lock);
fastlock_destroy(&rs->rlock);
-@@ -636,29 +985,54 @@ static void rs_save_conn_data(struct rsocket *rs, struct rs_conn_data *conn)
+@@ -636,29 +997,54 @@ static void rs_save_conn_data(struct rsocket *rs, struct rs_conn_data *conn)
rs->sseq_comp = ntohs(conn->credits);
}
+ ret = ds_init(rs, domain);
+ if (ret)
+ goto err;
-
-- ret = rs_insert(rs);
++
+ index = rs->udp_sock;
+ }
-+
+
+- ret = rs_insert(rs);
+ ret = rs_insert(rs, index);
if (ret < 0)
goto err;
return rs->index;
err:
-@@ -672,9 +1046,18 @@ int rbind(int socket, const struct sockaddr *addr, socklen_t addrlen)
+@@ -672,9 +1058,18 @@ int rbind(int socket, const struct sockaddr *addr, socklen_t addrlen)
int ret;
rs = idm_at(&idm, socket);
return ret;
}
-@@ -710,7 +1093,7 @@ int raccept(int socket, struct sockaddr *addr, socklen_t *addrlen)
+@@ -710,7 +1105,7 @@ int raccept(int socket, struct sockaddr *addr, socklen_t *addrlen)
int ret;
rs = idm_at(&idm, socket);
if (!new_rs)
return ERR(ENOMEM);
-@@ -718,7 +1101,7 @@ int raccept(int socket, struct sockaddr *addr, socklen_t *addrlen)
+@@ -718,7 +1113,7 @@ int raccept(int socket, struct sockaddr *addr, socklen_t *addrlen)
if (ret)
goto err;
if (ret < 0)
goto err;
-@@ -855,13 +1238,280 @@ connected:
+@@ -855,13 +1250,268 @@ connected:
return ret;
}
+
+ rs->dmsg = calloc(rs->rq_size + 1, sizeof(*rs->dmsg));
+ if (!rs->dmsg)
-+ return ERR(ENOEMEM);
++ return ERR(ENOMEM);
+
+ rs->sbuf_bytes_avail = rs->sbuf_size;
+ rs->sqe_avail = rs->sq_size;
+ return 0;
+}
+
-+static int ds_compare_addr(const void *dst1, const void *dst2)
-+{
-+ const struct sockaddr *sa1, *sa2;
-+ size_t len;
-+
-+ sa1 = (const struct sockaddr *) dst1;
-+ sa2 = (const struct sockaddr *) dst2;
-+
-+ len = (sa1->sa_family == AF_INET6 && sa2->sa_family == AF_INET6) ?
-+ sizeof(struct sockaddr_in6) : sizeof(struct sockaddr);
-+ return memcmp(dst1, dst2, len);
-+}
-+
+static int rs_any_addr(const union socket_addr *addr)
+{
+ if (addr->sa.sa_family == AF_INET) {
-+ return (addr->sin.sin_addr == INADDR_ANY ||
-+ addr->sin.sin_addr == INADDR_LOOPBACK);
++ return (addr->sin.sin_addr.s_addr == INADDR_ANY ||
++ addr->sin.sin_addr.s_addr == INADDR_LOOPBACK);
+ } else {
-+ return (addr->sin6.sin6_addr == in6addr_any ||
-+ addr->sin6.sin6_addr == in6addr_loopback);
++ return (!memcmp(&addr->sin6.sin6_addr, &in6addr_any, 16) ||
++ !memcmp(&addr->sin6.sin6_addr, &in6addr_loopback, 16));
+ }
+}
+
+ hdr->version = 4;
+ hdr->length = DS_IPV4_HDR_LEN;
+ hdr->port = addr->sin.sin_port;
-+ hdr->addr.ipv4 = addr->sin.sin_addr;
++ hdr->addr.ipv4 = addr->sin.sin_addr.s_addr;
+ } else {
+ hdr->version = 6;
+ hdr->length = DS_IPV6_HDR_LEN;
+{
+ struct ibv_qp_init_attr qp_attr;
+ struct epoll_event event;
-+ int ret;
++ int i, ret;
+
+ *qp = calloc(1, sizeof(struct ds_qp));
+ if (!*qp)
+ if (rs->qp_list) {
+ *qp = rs->qp_list;
+ do {
-+ if (!ds_compare_addr(rdma_get_local_addr((*qp)->cm_id)), src_addr)
++ if (!ds_compare_addr(rdma_get_local_addr((*qp)->cm_id),
++ src_addr))
+ return 0;
+
+ *qp = ds_next_qp(*qp);
+ if (ret)
+ goto out;
+
-+ ret = ds_get_qp(rs, src_addr, src_len, &qp);
++ ret = ds_get_qp(rs, &src_addr, src_len, &qp);
+ if (ret)
+ goto out;
+
-+ if ((addrlen != src_len) || memcmp(addr, src_addr, addrlen)) {
++ if ((addrlen != src_len) || memcmp(addr, &src_addr, addrlen)) {
+ *dest = calloc(1, sizeof(struct ds_dest));
+ if (!*dest) {
+ ret = ERR(ENOMEM);
+
+ memcpy(&(*dest)->addr, addr, addrlen);
+ (*dest)->qp = qp;
-+ tsearch((*dest)->addr, &rs->dest_map, ds_compare_addr);
++ tsearch(&(*dest)->addr, &rs->dest_map, ds_compare_addr);
+ }
+out:
+ fastlock_release(&rs->map_lock);
+ wr.sg_list = sge;
+ wr.num_sge = 1;
+ wr.opcode = IBV_WR_SEND;
-+ wr.send_flags = (sge.length <= rs->sq_inline) ? IBV_SEND_INLINE : 0;
++ wr.send_flags = (sge->length <= rs->sq_inline) ? IBV_SEND_INLINE : 0;
+ wr.wr.ud.ah = rs->conn_dest->ah;
+ wr.wr.ud.remote_qpn = rs->conn_dest->qpn;
+ wr.wr.ud.remote_qkey = RDMA_UDP_QKEY;
rs = idm_at(&idm, socket);
+ if (rs->type == SOCK_DGRAM) {
+ fastlock_acquire(&rs->slock);
-+ ret = ds_recvfrom(rs, buf, len, flags, src_addr, addrlen);
++ ret = ds_recvfrom(rs, buf, len, flags, NULL, 0);
+ fastlock_release(&rs->slock);
+ return ret;
+ }
fastlock_release(&rs->rlock);
return ret ? ret : len - left;
-@@ -1350,6 +2254,14 @@ ssize_t rrecvfrom(int socket, void *buf, size_t len, int flags,
+@@ -1348,8 +2252,17 @@ ssize_t rrecv(int socket, void *buf, size_t len, int flags)
+ ssize_t rrecvfrom(int socket, void *buf, size_t len, int flags,
+ struct sockaddr *src_addr, socklen_t *addrlen)
{
++ struct rsocket *rs;
int ret;
+ rs = idm_at(&idm, socket);
ret = rrecv(socket, buf, len, flags);
if (ret > 0 && src_addr)
rgetpeername(socket, src_addr, addrlen);
-@@ -1391,14 +2303,14 @@ static int rs_send_iomaps(struct rsocket *rs, int flags)
+@@ -1391,14 +2304,14 @@ static int rs_send_iomaps(struct rsocket *rs, int flags)
struct rs_iomap iom;
int ret;
ret = ERR(ECONNRESET);
break;
}
-@@ -1447,10 +2359,91 @@ static int rs_send_iomaps(struct rsocket *rs, int flags)
+@@ -1447,10 +2360,90 @@ static int rs_send_iomaps(struct rsocket *rs, int flags)
}
rs->iomap_pending = !dlist_empty(&rs->iomap_queue);
+ struct ds_udp_header hdr;
+ struct msghdr msg;
+ struct iovec miov[8];
-+ struct ds_qp *qp;
+
+ if (iovcnt > 8)
+ return ERR(ENOTSUP);
+ hdr.addr.ipv4 = rs->conn_dest->qp->hdr.addr.ipv4;
+ } else {
+ hdr.length = DS_UDP_IPV6_HDR_LEN;
-+ memcpy(hdr.addr.ipv6, rs->conn_dest->qp->hdr.addr.ipv6, 16);
++ memcpy(hdr.addr.ipv6, &rs->conn_dest->qp->hdr.addr.ipv6, 16);
+ }
+
+ miov[0].iov_base = &hdr;
+ memcpy(&miov[1], iov, sizeof *iov * iovcnt);
+
+ memset(&msg, 0, sizeof msg);
-+ msg.msg_name = rs->conn_dest->addr;
++ msg.msg_name = &rs->conn_dest->addr;
+ msg.msg_namelen = ucma_addrlen(&rs->conn_dest->addr.sa);
+ msg.msg_iov = miov;
+ msg.msg_iovlen = iovcnt + 1;
-+ return sendmsg(rs->udp_sock, msg, flags);
++ return sendmsg(rs->udp_sock, &msg, flags);
+}
+
+static ssize_t ds_send_udp(struct rsocket *rs, const void *buf, size_t len,
+{
+ struct iovec iov;
+ if (buf && len) {
-+ iov.iov_base = buf;
-+ iov_iov_len = len;
++ iov.iov_base = (void *) buf;
++ iov.iov_len = len;
+ return ds_sendv_udp(rs, &iov, 1, flags, op);
+ } else {
+ return ds_sendv_udp(rs, NULL, 0, flags, op);
+ struct ds_smsg *msg;
+ struct ibv_sge sge;
+ uint64_t offset;
-+ int flags, ret = 0;
++ int ret = 0;
+
+ if (!rs->conn_dest->ah)
+ return ds_send_udp(rs, buf, len, flags, RS_OP_DATA);
+ rs->smsg_free = msg->next;
+ rs->sqe_avail--;
+
-+ memcpy((void *) msg, rs->conn_dest->qp->hdr, rs->conn_dest->qp->hdr.length);
++ memcpy((void *) msg, &rs->conn_dest->qp->hdr, rs->conn_dest->qp->hdr.length);
+ memcpy((void *) msg + rs->conn_dest->qp->hdr.length, buf, len);
+ sge.addr = (uintptr_t) msg;
+ sge.length = rs->conn_dest->qp->hdr.length + len;
ret = ERR(ECONNRESET);
break;
}
-@@ -1538,10 +2538,26 @@ out:
+@@ -1538,10 +2538,27 @@ out:
ssize_t rsendto(int socket, const void *buf, size_t len, int flags,
const struct sockaddr *dest_addr, socklen_t addrlen)
{
- if (dest_addr || addrlen)
- return ERR(EISCONN);
+ struct rsocket *rs;
++ int ret;
+
+ rs = idm_at(&idm, socket);
+ if (rs->type == SOCK_STREAM) {
+ if (dest_addr || addrlen)
+ return ERR(EISCONN);
-
-- return rsend(socket, buf, len, flags);
++
+ return rsend(socket, buf, len, flags);
+ }
-+
+
+- return rsend(socket, buf, len, flags);
+ fastlock_acquire(&rs->slock);
+ if (!rs->conn_dest || ds_compare_addr(dest_addr, &rs->conn_dest->addr)) {
-+ ret = ds_get_dest(rs, addr, addrlen, &rs->conn_dest);
++ ret = ds_get_dest(rs, dest_addr, addrlen, &rs->conn_dest);
+ if (ret)
+ goto out;
+ }
}
static void rs_copy_iov(void *dst, const struct iovec **iov, size_t *offset, size_t len)
-@@ -1600,7 +2616,7 @@ static ssize_t rsendv(int socket, const struct iovec *iov, int iovcnt, int flags
+@@ -1600,7 +2617,7 @@ static ssize_t rsendv(int socket, const struct iovec *iov, int iovcnt, int flags
rs_conn_can_send);
if (ret)
break;
ret = ERR(ECONNRESET);
break;
}
-@@ -1653,7 +2669,7 @@ ssize_t rsendmsg(int socket, const struct msghdr *msg, int flags)
+@@ -1653,7 +2670,7 @@ ssize_t rsendmsg(int socket, const struct msghdr *msg, int flags)
if (msg->msg_control && msg->msg_controllen)
return ERR(ENOTSUP);
}
ssize_t rwrite(int socket, const void *buf, size_t count)
-@@ -1690,8 +2706,8 @@ static int rs_poll_rs(struct rsocket *rs, int events,
+@@ -1690,8 +2707,8 @@ static int rs_poll_rs(struct rsocket *rs, int events,
int ret;
check_cq:
rs_process_cq(rs, nonblock, test);
revents = 0;
-@@ -1707,6 +2723,16 @@ check_cq:
+@@ -1707,6 +2724,16 @@ check_cq:
}
return revents;
}
if (rs->state == rs_listening) {
-@@ -1766,11 +2792,14 @@ static int rs_poll_arm(struct pollfd *rfds, struct pollfd *fds, nfds_t nfds)
+@@ -1766,11 +2793,14 @@ static int rs_poll_arm(struct pollfd *rfds, struct pollfd *fds, nfds_t nfds)
if (fds[i].revents)
return 1;
rfds[i].events = POLLIN;
} else {
rfds[i].fd = fds[i].fd;
-@@ -1793,7 +2822,10 @@ static int rs_poll_events(struct pollfd *rfds, struct pollfd *fds, nfds_t nfds)
+@@ -1793,7 +2823,10 @@ static int rs_poll_events(struct pollfd *rfds, struct pollfd *fds, nfds_t nfds)
rs = idm_lookup(&idm, fds[i].fd);
if (rs) {
fds[i].revents = rs_poll_rs(rs, fds[i].events, 1, rs_poll_all);
} else {
fds[i].revents = rfds[i].revents;
-@@ -1949,7 +2981,7 @@ int rshutdown(int socket, int how)
+@@ -1949,7 +2982,7 @@ int rshutdown(int socket, int how)
rs = idm_at(&idm, socket);
if (how == SHUT_RD) {
return 0;
}
-@@ -1959,10 +2991,10 @@ int rshutdown(int socket, int how)
+@@ -1959,10 +2992,10 @@ int rshutdown(int socket, int how)
if (rs->state & rs_connected) {
if (how == SHUT_RDWR) {
ctrl = RS_CTRL_DISCONNECT;
RS_CTRL_SHUTDOWN : RS_CTRL_DISCONNECT;
}
if (!rs->ctrl_avail) {
-@@ -1987,13 +3019,31 @@ int rshutdown(int socket, int how)
+@@ -1987,13 +3020,29 @@ int rshutdown(int socket, int how)
return 0;
}
+static void ds_shutdown(struct rsocket *rs)
+{
-+ int ret = 0;
-+
+ if (rs->fd_flags & O_NONBLOCK)
+ rs_set_nonblocking(rs, 0);
+
rs_free(rs);
return 0;
-@@ -2018,8 +3068,12 @@ int rgetpeername(int socket, struct sockaddr *addr, socklen_t *addrlen)
+@@ -2018,8 +3067,12 @@ int rgetpeername(int socket, struct sockaddr *addr, socklen_t *addrlen)
struct rsocket *rs;
rs = idm_at(&idm, socket);
}
int rgetsockname(int socket, struct sockaddr *addr, socklen_t *addrlen)
-@@ -2027,8 +3081,12 @@ int rgetsockname(int socket, struct sockaddr *addr, socklen_t *addrlen)
+@@ -2027,8 +3080,12 @@ int rgetsockname(int socket, struct sockaddr *addr, socklen_t *addrlen)
struct rsocket *rs;
rs = idm_at(&idm, socket);
}
int rsetsockopt(int socket, int level, int optname,
-@@ -2040,18 +3098,26 @@ int rsetsockopt(int socket, int level, int optname,
+@@ -2040,18 +3097,26 @@ int rsetsockopt(int socket, int level, int optname,
ret = ERR(ENOTSUP);
rs = idm_at(&idm, socket);
+ if (rs->type == SOCK_DGRAM && level != SOL_RDMA) {
-+ ret = setsockopt(rs->udp_sock, optname, optval, optlen);
++ ret = setsockopt(rs->udp_sock, level, optname, optval, optlen);
+ if (ret)
+ return ret;
+ }
opt_on = *(int *) optval;
break;
case SO_RCVBUF:
-@@ -2101,9 +3167,11 @@ int rsetsockopt(int socket, int level, int optname,
+@@ -2101,9 +3166,11 @@ int rsetsockopt(int socket, int level, int optname,
opts = &rs->ipv6_opts;
switch (optname) {
case IPV6_V6ONLY:
opt_on = *(int *) optval;
break;
default:
-@@ -2315,7 +3383,7 @@ off_t riomap(int socket, void *buf, size_t len, int prot, int flags, off_t offse
+@@ -2315,7 +3382,7 @@ off_t riomap(int socket, void *buf, size_t len, int prot, int flags, off_t offse
if (!rs->cm_id->pd || (prot & ~(PROT_WRITE | PROT_NONE)))
return ERR(EINVAL);
if (prot & PROT_WRITE) {
iomr = rs_get_iomap_mr(rs);
access |= IBV_ACCESS_REMOTE_WRITE;
-@@ -2349,7 +3417,7 @@ off_t riomap(int socket, void *buf, size_t len, int prot, int flags, off_t offse
+@@ -2349,7 +3416,7 @@ off_t riomap(int socket, void *buf, size_t len, int prot, int flags, off_t offse
dlist_insert_tail(&iomr->entry, &rs->iomap_list);
}
out:
return offset;
}
-@@ -2361,7 +3429,7 @@ int riounmap(int socket, void *buf, size_t len)
+@@ -2361,7 +3428,7 @@ int riounmap(int socket, void *buf, size_t len)
int ret = 0;
rs = idm_at(&idm, socket);
for (entry = rs->iomap_list.next; entry != &rs->iomap_list;
entry = entry->next) {
-@@ -2382,7 +3450,7 @@ int riounmap(int socket, void *buf, size_t len)
+@@ -2382,7 +3449,7 @@ int riounmap(int socket, void *buf, size_t len)
}
ret = ERR(EINVAL);
out:
return ret;
}
-@@ -2426,7 +3494,7 @@ size_t riowrite(int socket, const void *buf, size_t count, off_t offset, int fla
+@@ -2426,7 +3493,7 @@ size_t riowrite(int socket, const void *buf, size_t count, off_t offset, int fla
rs_conn_can_send);
if (ret)
break;
ret = ERR(ECONNRESET);
break;
}
-@@ -2476,3 +3544,269 @@ out:
+@@ -2476,3 +3543,269 @@ out:
return (ret && left == count) ? ret : count - left;
}
+ struct ibv_port_attr attr;
+
+ if (!ibv_query_port(dest->qp->cm_id->verbs, dest->qp->cm_id->port_num, &attr))
-+ return (uint8_t) ((1 << attr.lmc) - 1));
++ return (uint8_t) ((1 << attr.lmc) - 1);
+ return 0x7f;
+}
+
+static void rs_svc_create_ah(struct rsocket *rs, struct ds_dest *dest, uint32_t qpn)
+{
-+ struct socket_addr saddr;
++ union socket_addr saddr;
+ struct rdma_cm_id *id;
+ struct ibv_ah_attr attr;
+ int ret;
+ fastlock_release(&rs->slock);
+ }
+
-+ ret = rdma_create_cm_id(NULL, &id, NULL, dest->qp->cm_id->ps);
++ ret = rdma_create_id(NULL, &id, NULL, dest->qp->cm_id->ps);
+ if (ret)
+ return;
+
+ saddr.sin.sin_port = 0;
+ else
+ saddr.sin6.sin6_port = 0;
-+ ret = rdma_resolve_addr(id, &saddr, &dest->addr.sa, 2000);
++ ret = rdma_resolve_addr(id, &saddr.sa, &dest->addr.sa, 2000);
+ if (ret)
+ goto out;
+
+ attr.is_global = 1;
+ attr.grh.dgid = id->route.path_rec->dgid;
+ attr.grh.flow_label = id->route.path_rec->flow_label;
-+ attr.grh.sgid_index = rs_svc_sgid_index(dest, id->route.path_rec->sgid);
++ attr.grh.sgid_index = rs_svc_sgid_index(dest, &id->route.path_rec->sgid);
+ attr.grh.hop_limit = id->route.path_rec->hop_limit;
+ attr.grh.traffic_class = id->route.path_rec->traffic_class;
+ }
+ rdma_destroy_id(id);
+}
+
-+static int rs_svc_valid_udp_hdr(struct ds_udp_header *udp_header,
++static int rs_svc_valid_udp_hdr(struct ds_udp_header *udp_hdr,
+ union socket_addr *addr)
+{
+ return (udp_hdr->tag == DS_UDP_TAG) &&
-+ ((udp_hdr->version == 4 && addr.sa.sa_family == AF_INET &&
++ ((udp_hdr->version == 4 && addr->sa.sa_family == AF_INET &&
+ udp_hdr->length == DS_UDP_IPV4_HDR_LEN) ||
-+ (udp_hdr->version == 6 && addr.sa.sa_family == AF_INET6 &&
++ (udp_hdr->version == 6 && addr->sa.sa_family == AF_INET6 &&
+ udp_hdr->length == DS_UDP_IPV6_HDR_LEN));
+}
+
+ cur_dest = rs->conn_dest;
+ if (udp_hdr->op == RS_OP_DATA) {
+ rs->conn_dest = &dest->qp->dest;
-+ rs_svc_forward(rs, buf, len, addr);
++ rs_svc_forward(rs, svc_buf + udp_hdr->length, len, &addr);
+ }
+
+ rs->conn_dest = dest;
+ fastlock_release(&rs->slock);
+}
+
-+static int rs_svc_run(void *arg)
++static void *rs_svc_run(void *arg)
+{
+ struct rs_svc_msg msg;
+ int i, ret;
+ ret = rs_svc_grow_sets();
+ if (ret) {
+ msg.status = ret;
-+ write(svc_sock[1] &msg, sizeof msg);
-+ return ret;
++ write(svc_sock[1], &msg, sizeof msg);
++ return (void *) (uintptr_t) ret;
+ }
+
+ svc_fds[0].fd = svc_sock[1];
+ }
+ } while (svc_cnt > 1);
+
-+ return 0;
++ return NULL;
+}
+++ /dev/null
-Bottom: ac38ccc22b5116e46418ba99f1924f1c35d49ffc
-Top: ae4df7b46b3d1c519383a31cb018c70f3e4e8201
-Author: Sean Hefty <sean.hefty@intel.com>
-Date: 2012-12-05 12:53:38 -0800
-
-Refresh of dsocket
-
----
-
-diff --git a/src/rsocket.c b/src/rsocket.c
-index 018af90..9996d33 100644
---- a/src/rsocket.c
-+++ b/src/rsocket.c
-@@ -48,6 +48,7 @@
- #include <netinet/in.h>
- #include <netinet/tcp.h>
- #include <sys/epoll.h>
-+#include <search.h>
-
- #include <rdma/rdma_cma.h>
- #include <rdma/rdma_verbs.h>
-@@ -85,7 +86,7 @@ static int svc_size;
- static struct rsocket **svc_rss;
- static struct pollfd *svc_fds;
- static uint8_t svc_buf[RS_SNDLOWAT];
--static int rs_svc_run(void *arg);
-+static void *rs_svc_run(void *arg);
-
- static uint16_t def_iomap_size = 0;
- static uint16_t def_inline = 64;
-@@ -241,7 +242,7 @@ struct ds_dest {
- };
-
- struct ds_qp {
-- dlist_t list;
-+ dlist_entry list;
- struct rsocket *rs;
- struct rdma_cm_id *cm_id;
- struct ds_header hdr;
-@@ -362,15 +363,15 @@ struct ds_udp_header {
- static void ds_insert_qp(struct rsocket *rs, struct ds_qp *qp)
- {
- if (!rs->qp_list)
-- list_init(&qp->list);
-+ dlist_init(&qp->list);
- else
-- list_insert_head(&qp->list, &rs->qp_list->list);
-- rs->qp_list = *qp;
-+ dlist_insert_head(&qp->list, &rs->qp_list->list);
-+ rs->qp_list = qp;
- }
-
- static void ds_remove_qp(struct rsocket *rs, struct ds_qp *qp)
- {
-- if (qp->list.next != qp->list) {
-+ if (qp->list.next != &qp->list) {
- rs->qp_list = ds_next_qp(qp);
- dlist_remove(&qp->list);
- } else {
-@@ -385,7 +386,7 @@ static int rs_add_to_svc(struct rsocket *rs)
-
- pthread_mutex_lock(&mut);
- if (!svc_cnt) {
-- ret = socketpair(AF_INET, SOCK_STREAM, 0, &svc_sock);
-+ ret = socketpair(AF_INET, SOCK_STREAM, 0, svc_sock);
- if (ret)
- goto err1;
-
-@@ -402,7 +403,7 @@ static int rs_add_to_svc(struct rsocket *rs)
- write(svc_sock[0], &msg, sizeof msg);
- read(svc_sock[0], &msg, sizeof msg);
- ret = ERR(msg.status);
-- if (ret && !svn_cnt)
-+ if (ret && !svc_cnt)
- goto err3;
-
- pthread_mutex_unlock(&mut);
-@@ -430,7 +431,7 @@ static int rs_remove_from_svc(struct rsocket *rs)
- write(svc_sock[0], &msg, sizeof msg);
- read(svc_sock[0], &msg, sizeof msg);
- ret = ERR(msg.status);
-- if (!svn_cnt) {
-+ if (!svc_cnt) {
- pthread_join(svc_id, NULL);
- close(svc_sock[0]);
- close(svc_sock[1]);
-@@ -440,6 +441,19 @@ static int rs_remove_from_svc(struct rsocket *rs)
- return ret;
- }
-
-+static int ds_compare_addr(const void *dst1, const void *dst2)
-+{
-+ const struct sockaddr *sa1, *sa2;
-+ size_t len;
-+
-+ sa1 = (const struct sockaddr *) dst1;
-+ sa2 = (const struct sockaddr *) dst2;
-+
-+ len = (sa1->sa_family == AF_INET6 && sa2->sa_family == AF_INET6) ?
-+ sizeof(struct sockaddr_in6) : sizeof(struct sockaddr);
-+ return memcmp(dst1, dst2, len);
-+}
-+
- static int rs_value_to_scale(int value, int bits)
- {
- return value <= (1 << (bits - 1)) ?
-@@ -515,7 +529,7 @@ out:
- pthread_mutex_unlock(&mut);
- }
-
--static int rs_insert(struct rsocket *rs, index)
-+static int rs_insert(struct rsocket *rs, int index)
- {
- pthread_mutex_lock(&mut);
- rs->index = idm_set(&idm, index, rs);
-@@ -644,11 +658,11 @@ static int rs_init_bufs(struct rsocket *rs)
-
- rs->rmsg = calloc(rs->rq_size + 1, sizeof(*rs->rmsg));
- if (!rs->rmsg)
-- return ERR(ENOEMEM);
-+ return ERR(ENOMEM);
-
- rs->sbuf = calloc(rs->sbuf_size, sizeof(*rs->sbuf));
- if (!rs->sbuf)
-- return ERR(ENOEMEM);
-+ return ERR(ENOMEM);
-
- rs->smr = rdma_reg_msgs(rs->cm_id, rs->sbuf, rs->sbuf_size);
- if (!rs->smr)
-@@ -658,7 +672,7 @@ static int rs_init_bufs(struct rsocket *rs)
- sizeof(*rs->target_iomap) * rs->target_iomap_size;
- rs->target_buffer_list = malloc(len);
- if (!rs->target_buffer_list)
-- return ERR(ENOEMEM);
-+ return ERR(ENOMEM);
-
- rs->target_mr = rdma_reg_write(rs->cm_id, rs->target_buffer_list, len);
- if (!rs->target_mr)
-@@ -671,7 +685,7 @@ static int rs_init_bufs(struct rsocket *rs)
-
- rs->rbuf = calloc(rs->rbuf_size, sizeof(*rs->rbuf));
- if (!rs->rbuf)
-- return ERR(ENOEMEM);
-+ return ERR(ENOMEM);
-
- rs->rmr = rdma_reg_write(rs->cm_id, rs->rbuf, rs->rbuf_size);
- if (!rs->rmr)
-@@ -753,9 +767,9 @@ static inline int ds_post_recv(struct rsocket *rs, struct ds_qp *qp, void *buf)
-
- sge.addr = (uintptr_t) buf;
- sge.length = RS_SNDLOWAT;
-- sge.lkey = qp->rmr;
-+ sge.lkey = qp->rmr->lkey;
-
-- wr.wr_id = ds_recv_wr_id((uint32_t) (buf - rs->rbuf));
-+ wr.wr_id = ds_recv_wr_id((uint32_t) ((uint8_t *) buf - rs->rbuf));
- wr.next = NULL;
- wr.sg_list = &sge;
- wr.num_sge = 1;
-@@ -843,7 +857,7 @@ static void ds_free_qp(struct ds_qp *qp)
-
- if (qp->cm_id) {
- if (qp->cm_id->qp) {
-- tdelete(&qp->dest.addr, &qp->rs->dest_map, ds_compare_dest);
-+ tdelete(&qp->dest.addr, &qp->rs->dest_map, ds_compare_addr);
- epoll_ctl(qp->rs->epfd, EPOLL_CTL_DEL,
- qp->cm_id->recv_cq_channel->fd, NULL);
- rdma_destroy_qp(qp->cm_id);
-@@ -856,8 +870,6 @@ static void ds_free_qp(struct ds_qp *qp)
-
- static void ds_free(struct rsocket *rs)
- {
-- struct ds_qp *qp;
--
- if (rs->state & (rs_readable | rs_writable))
- rs_remove_from_svc(rs);
-
-@@ -870,8 +882,8 @@ static void ds_free(struct rsocket *rs)
- if (rs->dmsg)
- free(rs->dmsg);
-
-- if (rs->smsg)
-- free(rs->smsg);
-+ if (rs->smsg_free)
-+ free(rs->smsg_free);
-
- while (rs->qp_list) {
- ds_remove_qp(rs, rs->qp_list);
-@@ -1251,7 +1263,7 @@ static int ds_init_ep(struct rsocket *rs)
-
- rs->dmsg = calloc(rs->rq_size + 1, sizeof(*rs->dmsg));
- if (!rs->dmsg)
-- return ERR(ENOEMEM);
-+ return ERR(ENOMEM);
-
- rs->sbuf_bytes_avail = rs->sbuf_size;
- rs->sqe_avail = rs->sq_size;
-@@ -1273,27 +1285,14 @@ static int ds_init_ep(struct rsocket *rs)
- return 0;
- }
-
--static int ds_compare_addr(const void *dst1, const void *dst2)
--{
-- const struct sockaddr *sa1, *sa2;
-- size_t len;
--
-- sa1 = (const struct sockaddr *) dst1;
-- sa2 = (const struct sockaddr *) dst2;
--
-- len = (sa1->sa_family == AF_INET6 && sa2->sa_family == AF_INET6) ?
-- sizeof(struct sockaddr_in6) : sizeof(struct sockaddr);
-- return memcmp(dst1, dst2, len);
--}
--
- static int rs_any_addr(const union socket_addr *addr)
- {
- if (addr->sa.sa_family == AF_INET) {
-- return (addr->sin.sin_addr == INADDR_ANY ||
-- addr->sin.sin_addr == INADDR_LOOPBACK);
-+ return (addr->sin.sin_addr.s_addr == INADDR_ANY ||
-+ addr->sin.sin_addr.s_addr == INADDR_LOOPBACK);
- } else {
-- return (addr->sin6.sin6_addr == in6addr_any ||
-- addr->sin6.sin6_addr == in6addr_loopback);
-+ return (!memcmp(&addr->sin6.sin6_addr, &in6addr_any, 16) ||
-+ !memcmp(&addr->sin6.sin6_addr, &in6addr_loopback, 16));
- }
- }
-
-@@ -1332,7 +1331,7 @@ static void ds_format_hdr(struct ds_header *hdr, union socket_addr *addr)
- hdr->version = 4;
- hdr->length = DS_IPV4_HDR_LEN;
- hdr->port = addr->sin.sin_port;
-- hdr->addr.ipv4 = addr->sin.sin_addr;
-+ hdr->addr.ipv4 = addr->sin.sin_addr.s_addr;
- } else {
- hdr->version = 6;
- hdr->length = DS_IPV6_HDR_LEN;
-@@ -1373,7 +1372,7 @@ static int ds_create_qp(struct rsocket *rs, union socket_addr *src_addr,
- {
- struct ibv_qp_init_attr qp_attr;
- struct epoll_event event;
-- int ret;
-+ int i, ret;
-
- *qp = calloc(1, sizeof(struct ds_qp));
- if (!*qp)
-@@ -1442,7 +1441,8 @@ static int ds_get_qp(struct rsocket *rs, union socket_addr *src_addr,
- if (rs->qp_list) {
- *qp = rs->qp_list;
- do {
-- if (!ds_compare_addr(rdma_get_local_addr((*qp)->cm_id)), src_addr)
-+ if (!ds_compare_addr(rdma_get_local_addr((*qp)->cm_id),
-+ src_addr))
- return 0;
-
- *qp = ds_next_qp(*qp);
-@@ -1475,11 +1475,11 @@ static int ds_get_dest(struct rsocket *rs, const struct sockaddr *addr,
- if (ret)
- goto out;
-
-- ret = ds_get_qp(rs, src_addr, src_len, &qp);
-+ ret = ds_get_qp(rs, &src_addr, src_len, &qp);
- if (ret)
- goto out;
-
-- if ((addrlen != src_len) || memcmp(addr, src_addr, addrlen)) {
-+ if ((addrlen != src_len) || memcmp(addr, &src_addr, addrlen)) {
- *dest = calloc(1, sizeof(struct ds_dest));
- if (!*dest) {
- ret = ERR(ENOMEM);
-@@ -1488,7 +1488,7 @@ static int ds_get_dest(struct rsocket *rs, const struct sockaddr *addr,
-
- memcpy(&(*dest)->addr, addr, addrlen);
- (*dest)->qp = qp;
-- tsearch((*dest)->addr, &rs->dest_map, ds_compare_addr);
-+ tsearch(&(*dest)->addr, &rs->dest_map, ds_compare_addr);
- }
- out:
- fastlock_release(&rs->map_lock);
-@@ -1563,7 +1563,7 @@ static int ds_post_send(struct rsocket *rs, struct ibv_sge *sge,
- wr.sg_list = sge;
- wr.num_sge = 1;
- wr.opcode = IBV_WR_SEND;
-- wr.send_flags = (sge.length <= rs->sq_inline) ? IBV_SEND_INLINE : 0;
-+ wr.send_flags = (sge->length <= rs->sq_inline) ? IBV_SEND_INLINE : 0;
- wr.wr.ud.ah = rs->conn_dest->ah;
- wr.wr.ud.remote_qpn = rs->conn_dest->qpn;
- wr.wr.ud.remote_qkey = RDMA_UDP_QKEY;
-@@ -2189,7 +2189,7 @@ ssize_t rrecv(int socket, void *buf, size_t len, int flags)
- rs = idm_at(&idm, socket);
- if (rs->type == SOCK_DGRAM) {
- fastlock_acquire(&rs->slock);
-- ret = ds_recvfrom(rs, buf, len, flags, src_addr, addrlen);
-+ ret = ds_recvfrom(rs, buf, len, flags, NULL, 0);
- fastlock_release(&rs->slock);
- return ret;
- }
-@@ -2252,6 +2252,7 @@ ssize_t rrecv(int socket, void *buf, size_t len, int flags)
- ssize_t rrecvfrom(int socket, void *buf, size_t len, int flags,
- struct sockaddr *src_addr, socklen_t *addrlen)
- {
-+ struct rsocket *rs;
- int ret;
-
- rs = idm_at(&idm, socket);
-@@ -2369,7 +2370,6 @@ static ssize_t ds_sendv_udp(struct rsocket *rs, const struct iovec *iov,
- struct ds_udp_header hdr;
- struct msghdr msg;
- struct iovec miov[8];
-- struct ds_qp *qp;
-
- if (iovcnt > 8)
- return ERR(ENOTSUP);
-@@ -2384,7 +2384,7 @@ static ssize_t ds_sendv_udp(struct rsocket *rs, const struct iovec *iov,
- hdr.addr.ipv4 = rs->conn_dest->qp->hdr.addr.ipv4;
- } else {
- hdr.length = DS_UDP_IPV6_HDR_LEN;
-- memcpy(hdr.addr.ipv6, rs->conn_dest->qp->hdr.addr.ipv6, 16);
-+ memcpy(hdr.addr.ipv6, &rs->conn_dest->qp->hdr.addr.ipv6, 16);
- }
-
- miov[0].iov_base = &hdr;
-@@ -2393,11 +2393,11 @@ static ssize_t ds_sendv_udp(struct rsocket *rs, const struct iovec *iov,
- memcpy(&miov[1], iov, sizeof *iov * iovcnt);
-
- memset(&msg, 0, sizeof msg);
-- msg.msg_name = rs->conn_dest->addr;
-+ msg.msg_name = &rs->conn_dest->addr;
- msg.msg_namelen = ucma_addrlen(&rs->conn_dest->addr.sa);
- msg.msg_iov = miov;
- msg.msg_iovlen = iovcnt + 1;
-- return sendmsg(rs->udp_sock, msg, flags);
-+ return sendmsg(rs->udp_sock, &msg, flags);
- }
-
- static ssize_t ds_send_udp(struct rsocket *rs, const void *buf, size_t len,
-@@ -2405,8 +2405,8 @@ static ssize_t ds_send_udp(struct rsocket *rs, const void *buf, size_t len,
- {
- struct iovec iov;
- if (buf && len) {
-- iov.iov_base = buf;
-- iov_iov_len = len;
-+ iov.iov_base = (void *) buf;
-+ iov.iov_len = len;
- return ds_sendv_udp(rs, &iov, 1, flags, op);
- } else {
- return ds_sendv_udp(rs, NULL, 0, flags, op);
-@@ -2418,7 +2418,7 @@ static ssize_t dsend(struct rsocket *rs, const void *buf, size_t len, int flags)
- struct ds_smsg *msg;
- struct ibv_sge sge;
- uint64_t offset;
-- int flags, ret = 0;
-+ int ret = 0;
-
- if (!rs->conn_dest->ah)
- return ds_send_udp(rs, buf, len, flags, RS_OP_DATA);
-@@ -2433,7 +2433,7 @@ static ssize_t dsend(struct rsocket *rs, const void *buf, size_t len, int flags)
- rs->smsg_free = msg->next;
- rs->sqe_avail--;
-
-- memcpy((void *) msg, rs->conn_dest->qp->hdr, rs->conn_dest->qp->hdr.length);
-+ memcpy((void *) msg, &rs->conn_dest->qp->hdr, rs->conn_dest->qp->hdr.length);
- memcpy((void *) msg + rs->conn_dest->qp->hdr.length, buf, len);
- sge.addr = (uintptr_t) msg;
- sge.length = rs->conn_dest->qp->hdr.length + len;
-@@ -2539,6 +2539,7 @@ ssize_t rsendto(int socket, const void *buf, size_t len, int flags,
- const struct sockaddr *dest_addr, socklen_t addrlen)
- {
- struct rsocket *rs;
-+ int ret;
-
- rs = idm_at(&idm, socket);
- if (rs->type == SOCK_STREAM) {
-@@ -2550,7 +2551,7 @@ ssize_t rsendto(int socket, const void *buf, size_t len, int flags,
-
- fastlock_acquire(&rs->slock);
- if (!rs->conn_dest || ds_compare_addr(dest_addr, &rs->conn_dest->addr)) {
-- ret = ds_get_dest(rs, addr, addrlen, &rs->conn_dest);
-+ ret = ds_get_dest(rs, dest_addr, addrlen, &rs->conn_dest);
- if (ret)
- goto out;
- }
-@@ -3021,8 +3022,6 @@ int rshutdown(int socket, int how)
-
- static void ds_shutdown(struct rsocket *rs)
- {
-- int ret = 0;
--
- if (rs->fd_flags & O_NONBLOCK)
- rs_set_nonblocking(rs, 0);
-
-@@ -3099,7 +3098,7 @@ int rsetsockopt(int socket, int level, int optname,
- ret = ERR(ENOTSUP);
- rs = idm_at(&idm, socket);
- if (rs->type == SOCK_DGRAM && level != SOL_RDMA) {
-- ret = setsockopt(rs->udp_sock, optname, optval, optlen);
-+ ret = setsockopt(rs->udp_sock, level, optname, optval, optlen);
- if (ret)
- return ret;
- }
-@@ -3643,13 +3642,13 @@ static uint8_t rs_svc_path_bits(struct ds_dest *dest)
- struct ibv_port_attr attr;
-
- if (!ibv_query_port(dest->qp->cm_id->verbs, dest->qp->cm_id->port_num, &attr))
-- return (uint8_t) ((1 << attr.lmc) - 1));
-+ return (uint8_t) ((1 << attr.lmc) - 1);
- return 0x7f;
- }
-
- static void rs_svc_create_ah(struct rsocket *rs, struct ds_dest *dest, uint32_t qpn)
- {
-- struct socket_addr saddr;
-+ union socket_addr saddr;
- struct rdma_cm_id *id;
- struct ibv_ah_attr attr;
- int ret;
-@@ -3661,7 +3660,7 @@ static void rs_svc_create_ah(struct rsocket *rs, struct ds_dest *dest, uint32_t
- fastlock_release(&rs->slock);
- }
-
-- ret = rdma_create_cm_id(NULL, &id, NULL, dest->qp->cm_id->ps);
-+ ret = rdma_create_id(NULL, &id, NULL, dest->qp->cm_id->ps);
- if (ret)
- return;
-
-@@ -3671,7 +3670,7 @@ static void rs_svc_create_ah(struct rsocket *rs, struct ds_dest *dest, uint32_t
- saddr.sin.sin_port = 0;
- else
- saddr.sin6.sin6_port = 0;
-- ret = rdma_resolve_addr(id, &saddr, &dest->addr.sa, 2000);
-+ ret = rdma_resolve_addr(id, &saddr.sa, &dest->addr.sa, 2000);
- if (ret)
- goto out;
-
-@@ -3684,7 +3683,7 @@ static void rs_svc_create_ah(struct rsocket *rs, struct ds_dest *dest, uint32_t
- attr.is_global = 1;
- attr.grh.dgid = id->route.path_rec->dgid;
- attr.grh.flow_label = id->route.path_rec->flow_label;
-- attr.grh.sgid_index = rs_svc_sgid_index(dest, id->route.path_rec->sgid);
-+ attr.grh.sgid_index = rs_svc_sgid_index(dest, &id->route.path_rec->sgid);
- attr.grh.hop_limit = id->route.path_rec->hop_limit;
- attr.grh.traffic_class = id->route.path_rec->traffic_class;
- }
-@@ -3702,13 +3701,13 @@ out:
- rdma_destroy_id(id);
- }
-
--static int rs_svc_valid_udp_hdr(struct ds_udp_header *udp_header,
-+static int rs_svc_valid_udp_hdr(struct ds_udp_header *udp_hdr,
- union socket_addr *addr)
- {
- return (udp_hdr->tag == DS_UDP_TAG) &&
-- ((udp_hdr->version == 4 && addr.sa.sa_family == AF_INET &&
-+ ((udp_hdr->version == 4 && addr->sa.sa_family == AF_INET &&
- udp_hdr->length == DS_UDP_IPV4_HDR_LEN) ||
-- (udp_hdr->version == 6 && addr.sa.sa_family == AF_INET6 &&
-+ (udp_hdr->version == 6 && addr->sa.sa_family == AF_INET6 &&
- udp_hdr->length == DS_UDP_IPV6_HDR_LEN));
- }
-
-@@ -3771,7 +3770,7 @@ static void rs_svc_process_rs(struct rsocket *rs)
- cur_dest = rs->conn_dest;
- if (udp_hdr->op == RS_OP_DATA) {
- rs->conn_dest = &dest->qp->dest;
-- rs_svc_forward(rs, buf, len, addr);
-+ rs_svc_forward(rs, svc_buf + udp_hdr->length, len, &addr);
- }
-
- rs->conn_dest = dest;
-@@ -3780,7 +3779,7 @@ static void rs_svc_process_rs(struct rsocket *rs)
- fastlock_release(&rs->slock);
- }
-
--static int rs_svc_run(void *arg)
-+static void *rs_svc_run(void *arg)
- {
- struct rs_svc_msg msg;
- int i, ret;
-@@ -3788,8 +3787,8 @@ static int rs_svc_run(void *arg)
- ret = rs_svc_grow_sets();
- if (ret) {
- msg.status = ret;
-- write(svc_sock[1] &msg, sizeof msg);
-- return ret;
-+ write(svc_sock[1], &msg, sizeof msg);
-+ return (void *) (uintptr_t) ret;
- }
-
- svc_fds[0].fd = svc_sock[1];
-@@ -3808,5 +3807,5 @@ static int rs_svc_run(void *arg)
- }
- } while (svc_cnt > 1);
-
-- return 0;
-+ return NULL;
- }
projects / ~shefty / librdmacm.git / commitdiff