[dpdk-dev] [PATCH 8/8] raw/ioat: add local API to perform copies

[Bruce Richardson <bruce.richardson@intel.com>]

Add local APIs to trigger data copies, and retrieve handle values once
those copies are completed. Included are unit tests to validate the data
is copies correctly.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
---
 app/test/test_ioat_rawdev.c        | 159 ++++++++++++++++++++++++++++-
 doc/guides/rawdevs/ioat_rawdev.rst | 100 ++++++++++++++++++
 drivers/raw/ioat/rte_ioat_rawdev.h | 155 +++++++++++++++++++++++++++-
 3 files changed, 410 insertions(+), 4 deletions(-)

diff --git a/app/test/test_ioat_rawdev.c b/app/test/test_ioat_rawdev.c
index 7081f3365..f2240adec 100644
--- a/app/test/test_ioat_rawdev.c
+++ b/app/test/test_ioat_rawdev.c
@@ -18,6 +18,131 @@ test_ioat_rawdev(void) { return TEST_SKIPPED; }
 #include <rte_rawdev.h>
 #include <rte_ioat_rawdev.h>
 
+static struct rte_mempool *pool;
+
+static int
+test_enqueue_copies(int dev_id)
+{
+	const unsigned int length = 1024;
+	unsigned int i;
+
+	do {
+		struct rte_mbuf *src, *dst;
+		char *src_data, *dst_data;
+		struct rte_mbuf *completed[2] = {0};
+
+		/* test doing a single copy */
+		src = rte_pktmbuf_alloc(pool);
+		dst = rte_pktmbuf_alloc(pool);
+		src->data_len = src->pkt_len = length;
+		dst->data_len = dst->pkt_len = length;
+		src_data = rte_pktmbuf_mtod(src, char *);
+		dst_data = rte_pktmbuf_mtod(dst, char *);
+
+		for (i = 0; i < length; i++)
+			src_data[i] = rand() & 0xFF;
+
+		if (rte_ioat_enqueue_copy(dev_id,
+				src->buf_iova + src->data_off,
+				dst->buf_iova + dst->data_off,
+				length,
+				(uintptr_t)src,
+				(uintptr_t)dst,
+				0 /* no fence */) != 1) {
+			printf("Error with rte_ioat_enqueue_copy\n");
+			return -1;
+		}
+		rte_ioat_do_copies(dev_id);
+		usleep(10);
+
+		if (rte_ioat_completed_copies(dev_id, 1, (void *)&completed[0],
+				(void *)&completed[1]) != 1) {
+			printf("Error with rte_ioat_completed_copies\n");
+			return -1;
+		}
+		if (completed[0] != src || completed[1] != dst) {
+			printf("Error with completions: got (%p, %p), not (%p,%p)\n",
+					completed[0], completed[1], src, dst);
+			return -1;
+		}
+
+		for (i = 0; i < length; i++)
+			if (dst_data[i] != src_data[i]) {
+				printf("Data mismatch at char %u\n", i);
+				return -1;
+			}
+		rte_pktmbuf_free(src);
+		rte_pktmbuf_free(dst);
+	} while(0);
+
+	/* test doing multiple copies */
+	do {
+		struct rte_mbuf *srcs[32], *dsts[32];
+		struct rte_mbuf *completed_src[64];
+		struct rte_mbuf *completed_dst[64];
+		unsigned int j;
+
+		for (i = 0; i < RTE_DIM(srcs); i++) {
+			char *src_data;
+
+			srcs[i] = rte_pktmbuf_alloc(pool);
+			dsts[i] = rte_pktmbuf_alloc(pool);
+			srcs[i]->data_len = srcs[i]->pkt_len = length;
+			dsts[i]->data_len = dsts[i]->pkt_len = length;
+			src_data = rte_pktmbuf_mtod(srcs[i], char *);
+
+			for (j = 0; j < length; j++)
+				src_data[j] = rand() & 0xFF;
+
+			if (rte_ioat_enqueue_copy(dev_id,
+					srcs[i]->buf_iova + srcs[i]->data_off,
+					dsts[i]->buf_iova + dsts[i]->data_off,
+					length,
+					(uintptr_t)srcs[i],
+					(uintptr_t)dsts[i],
+					0 /* nofence */) != 1) {
+				printf("Error with rte_ioat_enqueue_copy for buffer %u\n",
+						i);
+				return -1;
+			}
+		}
+		rte_ioat_do_copies(dev_id);
+		usleep(100);
+
+		if (rte_ioat_completed_copies(dev_id, 64, (void *)completed_src,
+				(void *)completed_dst) != RTE_DIM(srcs)) {
+			printf("Error with rte_ioat_completed_copies\n");
+			return -1;
+		}
+		for (i = 0; i < RTE_DIM(srcs); i++) {
+			char *src_data, *dst_data;
+
+			if (completed_src[i] != srcs[i]) {
+				printf("Error with source pointer %u\n", i);
+				return -1;
+			}
+			if (completed_dst[i] != dsts[i]) {
+				printf("Error with dest pointer %u\n", i);
+				return -1;
+			}
+
+			src_data = rte_pktmbuf_mtod(srcs[i], char *);
+			dst_data = rte_pktmbuf_mtod(dsts[i], char *);
+			for (j = 0; j < length; j++)
+				if (src_data[j] != dst_data[j]) {
+					printf("Error with copy of packet %u, byte %u\n",
+							i, j);
+					return -1;
+				}
+			rte_pktmbuf_free(srcs[i]);
+			rte_pktmbuf_free(dsts[i]);
+		}
+
+	} while(0);
+
+	return 0;
+}
+
 static int
 run_ioat_tests(int dev_id)
 {
@@ -54,6 +179,17 @@ run_ioat_tests(int dev_id)
 		return -1;
 	}
 
