summaryrefslogtreecommitdiffstats
path: root/extras/libmemif/src
diff options
context:
space:
mode:
authorJakub Grajciar <jgrajcia@cisco.com>2018-03-29 13:15:10 +0200
committerDamjan Marion <dmarion.lists@gmail.com>2018-03-30 20:48:36 +0000
commit3744fc7abce0cf8694d64b670589e35c6d7bf881 (patch)
tree04f5f42458a1ab4f3f4109f890e998221b22e622 /extras/libmemif/src
parent17ddc0fee1bc20d0da84dd70bb579c8844a48ef4 (diff)
libmemif: zero-copy-slave mode + header space
Slave is now able to dequeue buffers from rx queue and enqueue them to tx queue (zero-copy operation). Slave can produce buffers with headroom, which will allow adding encap without copy. Change-Id: Ia189f8de1a68be787545ed46cf78d36403e7e9bf Signed-off-by: Jakub Grajciar <jgrajcia@cisco.com>
Diffstat (limited to 'extras/libmemif/src')
-rw-r--r--extras/libmemif/src/libmemif.h26
-rw-r--r--extras/libmemif/src/main.c201
-rw-r--r--extras/libmemif/src/memif_private.h1
-rw-r--r--extras/libmemif/src/socket.c6
4 files changed, 189 insertions, 45 deletions
diff --git a/extras/libmemif/src/libmemif.h b/extras/libmemif/src/libmemif.h
index e1f96f90c80..88b6a10f7b4 100644
--- a/extras/libmemif/src/libmemif.h
+++ b/extras/libmemif/src/libmemif.h
@@ -203,15 +203,20 @@ typedef enum
/** \brief Memif buffer
@param desc_index - ring descriptor index
- @param len - buffer length
+ @param ring - pointer to ring containing descriptor for this buffer
+ @param len - available length
@param flags - memif buffer flags
@param data - pointer to shared memory data
*/
typedef struct
{
uint16_t desc_index;
+ void *ring;
uint32_t len;
+/** next buffer present (chained buffers) */
#define MEMIF_BUFFER_FLAG_NEXT (1 << 0)
+/** states that buffer is from rx ring */
+#define MEMIF_BUFFER_FLAG_RX (1 << 1)
uint8_t flags;
void *data;
} memif_buffer_t;
@@ -423,6 +428,22 @@ int memif_control_fd_handler (int fd, uint8_t events);
*/
int memif_delete (memif_conn_handle_t * conn);
+/** \brief Memif buffer enq tx
+ @param conn - memif conenction handle
+ @param qid - number indentifying queue
+ @param bufs - memif buffers
+ @param count - number of memif buffers to enque
+ @param count_out - returns number of allocated buffers
+
+ Slave is producer of buffers.
+ If connection handle points to master returns MEMIF_ERR_INVAL_ARG.
+
+ \return memif_err_t
+*/
+int memif_buffer_enq_tx (memif_conn_handle_t conn, uint16_t qid,
+ memif_buffer_t * bufs, uint16_t count,
+ uint16_t * count_out);
+
/** \brief Memif buffer alloc
@param conn - memif conenction handle
@param qid - number indentifying queue
@@ -441,11 +462,12 @@ int memif_buffer_alloc (memif_conn_handle_t conn, uint16_t qid,
@param conn - memif conenction handle
@param qid - number indentifying queue
@param count - number of buffers to be placed on ring
+ @param headroom - offset the buffer by headroom
\return memif_err_t
*/
int memif_refill_queue (memif_conn_handle_t conn, uint16_t qid,
- uint16_t count);
+ uint16_t count, uint16_t headroom);
/** \brief Memif transmit buffer burst
@param conn - memif conenction handle
diff --git a/extras/libmemif/src/main.c b/extras/libmemif/src/main.c
index dc4a6231118..56458515485 100644
--- a/extras/libmemif/src/main.c
+++ b/extras/libmemif/src/main.c
@@ -144,7 +144,7 @@ const char *memif_errlist[ERRLIST_LEN] = { /* MEMIF_ERR_SUCCESS */
/* MEMIF_ERR_POLL_CANCEL */
"Memif event polling was canceled.",
/* MEMIF_ERR_MAX_RING */
- "Maximum log2 ring size is 15.",
+ "Maximum log2 ring size is 15",
/* MEMIF_ERR_PRIVHDR */
"Private headers not supported."
};
@@ -1324,7 +1324,6 @@ int
memif_init_regions_and_queues (memif_connection_t * conn)
{
memif_ring_t *ring = NULL;
- uint64_t buffer_offset;
memif_region_t *r;
int i, j;
libmemif_main_t *lm = &libmemif_main;
@@ -1335,18 +1334,19 @@ memif_init_regions_and_queues (memif_connection_t * conn)
return MEMIF_ERR_NOMEM;
r = conn->regions;
- buffer_offset =
+ r->buffer_offset =
(conn->run_args.num_s2m_rings +
conn->run_args.num_m2s_rings) * (sizeof (memif_ring_t) +
sizeof (memif_desc_t) *
(1 << conn->run_args.log2_ring_size));
- r->region_size = buffer_offset +
+ r->region_size = r->buffer_offset +
conn->run_args.buffer_size * (1 << conn->run_args.log2_ring_size) *
(conn->run_args.num_s2m_rings + conn->run_args.num_m2s_rings);
if ((r->fd =
memfd_create ("memif region 0", MFD_ALLOW_SEALING)) == -1)
+ if ((r->fd = memfd_create ("memif region 0", MFD_ALLOW_SEALING)) == -1)
return memif_syscall_error_handler (errno);
if ((fcntl (r->fd, F_ADD_SEALS, F_SEAL_SHRINK)) == -1)
@@ -1370,7 +1370,7 @@ memif_init_regions_and_queues (memif_connection_t * conn)
{
uint16_t slot = i * (1 << conn->run_args.log2_ring_size) + j;
ring->desc[j].region = 0;
- ring->desc[j].offset = buffer_offset +
+ ring->desc[j].offset = r->buffer_offset +
(uint32_t) (slot * conn->run_args.buffer_size);
ring->desc[j].length = conn->run_args.buffer_size;
}
@@ -1389,7 +1389,7 @@ memif_init_regions_and_queues (memif_connection_t * conn)
conn->run_args.num_s2m_rings) *
(1 << conn->run_args.log2_ring_size) + j;
ring->desc[j].region = 0;
- ring->desc[j].offset = buffer_offset +
+ ring->desc[j].offset = r->buffer_offset +
(uint32_t) (slot * conn->run_args.buffer_size);
ring->desc[j].length = conn->run_args.buffer_size;
}
@@ -1417,7 +1417,6 @@ memif_init_regions_and_queues (memif_connection_t * conn)
mq[x].offset =
(void *) mq[x].ring - (void *) conn->regions[mq->region].shm;
mq[x].last_head = 0;
- mq[x].last_tail = 0;
mq[x].alloc_bufs = 0;
}
conn->tx_queues = mq;
@@ -1443,7 +1442,6 @@ memif_init_regions_and_queues (memif_connection_t * conn)
mq[x].offset =
(void *) mq[x].ring - (void *) conn->regions[mq->region].shm;
mq[x].last_head = 0;
- mq[x].last_tail = 0;
mq[x].alloc_bufs = 0;
}
conn->rx_queues = mq;
@@ -1452,6 +1450,87 @@ memif_init_regions_and_queues (memif_connection_t * conn)
}
int
+memif_buffer_enq_tx (memif_conn_handle_t conn, uint16_t qid,
+ memif_buffer_t * bufs, uint16_t count,
+ uint16_t * count_out)
+{
+ memif_connection_t *c = (memif_connection_t *) conn;
+ if (EXPECT_FALSE (c == NULL))
+ return MEMIF_ERR_NOCONN;
+ if (EXPECT_FALSE (c->fd < 0))
+ return MEMIF_ERR_DISCONNECTED;
+ uint8_t num =
+ (c->args.is_master) ? c->run_args.num_m2s_rings : c->run_args.
+ num_s2m_rings;
+ if (EXPECT_FALSE (qid >= num))
+ return MEMIF_ERR_QID;
+ if (EXPECT_FALSE (!count_out))
+ return MEMIF_ERR_INVAL_ARG;
+ if (EXPECT_FALSE (c->args.is_master))
+ return MEMIF_ERR_INVAL_ARG;
+
+ memif_queue_t *mq = &c->tx_queues[qid];
+ memif_ring_t *ring = mq->ring;
+ memif_buffer_t *b0;
+ uint16_t mask = (1 << mq->log2_ring_size) - 1;
+ uint16_t ring_size;
+ uint16_t slot, ns;
+ int i, err = MEMIF_ERR_SUCCESS; /* 0 */
+ *count_out = 0;
+
+ ring_size = (1 << mq->log2_ring_size);
+ ns = ring->tail - mq->last_tail;
+ mq->last_tail += ns;
+ slot = (c->args.is_master) ? ring->tail : ring->head;
+ slot += mq->alloc_bufs;
+
+ /* can only be called by slave */
+ ns = ring_size - ring->head + mq->alloc_bufs + mq->last_tail;
+
+ b0 = bufs;
+
+ while (count && ns)
+ {
+ if (EXPECT_FALSE ((b0->flags & MEMIF_BUFFER_FLAG_RX) == 0))
+ {
+ /* not a valid buffer */
+ count--;
+ continue;
+ }
+ b0->flags &= ~MEMIF_BUFFER_FLAG_RX;
+
+ ((memif_ring_t *) b0->ring)->desc[b0->desc_index & mask].offset = ring->desc[slot & mask].offset; /* put free buffer on rx ring */
+
+ ring->desc[slot & mask].offset =
+ (uint32_t) (b0->data - c->regions->shm);
+ ring->desc[slot & mask].flags |=
+ (b0->flags & MEMIF_BUFFER_FLAG_NEXT) ? MEMIF_DESC_FLAG_NEXT : 0;
+
+ b0->desc_index = slot;
+
+ mq->alloc_bufs++;
+ slot++;
+
+ count--;
+ ns--;
+ b0++;
+ *count_out += 1;
+ }
+
+ DBG ("allocated: %u/%u bufs. Total %u allocated bufs", *count_out, count,
+ mq->alloc_bufs);
+
+ if (count)
+ {
+ DBG ("ring buffer full! qid: %u", qid);
+ err = MEMIF_ERR_NOBUF_RING;
+ }
+
+error:
+ return err;
+}
+
+int
memif_buffer_alloc (memif_conn_handle_t conn, uint16_t qid,
memif_buffer_t * bufs, uint16_t count,
uint16_t * count_out, uint16_t size)
@@ -1503,12 +1582,13 @@ memif_buffer_alloc (memif_conn_handle_t conn, uint16_t qid,
ring->desc[slot & mask].flags = 0;
/* slave can produce buffer with original length */
- dst_left = (c->args.is_master) ? ring->desc[slot & mask].length : c->run_args.buffer_size; /* - headroom */
+ dst_left = (c->args.is_master) ? ring->desc[slot & mask].length :
+ c->run_args.buffer_size;
src_left = size;
while (src_left)
{
- if (dst_left == 0)
+ if (EXPECT_FALSE (dst_left == 0))
{
if (count && ns)
{
@@ -1524,7 +1604,9 @@ memif_buffer_alloc (memif_conn_handle_t conn, uint16_t qid,
b0 = (bufs + *count_out);
b0->desc_index = slot;
- dst_left = (c->args.is_master) ? ring->desc[slot & mask].length : c->run_args.buffer_size; /* - headroom */
+ dst_left =
+ (c->args.is_master) ? ring->desc[slot & mask].length : c->
+ run_args.buffer_size;
ring->desc[slot & mask].flags = 0;
}
else
@@ -1538,7 +1620,18 @@ memif_buffer_alloc (memif_conn_handle_t conn, uint16_t qid,
}
}
b0->len = memif_min (dst_left, src_left);
- b0->data = memif_get_buffer (c, ring, slot & mask);
+
+ /* slave resets buffer offset */
+ if (c->args.is_master == 0)
+ {
+ uint16_t x =
+ (ring->desc[slot & mask].offset -
+ c->regions->buffer_offset) / c->run_args.buffer_size;
+ ring->desc[slot & mask].offset =
+ c->regions->buffer_offset + (x * c->run_args.buffer_size);
+ }
+
+ b0->data = c->regions->shm + ring->desc[slot & mask].offset;
src_left -= b0->len;
dst_left -= b0->len;
@@ -1562,11 +1655,13 @@ no_ns:
err = MEMIF_ERR_NOBUF_RING;
}
+error:
return err;
}
int
-memif_refill_queue (memif_conn_handle_t conn, uint16_t qid, uint16_t count)
+memif_refill_queue (memif_conn_handle_t conn, uint16_t qid, uint16_t count,
+ uint16_t headroom)
{
memif_connection_t *c = (memif_connection_t *) conn;
if (EXPECT_FALSE (c == NULL))
@@ -1581,6 +1676,8 @@ memif_refill_queue (memif_conn_handle_t conn, uint16_t qid, uint16_t count)
libmemif_main_t *lm = &libmemif_main;
memif_queue_t *mq = &c->rx_queues[qid];
memif_ring_t *ring = mq->ring;
+ uint16_t mask = (1 << mq->log2_ring_size) - 1;
+ uint16_t slot;
if (c->args.is_master)
{
@@ -1588,16 +1685,32 @@ memif_refill_queue (memif_conn_handle_t conn, uint16_t qid, uint16_t count)
ring->tail =
(ring->tail + count <=
mq->last_head) ? ring->tail + count : mq->last_head;
+ return MEMIF_ERR_SUCCESS;
}
- else
+
+ uint16_t head = ring->head;
+ uint16_t ns = (1 << mq->log2_ring_size) - head + mq->last_tail;
+ head += (count < ns) ? count : ns;
+
+ if (headroom)
{
- uint16_t head = ring->head;
- uint16_t ns = (1 << mq->log2_ring_size) - head + mq->last_tail;
- head += ns;
- MEMIF_MEMORY_BARRIER ();
- ring->head = (ring->head + count <= head) ? ring->head + count : head;
+ slot = (c->args.is_master) ? ring->head : ring->tail;
+ while (slot < head)
+ {
+ uint16_t x =
+ (ring->desc[slot & mask].offset -
+ c->regions->buffer_offset) / c->run_args.buffer_size;
+ ring->desc[slot & mask].offset =
+ c->regions->buffer_offset + (x * c->run_args.buffer_size) +
+ headroom;
+
+ slot++;
+ }
}
+ MEMIF_MEMORY_BARRIER ();
+ ring->head = head;
+
return MEMIF_ERR_SUCCESS; /* 0 */
}
@@ -1620,32 +1733,40 @@ memif_tx_burst (memif_conn_handle_t conn, uint16_t qid,
memif_queue_t *mq = &c->tx_queues[qid];
memif_ring_t *ring = mq->ring;
- uint16_t slot;
-
- slot = (c->args.is_master) ? ring->tail : ring->head;
- *tx = (count <= mq->alloc_bufs) ? count : mq->alloc_bufs;
-
-#ifdef MEMIF_DBG_SHM
- uint16_t curr_buf = 0;
uint16_t mask = (1 << mq->log2_ring_size) - 1;
memif_buffer_t *b0;
- for (curr_buf = 0; curr_buf < count; curr_buf++)
+ *tx = 0;
+
+ if (count > mq->alloc_bufs)
+ count = mq->alloc_bufs;
+
+ if (EXPECT_FALSE (count == 0))
+ return MEMIF_ERR_SUCCESS;
+
+ while (count)
{
- b0 = (bufs + curr_buf);
+ b0 = (bufs + *tx);
+ ring->desc[b0->desc_index & mask].length = b0->len;
+
+#ifdef MEMIF_DBG_SHM
+ printf ("offset: %-6d\n", ring->desc[b0->desc_index & mask].offset);
+ printf ("data: %p\n",
+ memif_get_buffer (c, ring, b0->desc_index & mask));
+ printf ("index: %u\n", b0->desc_index);
print_bytes (memif_get_buffer (c, ring, b0->desc_index & mask),
ring->desc[b0->desc_index & mask].length, DBG_TX_BUF);
+#endif /* MEMIF_DBG_SHM */
+ *tx += 1;
+ count--;
}
-#endif /* MEMIF_DBG_SHM */
+
MEMIF_MEMORY_BARRIER ();
if (c->args.is_master)
- ring->tail = slot + *tx;
+ ring->tail = b0->desc_index + 1;
else
- ring->head = slot + *tx;
-
- /* zero out buffer fields so the client cant modify transmitted data */
- memset (bufs, 0, sizeof (memif_buffer_t) * *tx);
+ ring->head = b0->desc_index + 1;
mq->alloc_bufs -= *tx;
@@ -1687,7 +1808,7 @@ memif_rx_burst (memif_conn_handle_t conn, uint16_t qid,
uint64_t b;
ssize_t r = read (mq->int_fd, &b, sizeof (b));
- if ((r == -1) && (errno != EAGAIN))
+ if (EXPECT_FALSE ((r == -1) && (errno != EAGAIN)))
return memif_syscall_error_handler (errno);
cur_slot = (c->args.is_master) ? mq->last_head : mq->last_tail;
@@ -1709,13 +1830,19 @@ memif_rx_burst (memif_conn_handle_t conn, uint16_t qid,
{
ring->desc[cur_slot & mask].length = c->run_args.buffer_size;
}
+
+ b0->flags = MEMIF_BUFFER_FLAG_RX;
if (ring->desc[cur_slot & mask].flags & MEMIF_DESC_FLAG_NEXT)
{
- b0->flags = MEMIF_BUFFER_FLAG_NEXT;
+ b0->flags |= MEMIF_BUFFER_FLAG_NEXT;
ring->desc[cur_slot & mask].flags &= ~MEMIF_DESC_FLAG_NEXT;
}
-
+/* b0->offset = ring->desc[cur_slot & mask].offset;*/
+ b0->ring = ring;
#ifdef MEMIF_DBG_SHM
+ printf ("data: %p\n", b0->data);
+ printf ("index: %u\n", b0->desc_index);
+ printf ("ring: %p\n", b0->ring);
print_bytes (b0->data, b0->len, DBG_RX_BUF);
#endif /* MEMIF_DBG_SHM */
ns--;
diff --git a/extras/libmemif/src/memif_private.h b/extras/libmemif/src/memif_private.h
index accdac361b9..5ea5494decb 100644
--- a/extras/libmemif/src/memif_private.h
+++ b/extras/libmemif/src/memif_private.h
@@ -69,6 +69,7 @@ typedef struct
{
void *shm;
uint32_t region_size;
+ uint32_t buffer_offset;
int fd;
} memif_region_t;
diff --git a/extras/libmemif/src/socket.c b/extras/libmemif/src/socket.c
index a471e0ba943..e568079ab25 100644
--- a/extras/libmemif/src/socket.c
+++ b/extras/libmemif/src/socket.c
@@ -590,9 +590,6 @@ memif_msg_receive_connect (memif_connection_t * c, memif_msg_t * msg)
add_list_elt (&elt, &lm->interrupt_list, &lm->interrupt_list_len);
lm->control_fd_update (c->rx_queues[i].int_fd, MEMIF_FD_EVENT_READ);
-
- /* refill ring buffers */
- memif_refill_queue ((void *) c, i, -1);
}
}
@@ -623,9 +620,6 @@ memif_msg_receive_connected (memif_connection_t * c, memif_msg_t * msg)
for (i = 0; i < c->run_args.num_s2m_rings; i++)
{
lm->control_fd_update (c->rx_queues[i].int_fd, MEMIF_FD_EVENT_READ);
-
- /* refill ring buffers */
- memif_refill_queue ((void *) c, i, -1);
}
}