#define RS_OLAP_START_SIZE 2048
#define RS_MAX_TRANSFER 65536
-#define RS_SNDLOWAT 64
+#define RS_SNDLOWAT 2048
#define RS_QP_MAX_SIZE 0xFFFE
#define RS_QP_CTRL_SIZE 4
#define RS_CONN_RETRIES 6
static struct index_map idm;
static pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER;
+enum {
+ RS_SVC_INSERT,
+ RS_SVC_REMOVE
+};
+
+struct rsocket;
+
+struct rs_svc_msg {
+ uint32_t op;
+ uint32_t status;
+ struct rsocket *rs;
+};
+
+static pthread_t svc_id;
+static int svc_cnt;
+static int svc_fds[2];
+
static uint16_t def_iomap_size = 0;
static uint16_t def_inline = 64;
static uint16_t def_sqsize = 384;
rs_connecting = rs_opening | 0x0040,
rs_accepting = rs_opening | 0x0080,
rs_connected = 0x0100,
- rs_connect_wr = 0x0200,
- rs_connect_rd = 0x0400,
- rs_connect_rdwr = rs_connected | rs_connect_rd | rs_connect_wr,
+ rs_writable = 0x0200,
+ rs_readable = 0x0400,
+ rs_connect_rdwr = rs_connected | rs_readable | rs_writable,
rs_connect_error = 0x0800,
rs_disconnected = 0x1000,
rs_error = 0x2000,
};
struct ds_qp {
+ dlist_t list;
+ struct rsocket *rs;
struct rdma_cm_id *cm_id;
- int rbuf_cnt;
+
+ struct ibv_mr *rmr;
+ uint8_t *rbuf;
+
+ struct ibv_mr *smr;
uint16_t lid;
uint8_t sl;
};
struct ds_dest {
- union socket_addr addr;
+ union socket_addr addr; /* must be first */
struct ds_qp *qp;
struct ibv_ah *ah;
uint32_t qpn;
- atomic_t refcnt;
};
struct rsocket {
};
/* datagram */
struct {
- struct ds_qp *qp_array;
+ dlist_t qp_list;
void *dest_map;
struct ds_dest *conn_dest;
struct pollfd *fds;
nfds_t nfds;
- int fd;
+ int dsock;
int sbytes_needed;
};
};
};
+static int rs_svc_run(void *arg)
+{
+ return 0;
+}
+
+static int rs_svc_insert(struct rsocket *rs)
+{
+ struct rs_svc_msg msg;
+ int ret;
+
+ pthread_mutex_lock(&mut);
+ if (!svc_cnt) {
+ ret = socketpair(AF_INET, SOCK_STREAM, 0, &svc_fds);
+ if (ret)
+ goto out;
+
+ ret = pthread_create(&svc_id, NULL, rs_svc_run, NULL);
+ if (ret) {
+ close(svc_fds[0]);
+ close(svc_fds[1]);
+ ret = ERR(ret);
+ goto out;
+ }
+ }
+
+ msg.op = RS_SVC_INSERT;
+ msg.status = EINVAL;
+ msg.rs = rs;
+ svc_cnt++;
+ write(svc_fds[0], &msg, sizeof msg);
+ read(svc_fds[0], &msg, sizeof msg);
+ ret = ERR(msg.status);
+out:
+ pthread_mutex_unlock(&mut);
+ return ret;
+}
+
+static int rs_svc_remove(struct rsocket *rs)
+{
+ struct rs_svc_msg msg;
+ int ret;
+
+ pthread_mutex_lock(&mut);
+ msg.op = RS_SVC_REMOVE;
+ msg.status = EINVAL;
+ msg.rs = rs;
+ write(svc_fds[0], &msg, sizeof msg);
+ read(svc_fds[0], &msg, sizeof msg);
+ ret = ERR(msg.status);
+ if (!ret && !--svn_cnt)
+ pthread_join(svc_id, NULL);
+
+ pthread_mutex_unlock(&mut);
+ return ret;
+}
+
static int rs_value_to_scale(int value, int bits)
{
return value <= (1 << (bits - 1)) ?
rs->type = type;
rs->index = -1;
- rs->fd = -1;
+ rs->dsock = -1;
if (inherited_rs) {
rs->sbuf_size = inherited_rs->sbuf_size;
rs->rbuf_size = inherited_rs->rbuf_size;
return rs;
}
+/*
+ * TODO: Support datagram rsockets
+ */
static int rs_set_nonblocking(struct rsocket *rs, long arg)
{
int ret = 0;
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;
+
+ qp->rmr = rdma_reg_msgs(qp->cm_id, qp->rbuf, qp->rs->rbuf_size);
+ if (!qp->rmr)
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 0;
+}
+
+static int rs_create_cq(struct rsocket *rs, struct rdma_cm_id *cm_id)
+{
+ cm_id->recv_cq_channel = ibv_create_comp_channel(cm_id->verbs);
+ if (!cm_id->recv_cq_channel)
+ 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) {
goto err2;
}
- rs->cm_id->send_cq_channel = rs->cm_id->recv_cq_channel;
- rs->cm_id->send_cq = rs->cm_id->recv_cq;
+ cm_id->send_cq_channel = cm_id->recv_cq_channel;
+ cm_id->send_cq = cm_id->recv_cq;
return 0;
err2:
- ibv_destroy_cq(rs->cm_id->recv_cq);
- rs->cm_id->recv_cq = NULL;
+ ibv_destroy_cq(cm_id->recv_cq);
+ cm_id->recv_cq = NULL;
err1:
- ibv_destroy_comp_channel(rs->cm_id->recv_cq_channel);
- rs->cm_id->recv_cq_channel = NULL;
+ ibv_destroy_comp_channel(cm_id->recv_cq_channel);
+ cm_id->recv_cq_channel = NULL;
return -1;
}
-static inline int
-rs_post_recv(struct rsocket *rs)
+static inline int rs_post_recv(struct rsocket *rs)
{
struct ibv_recv_wr wr, *bad;
return rdma_seterrno(ibv_post_recv(rs->cm_id->qp, &wr, &bad));
}
+static inline int ds_post_recv(struct rsocket *rs, struct ds_qp *qp, void *buf)
+{
+ struct ibv_recv_wr wr, *bad;
+ struct ibv_sge sge;
+
+ sge.addr = (uintptr_t) buf;
+ sge.length = RS_SNDLOWAT;
+ sge.lkey = qp->rmr;
+
+ wr.wr_id = RS_RECV_WR_ID;
+ wr.next = NULL;
+ wr.sg_list = &sge;
+ wr.num_sge = 1;
+
+ return rdma_seterrno(ibv_post_recv(qp->cm_id->qp, &wr, &bad));
+}
+
static int rs_create_ep(struct rsocket *rs)
{
struct ibv_qp_init_attr qp_attr;
if (ret)
return ret;
- ret = rs_create_cq(rs);
+ ret = rs_create_cq(rs, rs->cm_id);
if (ret)
return ret;
}
}
+static void ds_free_qp(struct ds_qp *qp)
+{
+ if (qp->smr)
+ rdma_dereg_mr(qp->smr);
+
+ if (qp->rbuf) {
+ if (qp->rmr)
+ rdma_dereg_mr(qp->rmr);
+ free(qp->rbuf);
+ }
+
+ if (qp->cm_id) {
+ if (qp->cm_id->qp) {
+ dlist_remove(&qp->list);
+ rdma_destroy_qp(qp->cm_id);
+ }
+ rdma_destroy_id(qp->cm_id);
+ }
+ free(qp);
+}
+
+static void ds_free_qps(struct rsocket *rs)
+{
+ struct ds_qp *qp;
+ dlist_t *entry;
+
+ while (!dlist_empty(&rs->qp_list)) {
+ qp = container_of(rs->qp_list.next, struct ds_qp, list);
+ ds_free_qp(qp);
+ }
+}
+
+static void ds_free(struct rsocket *rs)
+{
+ if (rs->state & (rs_readable | rs_writable))
+ rs_svc_remove(rs);
+
+ if (rs->dsock >= 0)
+ close(rs->dsock);
+
+ if (rs->index >= 0)
+ rs_remove(rs);
+
+ ds_free_qps(rs);
+ if (rs->fds)
+ free(rs->fds);
+
+ if (rs->sbuf)
+ free(rs->sbuf);
+
+ fastlock_destroy(&rs->map_lock);
+ fastlock_destroy(&rs->cq_wait_lock);
+ fastlock_destroy(&rs->cq_lock);
+ fastlock_destroy(&rs->rlock);
+ fastlock_destroy(&rs->slock);
+ free(rs);
+}
+
static void rs_free(struct rsocket *rs)
{
+ if (rs->type == SOCK_DGRAM) {
+ ds_free(rs);
+ return;
+ }
+
if (rs->index >= 0)
rs_remove(rs);
rdma_destroy_id(rs->cm_id);
}
- if (rs->fd >= 0)
- close(rs->fd);
- if (rs->fds)
- free(rs->fds);
-
fastlock_destroy(&rs->map_lock);
fastlock_destroy(&rs->cq_wait_lock);
fastlock_destroy(&rs->cq_lock);
static int ds_init(struct rsocket *rs, int domain)
{
- rs->fd = socket(domain, SOCK_DGRAM, 0);
- if (rs->fd < 0)
- return rs->fd;
+ rs->dsock = socket(domain, SOCK_DGRAM, 0);
+ if (rs->dsock < 0)
+ return rs->dsock;
rs->fds = calloc(1, sizeof *fds);
if (!rs->fds)
return ERR(ENOMEM);
rs->nfds = 1;
+ dlist_init(&rs->qp_list);
return 0;
}
if (ret)
goto err;
- index = rs->fd;
+ index = rs->dsock;
}
ret = rs_insert(rs, index);
if (!ret)
rs->state = rs_bound;
} else {
- ret = bind(rs->fd, addr, addrlen);
+ ret = bind(rs->dsock, addr, addrlen);
+ if (!ret) {
+ ret = rs_svc_insert(rs);
+ if (!ret)
+ rs->state = rs_readable | rs_writable;
+ }
}
return ret;
}
return ret;
}
-static int ds_compare_dest(const void *dst1, const void *dst2)
+static int ds_init_ep(struct rsocket *rs)
+{
+ int ret;
+
+ rs_set_qp_size(rs);
+ if (rs->rq_size > (rs->rbuf_size / RS_SNDLOWAT))
+ rs->rq_size = rs->rbuf_size / RS_SNDLOWAT;
+ else
+ rs->rbuf_size = rs->rq_size * RS_SNDLOWAT;
+
+ rs->sbuf = calloc(rs->sbuf_size, sizeof(*rs->sbuf));
+ if (!rs->sbuf)
+ return ERR(ENOMEM);
+
+ rs->ssgl[0].addr = rs->ssgl[1].addr = (uintptr_t) rs->sbuf;
+ rs->sbuf_bytes_avail = rs->sbuf_size;
+ rs->sqe_avail = rs->sq_size;
+
+ ret = rs_svc_insert(rs);
+ if (ret)
+ return ret;
+
+ rs->state = rs_readable | rs_writable;
+ return 0;
+}
+
+static int ds_compare_addr(const void *dst1, const void *dst2)
{
const struct sockaddr *sa1, *sa2;
size_t len;
return memcmp(dst1, dst2, len);
}
-/* Caller must hold map_lock around accessing source address */
-static union socket_addr *ds_get_src_addr(const struct sockaddr *dst_addr,
- socklen_t dst_len, socklen_t *src_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);
+ } else {
+ return (addr->sin6.sin6_addr == in6addr_any ||
+ addr->sin6.sin6_addr == in6addr_loopback);
+ }
+}
+
+static int ds_get_src_addr(struct rsocket *rs,
+ const struct sockaddr *dest_addr, socklen_t dest_len,
+ union socket_addr *src_addr, socklen_t *src_len)
{
- static union socket_addr src_addr;
int sock, ret;
+ uint16_t port;
- sock = socket(dst_addr->sa.sa_family, SOCK_DGRAM, 0);
+ *src_len = sizeof src_addr;
+ ret = getsockname(rs->dsock, &src_addr->sa, src_len);
+ if (ret || !rs_any_addr(src_addr))
+ return ret;
+
+ port = src_addr->sin.sin_port;
+ sock = socket(dest_addr->sa_family, SOCK_DGRAM, 0);
if (sock < 0)
- return NULL;
+ return sock;
- ret = connect(sock, &dst_addr->sa, dst_len);
+ ret = connect(sock, dest_addr, dest_len);
if (ret)
goto out;
*src_len = sizeof src_addr;
- ret = getsockname(sock, &src_addr.sa, src_len);
-
+ ret = getsockname(sock, &src_addr->sa, src_len);
+ src_addr->sin.sin_port = port;
out:
close(sock);
- return ret ? NULL : &src_addr;
+ return ret;
+}
+
+static int ds_create_qp(struct rsocket *rs, union socket_addr *src_addr,
+ socklen_t addrlen, struct ds_qp **qp)
+{
+ struct ibv_qp_init_attr qp_attr;
+ int ret;
+
+ *qp = calloc(1, sizeof(struct ds_qp));
+ if (!*qp)
+ return ERR(ENOMEM);
+
+ (*qp)->rs = rs;
+ ret = rdma_create_id(NULL, &(*qp)->cm_id, *qp, RDMA_PS_UDP);
+ if (ret)
+ goto err;
+
+ ret = rdma_bind_addr((*qp)->cm_id, &src_addr->sa);
+ if (ret)
+ goto err;
+
+ ret = ds_init_bufs(*qp);
+ if (ret)
+ goto err;
+
+ ret = rs_create_cq(rs, (*qp)->cm_id);
+ if (ret)
+ goto err;
+
+ memset(&qp_attr, 0, sizeof qp_attr);
+ qp_attr.qp_context = qp;
+ qp_attr.send_cq = rs->cm_id->send_cq;
+ qp_attr.recv_cq = rs->cm_id->recv_cq;
+ qp_attr.qp_type = IBV_QPT_UD;
+ qp_attr.sq_sig_all = 1;
+ qp_attr.cap.max_send_wr = rs->sq_size;
+ qp_attr.cap.max_recv_wr = rs->rq_size;
+ qp_attr.cap.max_send_sge = 2;
+ qp_attr.cap.max_recv_sge = 1;
+ qp_attr.cap.max_inline_data = rs->sq_inline;
+
+ ret = rdma_create_qp((*qp)->cm_id, NULL, &qp_attr);
+ if (ret)
+ return ret;
+
+ for (i = 0; i < rs->rq_size; i++) {
+ ret = ds_post_recv(rs, *qp, (*qp)->rbuf + i * RS_SNDLOWAT);
+ if (ret)
+ goto err;
+ }
+ list_insert_head(&(*qp)->list, &rs->qp_list);
+ return 0;
+err:
+ ds_free_qp(*qp);
+ return ret;
}
-static struct ds_qp *ds_get_qp(struct rsocket *rs,
- const struct sockaddr *src_addr, socklen_t addrlen)
+static int ds_get_qp(struct rsocket *rs, union socket_addr *src_addr,
+ socklen_t addrlen, struct ds_qp **qp)
{
- union socket_addr *addr;
- socklen_t len;
+ dlist_t *entry;
+ for (entry = rs->qp_list.next; entry != &rs->qp_list; entry = entry->next) {
+ *qp = container_of(entry, struct ds_qp, list);
+ if (!ds_compare_addr(rdma_get_local_addr((*qp)->cm_id)), src_addr)
+ return 0;
+ }
+
+ return ds_create_qp(rs, src_addr, addrlen, qp);
}
-static int ds_connect(struct rsocket *rs,
- const struct sockaddr *dest_addr, socklen_t addrlen)
+static int ds_get_dest(struct rsocket *rs, const struct sockaddr *addr,
+ socklen_t addrlen, struct ds_dest **dest)
{
- struct ds_dest **dest;
+ union socket_addr src_addr;
+ socklen_t src_len;
+ struct ds_qp *qp;
+ int ret = 0;
fastlock_acquire(&rs->map_lock);
- dest = tfind(dest_addr, rs->dest_map, ds_compare_dest);
- if (dest) {
- rs->conn_dest = *dest;
+ dest = tfind(addr, &rs->dest_map, ds_compare_addr);
+ if (dest)
+ goto out;
+
+ if (rs->state == rs_init) {
+ ret = ds_init_ep(rs);
+ if (ret)
+ goto out;
+ }
+
+ ret = ds_get_src_addr(rs, addr, addrlen, &src_addr, &src_len);
+ if (ret)
+ goto out;
+
+ ret = ds_get_qp(rs, src_addr, src_len, &qp);
+ if (ret)
+ goto out;
+
+ *dest = calloc(1, sizeof(struct ds_dest));
+ if (!*dest) {
+ ret = ERR(ENOMEM);
goto out;
}
+
+ memcpy(&(*dest)->addr, addr, addrlen);
+ (*dest)->qp = qp;
+ tsearch((*dest)->addr, &rs->dest_map, ds_compare_addr);
out:
fastlock_release(&rs->map_lock);
- return 0;
+ return ret;
}
int rconnect(int socket, const struct sockaddr *addr, socklen_t addrlen)
{
struct rsocket *rs;
+ int ret;
rs = idm_at(&idm, socket);
- memcpy(&rs->cm_id->route.addr.dst_addr, addr, addrlen);
- return rs_do_connect(rs);
+ if (rs->type == SOCK_STREAM) {
+ memcpy(&rs->cm_id->route.addr.dst_addr, addr, addrlen);
+ ret = rs_do_connect(rs);
+ } else {
+ fastlock_acquire(&rs->slock);
+ ret = connect(rs->dsock, addr, addrlen);
+ if (!ret)
+ ret = ds_get_dest(rs, addr, addrlen, &rs->conn_dest);
+ fastlock_release(&rs->slock);
+ }
+ return ret;
}
static int rs_post_write_msg(struct rsocket *rs,
rs->state = rs_disconnected;
return 0;
} else if (rs_msg_data(imm_data) == RS_CTRL_SHUTDOWN) {
- rs->state &= ~rs_connect_rd;
+ rs->state &= ~rs_readable;
}
break;
case RS_OP_WRITE:
static int rs_conn_can_send(struct rsocket *rs)
{
- return rs_can_send(rs) || !(rs->state & rs_connect_wr);
+ return rs_can_send(rs) || !(rs->state & rs_writable);
}
static int rs_conn_can_send_ctrl(struct rsocket *rs)
static int rs_conn_have_rdata(struct rsocket *rs)
{
- return rs_have_rdata(rs) || !(rs->state & rs_connect_rd);
+ return rs_have_rdata(rs) || !(rs->state & rs_readable);
}
static int rs_conn_all_sends_done(struct rsocket *rs)
rs->rbuf_bytes_avail += rsize;
}
- } while (left && (flags & MSG_WAITALL) && (rs->state & rs_connect_rd));
+ } while (left && (flags & MSG_WAITALL) && (rs->state & rs_readable));
fastlock_release(&rs->rlock);
return ret ? ret : len - left;
rs_conn_can_send);
if (ret)
break;
- if (!(rs->state & rs_connect_wr)) {
+ if (!(rs->state & rs_writable)) {
ret = ERR(ECONNRESET);
break;
}
rs_conn_can_send);
if (ret)
break;
- if (!(rs->state & rs_connect_wr)) {
+ if (!(rs->state & rs_writable)) {
ret = ERR(ECONNRESET);
break;
}
rs_conn_can_send);
if (ret)
break;
- if (!(rs->state & rs_connect_wr)) {
+ if (!(rs->state & rs_writable)) {
ret = ERR(ECONNRESET);
break;
}
rs = idm_at(&idm, socket);
if (how == SHUT_RD) {
- rs->state &= ~rs_connect_rd;
+ rs->state &= ~rs_readable;
return 0;
}
if (rs->state & rs_connected) {
if (how == SHUT_RDWR) {
ctrl = RS_CTRL_DISCONNECT;
- rs->state &= ~(rs_connect_rd | rs_connect_wr);
+ rs->state &= ~(rs_readable | rs_writable);
} else {
- rs->state &= ~rs_connect_wr;
- ctrl = (rs->state & rs_connect_rd) ?
+ rs->state &= ~rs_writable;
+ ctrl = (rs->state & rs_readable) ?
RS_CTRL_SHUTDOWN : RS_CTRL_DISCONNECT;
}
if (!rs->ctrl_avail) {
rs_conn_can_send);
if (ret)
break;
- if (!(rs->state & rs_connect_wr)) {
+ if (!(rs->state & rs_writable)) {
ret = ERR(ECONNRESET);
break;
}
projects / ~shefty / librdmacm.git / commitdiff