From c25bf6431d76aca86ee4bf93d71a319fe3096a83 Mon Sep 17 00:00:00 2001 From: Sean Hefty Date: Sun, 21 Oct 2012 14:16:03 -0700 Subject: [PATCH] rsocket: Add direct data placement Signed-off-by: Sean Hefty --- include/rdma/rsocket.h | 6 +- src/indexer.h | 43 +++++++ src/rsocket.c | 252 +++++++++++++++++++++++++++++++++++++++-- 3 files changed, 288 insertions(+), 13 deletions(-) diff --git a/include/rdma/rsocket.h b/include/rdma/rsocket.h index 65feda96..21477e41 100644 --- a/include/rdma/rsocket.h +++ b/include/rdma/rsocket.h @@ -1,5 +1,5 @@ /* - * Copyright (c) 2011 Intel Corporation. All rights reserved. + * Copyright (c) 2011-2012 Intel Corporation. All rights reserved. * * This software is available to you under a choice of one of two * licenses. You may choose to be licensed under the terms of the GNU @@ -85,6 +85,10 @@ int rgetsockopt(int socket, int level, int optname, void *optval, socklen_t *optlen); int rfcntl(int socket, int cmd, ... /* arg */ ); +off_t riomap(int socket, void *buf, size_t len, int prot, int flags, off_t offset); +int riounmap(int socket, void *buf, size_t len); +size_t riowrite(int socket, const void *buf, size_t count, off_t offset, int flags); + #ifdef __cplusplus } #endif diff --git a/src/indexer.h b/src/indexer.h index 26e7f986..0c5f3882 100644 --- a/src/indexer.h +++ b/src/indexer.h @@ -31,6 +31,9 @@ * */ +#if !defined(INDEXER_H) +#define INDEXER_H + #if HAVE_CONFIG_H # include #endif /* HAVE_CONFIG_H */ @@ -99,3 +102,43 @@ static inline void *idm_lookup(struct index_map *idm, int index) return ((index <= IDX_MAX_INDEX) && idm->array[idx_array_index(index)]) ? idm_at(idm, index) : NULL; } + +typedef struct _dlist_entry { + struct _dlist_entry *next; + struct _dlist_entry *prev; +} dlist_entry; + +static inline void dlist_init(dlist_entry *head) +{ + head->next = head; + head->prev = head; +} + +static inline int dlist_empty(dlist_entry *head) +{ + return head->next == head; +} + +static inline void dlist_insert_after(dlist_entry *item, dlist_entry *head) +{ + item->next = head->next; + item->prev = head; + head->next->prev = item; + head->next = item; +} + +static inline void dlist_insert_before(dlist_entry *item, dlist_entry *head) +{ + dlist_insert_after(item, head->prev); +} + +#define dlist_insert_head dlist_insert_after +#define dlist_insert_tail dlist_insert_before + +static inline void dlist_remove(dlist_entry *item) +{ + item->prev->next = item->next; + item->next->prev = item->prev; +} + +#endif /* INDEXER_H */ diff --git a/src/rsocket.c b/src/rsocket.c index cc5effe1..ed708d43 100644 --- a/src/rsocket.c +++ b/src/rsocket.c @@ -62,6 +62,7 @@ static struct index_map idm; static pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER; +static uint16_t def_iomap_size = 0; static uint16_t def_inline = 64; static uint16_t def_sqsize = 384; static uint16_t def_rqsize = 384; @@ -76,7 +77,7 @@ static uint32_t polling_time = 10; * bit 29: more data, 0 - end of transfer, 1 - more data available * * for data transfers: - * bits [28:0]: bytes transfered, 0 = 1 GB + * bits [28:0]: bytes transfered * for control messages: * bits [28-0]: receive credits granted */ @@ -111,15 +112,30 @@ struct rs_sge { uint32_t length; }; -#define RS_MIN_INLINE (sizeof(struct rs_sge)) -#define rs_host_is_net() (1 == htonl(1)) -#define RS_CONN_FLAG_NET 1 +struct rs_iomap { + uint64_t offset; + struct rs_sge sge; +}; + +struct rs_iomap_mr { + uint64_t offset; + struct ibv_mr *mr; + dlist_entry entry; + atomic_t refcnt; + int index; /* -1 if mapping is local and not in iomap_list */ +}; + +#define RS_MIN_INLINE (sizeof(struct rs_sge)) +#define rs_host_is_net() (1 == htonl(1)) +#define RS_CONN_FLAG_NET (1 << 0) +#define RS_CONN_FLAG_IOMAP (1 << 1) struct rs_conn_data { uint8_t version; uint8_t flags; uint16_t credits; - uint32_t reserved2; + uint8_t reserved[3]; + uint8_t target_iomap_size; struct rs_sge target_sgl; struct rs_sge data_buf; }; @@ -155,6 +171,7 @@ struct rsocket { fastlock_t rlock; fastlock_t cq_lock; fastlock_t cq_wait_lock; + fastlock_t iomap_lock; int opts; long fd_flags; @@ -186,10 +203,19 @@ struct rsocket { int remote_sge; struct rs_sge remote_sgl; + struct rs_sge remote_iomap; + + struct rs_iomap_mr *remote_iomappings; + dlist_entry iomap_list; + dlist_entry iomap_queue; + int iomap_pending; struct ibv_mr *target_mr; int target_sge; - volatile struct rs_sge target_sgl[RS_SGL_SIZE]; + int target_iomap_size; + void *target_buffer_list; + volatile struct rs_sge *target_sgl; + volatile struct rs_iomap *target_iomap; uint32_t rbuf_size; struct ibv_mr *rmr; @@ -201,6 +227,18 @@ struct rsocket { uint8_t *sbuf; }; +static int rs_value_to_scale(int value, int bits) +{ + return value <= (1 << (bits - 1)) ? + value : (1 << (bits - 1)) | (value >> bits); +} + +static int rs_scale_to_value(int value, int bits) +{ + return value <= (1 << (bits - 1)) ? + value : (value & ~(1 << (bits - 1))) << bits; +} + void rs_configure(void) { FILE *f; @@ -251,6 +289,15 @@ void rs_configure(void) if (def_wmem < 1) def_wmem = 1; } + + if ((f = fopen(RS_CONN_DIR "/iomap_size", "r"))) { + fscanf(f, "%hu", &def_iomap_size); + fclose(f); + + /* round to supported values */ + def_iomap_size = (uint8_t) rs_value_to_scale(def_iomap_size, 8); + def_iomap_size = (uint16_t) rs_scale_to_value(def_iomap_size, 8); + } init = 1; out: pthread_mutex_unlock(&mut); @@ -287,6 +334,7 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs) rs->sq_size = inherited_rs->sq_size; rs->rq_size = inherited_rs->rq_size; rs->ctrl_avail = inherited_rs->ctrl_avail; + rs->target_iomap_size = inherited_rs->target_iomap_size; } else { rs->sbuf_size = def_wmem; rs->rbuf_size = def_mem; @@ -294,11 +342,15 @@ static struct rsocket *rs_alloc(struct rsocket *inherited_rs) rs->sq_size = def_sqsize; rs->rq_size = def_rqsize; rs->ctrl_avail = RS_QP_CTRL_SIZE; + rs->target_iomap_size = def_iomap_size; } fastlock_init(&rs->slock); fastlock_init(&rs->rlock); fastlock_init(&rs->cq_lock); fastlock_init(&rs->cq_wait_lock); + fastlock_init(&rs->iomap_lock); + dlist_init(&rs->iomap_list); + dlist_init(&rs->iomap_queue); return rs; } @@ -336,6 +388,8 @@ static void rs_set_qp_size(struct rsocket *rs) static int rs_init_bufs(struct rsocket *rs) { + size_t len; + rs->rmsg = calloc(rs->rq_size + 1, sizeof(*rs->rmsg)); if (!rs->rmsg) return -1; @@ -348,11 +402,21 @@ static int rs_init_bufs(struct rsocket *rs) if (!rs->smr) return -1; - rs->target_mr = rdma_reg_write(rs->cm_id, (void *) rs->target_sgl, - sizeof(rs->target_sgl)); + len = sizeof(*rs->target_sgl) * RS_SGL_SIZE + + sizeof(*rs->target_iomap) * rs->target_iomap_size; + rs->target_buffer_list = malloc(len); + if (!rs->target_buffer_list) + return -1; + + rs->target_mr = rdma_reg_write(rs->cm_id, rs->target_buffer_list, len); if (!rs->target_mr) return -1; + memset(rs->target_buffer_list, 0, len); + rs->target_sgl = rs->target_buffer_list; + if (rs->target_iomap_size) + rs->target_iomap = (struct rs_iomap *) (rs->target_sgl + RS_SGL_SIZE); + rs->rbuf = calloc(rs->rbuf_size, sizeof(*rs->rbuf)); if (!rs->rbuf) return -1; @@ -452,6 +516,42 @@ static int rs_create_ep(struct rsocket *rs) return 0; } +/* +static xxx rs_acquire_iomap_mr(struct rsocket *rs, ...) +{ + TODO: write me +} +*/ + +static void rs_release_iomap_mr(struct rs_iomap_mr *iomr) +{ + if (atomic_dec(&iomr->refcnt)) + return; + + dlist_remove(&iomr->entry); + ibv_dereg_mr(iomr->mr); + if (iomr->index >= 0) + iomr->mr = NULL; + else + free(iomr); +} + +static void rs_free_iomappings(struct rsocket *rs) +{ + struct rs_iomap_mr *iomr; + + while (!dlist_empty(&rs->iomap_list)) { + iomr = container_of(rs->iomap_list.next, + struct rs_iomap_mr, entry); + riounmap(iomr->mr->addr, iomr->mr->length); + } + while (!dlist_empty(&rs->iomap_queue)) { + iomr = container_of(rs->iomap_queue.next, + struct rs_iomap_mr, entry); + riounmap(iomr->mr->addr, iomr->mr->length); + } +} + static void rs_free(struct rsocket *rs) { if (rs->index >= 0) @@ -472,15 +572,20 @@ static void rs_free(struct rsocket *rs) free(rs->rbuf); } - if (rs->target_mr) - rdma_dereg_mr(rs->target_mr); + if (rs->target_buffer_list) { + if (rs->target_mr) + rdma_dereg_mr(rs->target_mr); + free(rs->target_buffer_list); + } if (rs->cm_id) { + rs_free_iomappings(rs); if (rs->cm_id->qp) rdma_destroy_qp(rs->cm_id); rdma_destroy_id(rs->cm_id); } + fastlock_destroy(&rs->iomap_lock); fastlock_destroy(&rs->cq_wait_lock); fastlock_destroy(&rs->cq_lock); fastlock_destroy(&rs->rlock); @@ -492,9 +597,11 @@ static void rs_set_conn_data(struct rsocket *rs, struct rdma_conn_param *param, struct rs_conn_data *conn) { conn->version = 1; - conn->flags = rs_host_is_net() ? RS_CONN_FLAG_NET : 0; + conn->flags = RS_CONN_FLAG_IOMAP | + (rs_host_is_net() ? RS_CONN_FLAG_NET : 0); conn->credits = htons(rs->rq_size); - conn->reserved2 = 0; + memset(conn->reserved, 0, sizeof conn->reserved); + conn->target_iomap_size = (uint8_t) rs_value_to_scale(rs->target_iomap_size, 8); conn->target_sgl.addr = htonll((uintptr_t) rs->target_sgl); conn->target_sgl.length = htonl(RS_SGL_SIZE); @@ -518,6 +625,13 @@ static void rs_save_conn_data(struct rsocket *rs, struct rs_conn_data *conn) (!rs_host_is_net() && (conn->flags & RS_CONN_FLAG_NET))) rs->opts = RS_OPT_SWAP_SGL; + if (conn->flags & RS_CONN_FLAG_IOMAP) { + rs->remote_iomap.addr = rs->remote_sgl.addr + + sizeof(rs->remote_sgl) * rs->remote_sgl.length; + rs->remote_iomap.length = rs_scale_to_value(conn->target_iomap_size, 8); + rs->remote_iomap.key = rs->remote_sgl.key; + } + rs->target_sgl[0].addr = ntohll(conn->data_buf.addr); rs->target_sgl[0].length = ntohl(conn->data_buf.length); rs->target_sgl[0].key = ntohl(conn->data_buf.key); @@ -2020,3 +2134,117 @@ int rfcntl(int socket, int cmd, ... /* arg */ ) va_end(args); return ret; } + +static struct rs_iomap_mr *rs_get_iomap_mr(struct rsocket *rs) +{ + struct rs_iomap_mr *iomr; + int i; + + if (!rs->remote_iomappings) { + rs->remote_iomappings = calloc(rs->remote_iomap.length, + sizeof(*rs->remote_iomappings)); + if (!rs->remote_iomappings) + return NULL; + + for (i = 0; i < rs->remote_iomap.length; i++) + rs->remote_iomappings[i].index = i; + } + + for (i = 0; i < rs->remote_iomap.length; i++) { + if (!rs->remote_iomappings[i]->mr) + return &rs->remote_iomappings[i]; + } + return NULL; +} + +/* + * If an offset is given, we map to it. If offset is -1, then we map the + * offset to the address of buf. We do not check for conflicts, which must + * be fixed at some point. + */ +off_t riomap(int socket, void *buf, size_t len, int prot, int flags, off_t offset) +{ + struct rsocket *rs; + struct rs_iomap_mr *iomr; + int ret, access = IBV_ACCESS_LOCAL_WRITE; + + rs = idm_at(&idm, socket); + if ((rs->state != rs_connect_rdwr) || (prot & ~(PROT_WRITE | PROT_NONE))) + return ERR(EINVAL); + + fastlock_acquire(&rs->iomap_lock); + if (prot & PROT_WRITE) { + iomr = rs_get_iomap_mr(rs); + access |= IBV_ACCESS_REMOTE_WRITE; + } else { + iomr = calloc(1, sizeof *iomr); + iomr->index = -1; + } + if (!iomr) { + offset = ERR(ENOMEM); + goto out; + } + + iomr->mr = ibv_reg_mr(rs->cm_id->pd, buf, len, access); + if (!iomr->mr) { + if (iomr->index < 0) + free(iomr); + offset = -1; + goto out; + } + + if (offset == -1) + offset = (uintptr_t) buf; + iomr->offset = offset; + iomr->prot = prot; + atomic_init(&iomr->refcnt); + atomic_set(&iomr->refcnt, 1); + + if (iomr->index >= 0) { + dlist_insert_tail(&iomr->entry, &rs->iomap_queue); + rs->iomap_pending = 1; + } else { + dlist_insert_tail(&iomr->entry, &rs->iomap_list); + } +out: + fastlock_release(&rs->iomap_lock); + return offset; +} + +int riounmap(int socket, void *buf, size_t len) +{ + struct rsocket *rs; + struct rs_iomap_mr *iomr; + struct dlist_entry *entry; + int ret = 0; + + rs = idm_at(&idm, socket); + fastlock_acquire(&rs->iomap_lock); + + for (entry = rs->iomap_list.next; entry != &rs->iomap_list; + entry = entry->next) { + iomr = container_of(entry, struct rs_iomap_mr, entry); + if (iomr->mr->addr == buf && iomr->mr->length == len) { + rs_release_iomap_mr(iomr); + goto out; + } + } + + for (entry = rs->iomap_queue.next; entry != &rs->iomap_queue; + entry = entry->next) { + iomr = container_of(entry, struct rs_iomap_mr, entry); + if (iomr->mr->addr == buf && iomr->mr->length == len) { + rs_release_iomap_mr(iomr); + goto out; + } + } + ret = ERR(EINVAL); +out: + fastlock_release(&rs->iomap_lock); + return ret; +} + +size_t riowrite(int socket, const void *buf, size_t count, off_t offset, int flags) +{ + +} -- 2.41.0