+	pool = rte_pktmbuf_pool_create("TEST_IOAT_POOL",
+			256, /* n == num elements */
+			32,  /* cache size */
+			0,   /* priv size */
+			2048, /* data room size */
+			info.socket_id);
+	if (pool == NULL) {
+		printf("Error with mempool creation\n");
+		return -1;
+	}
+
 	/* allocate memory for xstats names and values */
 	nb_xstats = rte_rawdev_xstats_names_get(dev_id, NULL, 0);
 
@@ -78,15 +214,32 @@ run_ioat_tests(int dev_id)
 		return -1;
 	}
 
-	rte_rawdev_xstats_get(dev_id, ids, stats, nb_xstats);
-	for (i = 0; i < nb_xstats; i++)
-		printf("%s: %"PRIu64"   ", snames[i].name, stats[i]);
+	/* run the test cases */
+	for (i = 0; i < 100; i++) {
+		unsigned int j;
+
+		if (test_enqueue_copies(dev_id) != 0)
+			goto err;
+
+		rte_rawdev_xstats_get(dev_id, ids, stats, nb_xstats);
+		for (j = 0; j < nb_xstats; j++)
+			printf("%s: %"PRIu64"   ", snames[j].name, stats[j]);
+		printf("\r");
+	}
 	printf("\n");
 
+	rte_mempool_free(pool);
 	free(snames);
 	free(stats);
 	free(ids);
 	return 0;
+
+err:
+	rte_mempool_free(pool);
+	free(snames);
+	free(stats);
+	free(ids);
+	return -1;
 }
 
 static int
diff --git a/doc/guides/rawdevs/ioat_rawdev.rst b/doc/guides/rawdevs/ioat_rawdev.rst
index 47f12e95c..3ed12c964 100644
--- a/doc/guides/rawdevs/ioat_rawdev.rst
+++ b/doc/guides/rawdevs/ioat_rawdev.rst
@@ -112,6 +112,106 @@ The following code shows how the device is configured in
 Once configured, the device can then be made ready for use by calling the
 ``rte_rawdev_start()`` API.
 
+Performing Data Copies
+~~~~~~~~~~~~~~~~~~~~~~~
+
+To perform data copies using IOAT rawdev devices, the functions
+``rte_ioat_enqueue_copy()`` and ``rte_ioat_do_copies()`` should be used.
+Once copies have been completed, the completion will be reported back when
+the application calls ``rte_ioat_completed_copies()``.
+
+The ``rte_ioat_enqueue_copy()`` function enqueues a single copy to the
+device ring for copying at a later point. The parameters to that function
+include the physical addresses of both the source and destination buffers,
+as well as two "handles" to be returned to the user when the copy is
+completed. These handles can be arbitrary values, but two are provided so
+that the library can track handles for both source and destination on
+behalf of the user, e.g. virtual addresses for the buffers, or mbuf
+pointers if packet data is being copied.
+
+While the ``rte_ioat_enqueue_copy()`` function enqueues a copy operation on
+the device ring, the copy will not actually be performed until after the
+application calls the ``rte_ioat_do_copies()`` function. This function
+informs the device hardware of the elements enqueued on the ring, and the
+device will begin to process them. It is expected that, for efficiency
+reasons, a burst of operations will be enqueued to the device via multiple
+enqueue calls between calls to the ``rte_ioat_do_copies()`` function.
+
+The following code from ``test_ioat_rawdev.c`` demonstrates how to enqueue
+a burst of copies to the device and start the hardware processing of them:
+
+.. code-block:: C
+
+        struct rte_mbuf *srcs[32], *dsts[32];
+        unsigned int j;
+
+        for (i = 0; i < RTE_DIM(srcs); i++) {
+                char *src_data;
+
+                srcs[i] = rte_pktmbuf_alloc(pool);
+                dsts[i] = rte_pktmbuf_alloc(pool);
+                srcs[i]->data_len = srcs[i]->pkt_len = length;
+                dsts[i]->data_len = dsts[i]->pkt_len = length;
+                src_data = rte_pktmbuf_mtod(srcs[i], char *);
+
+                for (j = 0; j < length; j++)
+                        src_data[j] = rand() & 0xFF;
+
+                if (rte_ioat_enqueue_copy(dev_id,
+                                srcs[i]->buf_iova + srcs[i]->data_off,
+                                dsts[i]->buf_iova + dsts[i]->data_off,
+                                length,
+                                (uintptr_t)srcs[i],
+                                (uintptr_t)dsts[i],
+                                0 /* nofence */) != 1) {
+                        printf("Error with rte_ioat_enqueue_copy for buffer %u\n",
+                                        i);
+                        return -1;
+                }
+        }
+        rte_ioat_do_copies(dev_id);
+
+To retrieve information about completed copies, the API
+``rte_ioat_completed_copies()`` should be used. This API will return to the
+application a set of completion handles passed in when the relevant copies
+were enqueued.
+
+The following code from ``test_ioat_rawdev.c`` shows the test code
+retrieving information about the completed copies and validating the data
+is correct before freeing the data buffers using the returned handles:
+
+.. code-block:: C
+
+        if (rte_ioat_completed_copies(dev_id, 64, (void *)completed_src,
+                        (void *)completed_dst) != RTE_DIM(srcs)) {
+                printf("Error with rte_ioat_completed_copies\n");
+                return -1;
+        }
+        for (i = 0; i < RTE_DIM(srcs); i++) {
+                char *src_data, *dst_data;
+
+                if (completed_src[i] != srcs[i]) {
+                        printf("Error with source pointer %u\n", i);
+                        return -1;
+                }
+                if (completed_dst[i] != dsts[i]) {
+                        printf("Error with dest pointer %u\n", i);
+                        return -1;
+                }
+
+                src_data = rte_pktmbuf_mtod(srcs[i], char *);
+                dst_data = rte_pktmbuf_mtod(dsts[i], char *);
+                for (j = 0; j < length; j++)
+                        if (src_data[j] != dst_data[j]) {
+                                printf("Error with copy of packet %u, byte %u\n",
+                                                i, j);
+                                return -1;
+                        }
+                rte_pktmbuf_free(srcs[i]);
+                rte_pktmbuf_free(dsts[i]);
+        }
+
+
 Querying Device Statistics
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
diff --git a/drivers/raw/ioat/rte_ioat_rawdev.h b/drivers/raw/ioat/rte_ioat_rawdev.h
index abe5ba298..7750edba7 100644
--- a/drivers/raw/ioat/rte_ioat_rawdev.h
+++ b/drivers/raw/ioat/rte_ioat_rawdev.h
@@ -14,7 +14,10 @@
  * @b EXPERIMENTAL: these structures and APIs may change without prior notice
  */
 
+#include <x86intrin.h>
+
 #include <rte_memory.h>
+#include <rte_prefetch.h>
 #include <rte_ioat_spec.h>
 
 /** Name of the device driver */
@@ -49,6 +52,10 @@ struct rte_ioat_rawdev {
 	struct rte_ioat_desc *desc_ring;
 	__m128i *hdls; /* completion handles for returning to user */
 
+
+	unsigned short next_read;
+	unsigned short next_write;
+
 	/* some statistics for tracking, if added/changed update xstats fns*/
 	uint64_t enqueue_failed __rte_cache_aligned;
 	uint64_t enqueued;
@@ -72,4 +79,150 @@ struct rte_ioat_desc {
 	uint64_t op_type_specific[4];
 };
 
-#endif
+/**
+ * Enqueue a copy operation onto the ioat device
+ *
+ * This queues up a copy operation to be performed by hardware, but does not
+ * trigger hardware to begin that operation.
+ *
+ * @param dev_id
+ *   The rawdev device id of the ioat instance
+ * @param src
+ *   The physical address of the source buffer
+ * @param dst
+ *   The physical address of the destination buffer
+ * @param length
+ *   The length of the data to be copied
+ * @param src_hdl
+ *   An opaque handle for the source data, to be returned when this operation
+ *   has been completed and the user polls for the completion details
+ * @param dst_hdl
+ *   An opaque handle for the destination data, to be returned when this
+ *   operation has been completed and the user polls for the completion details
+ * @param fence
+ *   A flag parameter indicating that hardware should not begin to perform any
+ *   subsequently enqueued copy operations until after this operation has
+ *   completed
+ * @return
+ *   Number of operations enqueued, either 0 or 1
+ */
+static inline int
+rte_ioat_enqueue_copy(int dev_id, phys_addr_t src, phys_addr_t dst,
+		unsigned int length, uintptr_t src_hdl, uintptr_t dst_hdl,
+		int fence)
+{
+	struct rte_ioat_rawdev *ioat = rte_rawdevs[dev_id].dev_private;
+	unsigned short read = ioat->next_read;
+	unsigned short write = ioat->next_write;
+	unsigned short mask = ioat->ring_size - 1;
+	unsigned short space = mask + read - write;
+
+	if (space == 0) {
+		ioat->enqueue_failed++;
+		return 0;
+	}
+
+	ioat->next_write = write + 1;
+	write &= mask;
+	/* write transfer size and the descriptor control bits */
+	*((uint64_t *)&ioat->desc_ring[write]) =
+			length | (uint64_t)!!fence << 36 |
+			(uint64_t)(!(write & 0xF)) << 35;
+	ioat->desc_ring[write].src_addr = src;
+	ioat->desc_ring[write].dest_addr = dst;
+	ioat->hdls[write] = _mm_set_epi64((__m64)dst_hdl, (__m64)src_hdl);
+	rte_prefetch0(&ioat->desc_ring[ioat->next_write & mask]);
+
+	ioat->enqueued++;
+	return 1;
+}
+
+/**
+ * Trigger hardware to begin performing enqueued copy operations
+ *
+ * This API is used to write the "doorbell" to the hardware to trigger it
+ * to begin the copy operations previously enqueued by rte_ioat_enqueue_copy()
+ *
+ * @param dev_id
+ *   The rawdev device id of the ioat instance
+ */
+static inline void
+rte_ioat_do_copies(int dev_id)
+{
+	struct rte_ioat_rawdev *ioat = rte_rawdevs[dev_id].dev_private;
+	ioat->desc_ring[(ioat->next_write - 1) & (ioat->ring_size - 1)].desc_control = 8;
+	rte_compiler_barrier();
+	ioat->regs->dmacount = ioat->next_write;
+	ioat->started = ioat->enqueued;
+}
+
+/**
+ * @internal
+ * Returns the index of the last completed operation.
+ */
+static inline int
+rte_ioat_get_last_completed(struct rte_ioat_rawdev *ioat)
+{
+	uint64_t status = ioat->status;
+	/* lower 3 bits indicate "transfer status" : active, idle, halted.. */
+	if (status & 0x6)
+		rte_panic("Error with cbdma: %u\n", (unsigned)status & 0x7);
+
+	return (status - ioat->ring_addr) >> 6;
+}
+
+/**
+ * Returns details of copy operations that have been completed
+ *
+ * Returns to the caller the user-provided "handles" for the copy operations
+ * which have been completed by the hardware, and not already returned by
+ * a previous call to this API.
+ *
+ * @param dev_id
+ *   The rawdev device id of the ioat instance
+ * @param max_copies
+ *   The number of entries which can fit in the src_hdls and dst_hdls
+ *   arrays, i.e. max number of completed operations to report
+ * @param src_hdls
+ *   Array to hold the source handle parameters of the completed copies
+ * @param dst_hdls
+ *   Array to hold the destination handle parameters of the completed copies
+ * @return
+ *   Number of completed operations i.e. number of entries written to the
+ *   src_hdls and dst_hdls array parameters.
+ */
+static inline int
+rte_ioat_completed_copies(int dev_id, uint8_t max_copies,
+		uintptr_t *src_hdls, uintptr_t *dst_hdls)
+{
+	struct rte_ioat_rawdev *ioat = rte_rawdevs[dev_id].dev_private;
+	unsigned short mask = (ioat->ring_size - 1);
+	unsigned short read = ioat->next_read;
+	unsigned short end_read = (rte_ioat_get_last_completed(ioat) + 1) & mask;
+	unsigned short count = (end_read - (read & mask)) & mask;
+	int i = 0;
+
+	if (count > max_copies)
+		count = max_copies;
+
+	for (; i < count - 1; i += 2, read += 2) {
+		__m128i hdls0 = _mm_load_si128(&ioat->hdls[read & mask]);
+		__m128i hdls1 = _mm_load_si128(&ioat->hdls[(read + 1) & mask]);
+
+		_mm_storeu_si128((void *)&src_hdls[i],
+				_mm_unpacklo_epi64(hdls0, hdls1));
+		_mm_storeu_si128((void *)&dst_hdls[i],
+				_mm_unpackhi_epi64(hdls0, hdls1));
+	}
+	for (; i < count; i++, read++) {
+		uintptr_t *hdls = (void *)&ioat->hdls[read & mask];
+		src_hdls[i] = hdls[0];
+		dst_hdls[i] = hdls[1];
+	}
+
+	ioat->next_read = read;
+	ioat->completed += count;
+	return count;
+}
+
+#endif /* _RTE_IOAT_RAWDEV_H_ */
-- 
2.21.0