[dpdk-dev] [PATCH 3/5] test/compress: add multi op test

[Pablo de Lara <pablo.de.lara.guarch@intel.com>]

Add test that checks if multiple operations with
different buffers can be handled on a single enqueue call.

Signed-off-by: Pablo de Lara <pablo.de.lara.guarch@intel.com>
---
 test/test/test_compressdev.c | 366 ++++++++++++++++++++++++++++---------------
 1 file changed, 240 insertions(+), 126 deletions(-)

diff --git a/test/test/test_compressdev.c b/test/test/test_compressdev.c
index e92b8142b..ad7585f13 100644
--- a/test/test/test_compressdev.c
+++ b/test/test/test_compressdev.c
@@ -45,6 +45,10 @@ enum zlib_direction {
 	ZLIB_ALL
 };
 
+struct priv_op_data {
+	uint16_t orig_idx;
+};
+
 struct comp_testsuite_params {
 	struct rte_mempool *mbuf_pool;
 	struct rte_mempool *sess_pool;
@@ -114,7 +118,8 @@ testsuite_setup(void)
 		goto exit;
 	}
 	ts_params->op_pool = rte_comp_op_pool_create("op_pool", NUM_OPS,
-						0, 0, rte_socket_id());
+				0, sizeof(struct priv_op_data),
+				rte_socket_id());
 	if (ts_params->op_pool == NULL) {
 		RTE_LOG(ERR, USER1, "Operation pool could not be created\n");
 		goto exit;
@@ -351,7 +356,9 @@ decompress_zlib(struct rte_comp_op *op,
  * Compresses and decompresses buffer with compressdev API and Zlib API
  */
 static int
-test_deflate_comp_decomp(const char *test_buffer,
+test_deflate_comp_decomp(const char * const test_bufs[],
+		unsigned int num_bufs,
+		uint16_t buf_idx[],
 		const struct rte_comp_xform *compress_xform,
 		const struct rte_comp_xform *decompress_xform,
 		enum rte_comp_op_type state,
@@ -359,73 +366,97 @@ test_deflate_comp_decomp(const char *test_buffer,
 {
 	struct comp_testsuite_params *ts_params = &testsuite_params;
 	int ret;
-	struct rte_mbuf *comp_buf = NULL;
-	struct rte_mbuf *uncomp_buf = NULL;
-	struct rte_comp_op *op = NULL;
-	struct rte_comp_op *op_processed = NULL;
+	struct rte_mbuf *uncomp_bufs[num_bufs];
+	struct rte_mbuf *comp_bufs[num_bufs];
+	struct rte_comp_op *ops[num_bufs];
+	struct rte_comp_op *ops_processed[num_bufs];
 	struct rte_comp_session *sess = NULL;
+	uint16_t num_enqd, num_deqd;
+	struct priv_op_data *priv_data;
 	char *data_ptr;
+	unsigned int i;
+
+	/* Initialize all arrays to NULL */
+	memset(uncomp_bufs, 0, sizeof(struct rte_mbuf *) * num_bufs);
+	memset(comp_bufs, 0, sizeof(struct rte_mbuf *) * num_bufs);
+	memset(ops, 0, sizeof(struct rte_comp_op *) * num_bufs);
+	memset(ops_processed, 0, sizeof(struct rte_comp_op *) * num_bufs);
 
-	/* Prepare the source mbuf with the data */
-	uncomp_buf = rte_pktmbuf_alloc(ts_params->mbuf_pool);
-	if (uncomp_buf == NULL) {
+	/* Prepare the source mbufs with the data */
+	ret = rte_pktmbuf_alloc_bulk(ts_params->mbuf_pool, uncomp_bufs, num_bufs);
+	if (ret < 0) {
 		RTE_LOG(ERR, USER1,
-			"Source mbuf could not be allocated "
+			"Source mbufs could not be allocated "
 			"from the mempool\n");
 		goto err;
 	}
 
-	data_ptr = rte_pktmbuf_append(uncomp_buf, strlen(test_buffer) + 1);
-	snprintf(data_ptr, strlen(test_buffer) + 1, "%s", test_buffer);
+	for (i = 0; i < num_bufs; i++) {
+		data_ptr = rte_pktmbuf_append(uncomp_bufs[i], strlen(test_bufs[i]) + 1);
+		snprintf(data_ptr, strlen(test_bufs[i]) + 1, "%s", test_bufs[i]);
+	}
 
-	/* Prepare the destination mbuf */
-	comp_buf = rte_pktmbuf_alloc(ts_params->mbuf_pool);
-	if (comp_buf == NULL) {
+	/* Prepare the destination mbufs */
+	ret = rte_pktmbuf_alloc_bulk(ts_params->mbuf_pool, comp_bufs, num_bufs);
+	if (ret < 0) {
 		RTE_LOG(ERR, USER1,
-			"Destination mbuf could not be allocated "
+			"Destination mbufs could not be allocated "
 			"from the mempool\n");
 		goto err;
 	}
 
-	rte_pktmbuf_append(comp_buf,
-			strlen(test_buffer) * COMPRESS_BUF_SIZE_RATIO);
+	for (i = 0; i < num_bufs; i++)
+		rte_pktmbuf_append(comp_bufs[i],
+			strlen(test_bufs[i]) * COMPRESS_BUF_SIZE_RATIO);
 
 	/* Build the compression operations */
-	op = rte_comp_op_alloc(ts_params->op_pool);
-	if (op == NULL) {
+	ret = rte_comp_op_bulk_alloc(ts_params->op_pool, ops, num_bufs);
+	if (ret < 0) {
 		RTE_LOG(ERR, USER1,
-			"Compress operation could not be allocated "
+			"Compress operations could not be allocated "
 			"from the mempool\n");
 		goto err;
 	}
 
-	op->m_src = uncomp_buf;
-	op->m_dst = comp_buf;
-	op->src.offset = 0;
-	op->src.length = rte_pktmbuf_pkt_len(uncomp_buf);
-	op->dst.offset = 0;
-	if (state == RTE_COMP_OP_STATELESS) {
-		//TODO: FULL or FINAL?
-		op->flush_flag = RTE_COMP_FLUSH_FINAL;
-	} else {
-		RTE_LOG(ERR, USER1,
-			"Stateful operations are not supported "
-			"in these tests yet\n");
-		goto err;
+	for (i = 0; i < num_bufs; i++) {
+		ops[i]->m_src = uncomp_bufs[i];
+		ops[i]->m_dst = comp_bufs[i];
+		ops[i]->src.offset = 0;
+		ops[i]->src.length = rte_pktmbuf_pkt_len(uncomp_bufs[i]);
+		ops[i]->dst.offset = 0;
+		if (state == RTE_COMP_OP_STATELESS) {
+			//TODO: FULL or FINAL?
+			ops[i]->flush_flag = RTE_COMP_FLUSH_FINAL;
+		} else {
+			RTE_LOG(ERR, USER1,
+				"Stateful operations are not supported "
+				"in these tests yet\n");
+			goto err;
+		}
+		ops[i]->input_chksum = 0;
+		/*
+		 * Store original operation index in private data,
+		 * since ordering does not have to be maintained,
+		 * when dequeueing from compressdev, so a comparison
+		 * at the end of the test can be done.
+		 */
+		priv_data = (struct priv_op_data *) (ops[i] + 1);
+		priv_data->orig_idx = i;
 	}
-	op->input_chksum = 0;
 
 	/* Compress data (either with Zlib API or compressdev API */
 	if (zlib_dir == ZLIB_COMPRESS || zlib_dir == ZLIB_ALL) {
-		ret = compress_zlib(op,
-				&compress_xform->compress,
-				DEFAULT_MEM_LEVEL);
-		if (ret < 0)
-			goto err;
-
-		op_processed = op;
+		for (i = 0; i < num_bufs; i++) {
+			ret = compress_zlib(ops[i],
+					&compress_xform->compress,
+					DEFAULT_MEM_LEVEL);
+			if (ret < 0)
+				goto err;
+
+			ops_processed[i] = ops[i];
+		}
 	} else {
-		uint16_t nb_deqd = 0;
+		num_deqd = 0;
 		/* Create compress session */
 		sess = rte_compressdev_session_create(ts_params->sess_pool);
 		ret = rte_compressdev_session_init(0, sess, compress_xform,
@@ -437,19 +468,21 @@ test_deflate_comp_decomp(const char *test_buffer,
 		}
 
 		/* Attach session to operation */
-		op->session = sess;
+		for (i = 0; i < num_bufs; i++)
+			ops[i]->session = sess;
 
 		/* Enqueue and dequeue all operations */
-		ret = rte_compressdev_enqueue_burst(0, 0, &op, 1);
-		if (ret == 0) {
+		num_enqd = rte_compressdev_enqueue_burst(0, 0, ops, num_bufs);
+		if (num_enqd < num_bufs) {
 			RTE_LOG(ERR, USER1,
-				"The operation could not be enqueued\n");
+				"The operations could not be enqueued\n");
 			goto err;
 		}
+
 		do {
-			nb_deqd = rte_compressdev_dequeue_burst(0, 0,
-						&op_processed, 1);
-		} while (nb_deqd < 1);
+			num_deqd += rte_compressdev_dequeue_burst(0, 0,
+					&ops_processed[num_deqd], num_bufs);
+		} while (num_deqd < num_enqd);
 
 		/* Free compress session */
 		rte_compressdev_session_clear(0, sess);
@@ -459,81 +492,104 @@ test_deflate_comp_decomp(const char *test_buffer,
 
 	enum rte_comp_huffman huffman_type =
 		compress_xform->compress.deflate.huffman;
-	RTE_LOG(DEBUG, USER1, "Buffer compressed from %u to %u bytes "
+	for (i = 0; i < num_bufs; i++) {
+		priv_data = (struct priv_op_data *) (ops_processed[i] + 1);
+		RTE_LOG(DEBUG, USER1, "Buffer %u compressed from %u to %u bytes "
 			"(level = %u, huffman = %s)\n",
-			op_processed->consumed, op_processed->produced,
+			buf_idx[priv_data->orig_idx],
+			ops_processed[i]->consumed, ops_processed[i]->produced,
 			compress_xform->compress.level,
 			huffman_type_strings[huffman_type]);
+	}
 
 	/*
-	 * Check operation status and free source mbuf (destination mbuf and
+	 * Check operation status and free source mbufs (destination mbuf and
 	 * compress operation information is needed for the decompression stage)
 	 */
-	if (op_processed->status != RTE_COMP_OP_STATUS_SUCCESS) {
-		RTE_LOG(ERR, USER1,
-			"Some operations were not successful\n");
-		goto err;
+	for (i = 0; i < num_bufs; i++) {
+		if (ops_processed[i]->status != RTE_COMP_OP_STATUS_SUCCESS) {
+			RTE_LOG(ERR, USER1,
+				"Some operations were not successful\n");
+			goto err;
+		}
+		priv_data = (struct priv_op_data *) (ops_processed[i] + 1);
+		rte_pktmbuf_free(uncomp_bufs[priv_data->orig_idx]);
+		uncomp_bufs[priv_data->orig_idx] = NULL;
 	}
-	rte_pktmbuf_free(uncomp_buf);
-	uncomp_buf = NULL;
 
-	/* Allocate buffer for decompressed data */
-	comp_buf = rte_pktmbuf_alloc(ts_params->mbuf_pool);
-	if (comp_buf == NULL) {
+	/* Allocate buffers for decompressed data */
+	ret = rte_pktmbuf_alloc_bulk(ts_params->mbuf_pool, uncomp_bufs, num_bufs);
+	if (ret < 0) {
 		RTE_LOG(ERR, USER1,
-			"Destination mbuf could not be allocated "
+			"Destination mbufs could not be allocated "
 			"from the mempool\n");
 		goto err;
 	}
 
-	rte_pktmbuf_append(comp_buf, strlen(test_buffer) + 1);
+	for (i = 0; i < num_bufs; i++) {
+		priv_data = (struct priv_op_data *) (ops_processed[i] + 1);
+		rte_pktmbuf_append(uncomp_bufs[i],
+				strlen(test_bufs[priv_data->orig_idx]) + 1);
+	}
 
 	/* Build the decompression operations */
-	op = rte_comp_op_alloc(ts_params->op_pool);
-	if (op == NULL) {
+	ret = rte_comp_op_bulk_alloc(ts_params->op_pool, ops, num_bufs);
+	if (ret < 0) {
 		RTE_LOG(ERR, USER1,
-			"Decompress operation could not be allocated "
+			"Decompress operations could not be allocated "
 			"from the mempool\n");
 		goto err;
 	}
 
-	/* Source buffer is the compressed data from the previous operation */
-	op->m_src = op_processed->m_dst;
-	op->m_dst = uncomp_buf;
-	op->src.offset = 0;
-	/*
-	 * Set the length of the compressed data to the
-	 * number of bytes that were produced in the previous stage
-	 */
-	op->src.length = op_processed->produced;
-	op->dst.offset = 0;
-	if (state == RTE_COMP_OP_STATELESS) {
-		//TODO: FULL or FINAL?
-		op->flush_flag = RTE_COMP_FLUSH_FINAL;
-	} else {
-		RTE_LOG(ERR, USER1,
-			"Stateful operations are not supported "
-			"in these tests yet\n");
-		goto err;
+	/* Source buffer is the compressed data from the previous operations */
+	for (i = 0; i < num_bufs; i++) {
+		ops[i]->m_src = ops_processed[i]->m_dst;
+		ops[i]->m_dst = uncomp_bufs[i];
+		ops[i]->src.offset = 0;
+		/*
+		 * Set the length of the compressed data to the
+		 * number of bytes that were produced in the previous stage
+		 */
+		ops[i]->src.length = ops_processed[i]->produced;
+		ops[i]->dst.offset = 0;
+		if (state == RTE_COMP_OP_STATELESS) {
+			//TODO: FULL or FINAL?
+			ops[i]->flush_flag = RTE_COMP_FLUSH_FINAL;
+		} else {
+			RTE_LOG(ERR, USER1,
+				"Stateful operations are not supported "
+				"in these tests yet\n");
+			goto err;
+		}
+		ops[i]->input_chksum = 0;
+		/*
+		 * Copy private data from previous operations,
+		 * to keep the pointer to the original buffer
+		 */
+		memcpy(ops[i] + 1, ops_processed[i] + 1, sizeof(struct priv_op_data));
 	}
-	op->input_chksum = 0;
 
 	/*
-	 * Free the previous compress operation,
+	 * Free the previous compress operations,
 	 * as it is not needed anymore
 	 */
-	rte_comp_op_free(op_processed);
-	op_processed = NULL;
+	for (i = 0; i < num_bufs; i++) {
+		rte_comp_op_free(ops_processed[i]);
+		ops_processed[i] = NULL;
+	}
 
 	/* Decompress data (either with Zlib API or compressdev API */
 	if (zlib_dir == ZLIB_DECOMPRESS || zlib_dir == ZLIB_ALL) {
-		ret = decompress_zlib(op, &decompress_xform->decompress);
-		if (ret < 0)
-			goto err;
+		for (i = 0; i < num_bufs; i++) {
+			ret = decompress_zlib(ops[i],
+					&decompress_xform->decompress);
+			if (ret < 0)
+				goto err;
 
-		op_processed = op;
+			ops_processed[i] = ops[i];
+		}
 	} else {
-		uint16_t nb_deqd = 0;
+		num_deqd = 0;
 		/* Create compress session */
 		sess = rte_compressdev_session_create(ts_params->sess_pool);
 		ret = rte_compressdev_session_init(0, sess, decompress_xform,
@@ -545,19 +601,21 @@ test_deflate_comp_decomp(const char *test_buffer,
 		}
 
 		/* Attach session to operation */
-		op->session = sess;
+		for (i = 0; i < num_bufs; i++)
+			ops[i]->session = sess;
 
 		/* Enqueue and dequeue all operations */
-		ret = rte_compressdev_enqueue_burst(0, 0, &op, 1);
-		if (ret == 0) {
+		num_enqd = rte_compressdev_enqueue_burst(0, 0, ops, num_bufs);
+		if (num_enqd < num_bufs) {
 			RTE_LOG(ERR, USER1,
-				"The operation could not be enqueued\n");
+				"The operations could not be enqueued\n");
 			goto err;
 		}
+
 		do {
-			nb_deqd = rte_compressdev_dequeue_burst(0, 0,
-						&op_processed, 1);
-		} while (nb_deqd < 1);
+			num_deqd += rte_compressdev_dequeue_burst(0, 0,
+					&ops_processed[num_deqd], num_bufs);
+		} while (num_deqd < num_enqd);
 
 		/* Free compress session */
 		rte_compressdev_session_clear(0, sess);
@@ -565,44 +623,62 @@ test_deflate_comp_decomp(const char *test_buffer,
 		sess = NULL;
 	}
 
-	RTE_LOG(DEBUG, USER1, "Buffer decompressed from %u to %u bytes\n",
-			op_processed->consumed, op_processed->produced);
+	for (i = 0; i < num_bufs; i++) {
+		priv_data = (struct priv_op_data *) (ops_processed[i] + 1);
+		RTE_LOG(DEBUG, USER1, "Buffer %u decompressed from %u to %u bytes\n",
+			buf_idx[priv_data->orig_idx],
+			ops_processed[i]->consumed, ops_processed[i]->produced);
+	}
+
 	/*
 	 * Check operation status and free source mbuf (destination mbuf and
 	 * compress operation information is still needed)
 	 */
-	if (op_processed->status != RTE_COMP_OP_STATUS_SUCCESS) {
-		RTE_LOG(ERR, USER1,
-			"Some operations were not successful\n");
-		goto err;
+	for (i = 0; i < num_bufs; i++) {
+		if (ops_processed[i]->status != RTE_COMP_OP_STATUS_SUCCESS) {
+			RTE_LOG(ERR, USER1,
+				"Some operations were not successful\n");
+			goto err;
+		}
+		priv_data = (struct priv_op_data *) (ops_processed[i] + 1);
+		rte_pktmbuf_free(comp_bufs[priv_data->orig_idx]);
+		comp_bufs[priv_data->orig_idx] = NULL;
 	}
-	rte_pktmbuf_free(comp_buf);
-	comp_buf = NULL;
 
 	/*
 	 * Compare the original stream with the decompressed stream
 	 * (in size and the data)
 	 */
-	if (compare_buffers(test_buffer, strlen(test_buffer) + 1,
-			rte_pktmbuf_mtod(op_processed->m_dst, const char *),
-			op_processed->produced) < 0)
-		goto err;
+	for (i = 0; i < num_bufs; i++) {
+		priv_data = (struct priv_op_data *) (ops_processed[i] + 1);
+		const char *buf1 = test_bufs[priv_data->orig_idx];
+		const char *buf2 = rte_pktmbuf_mtod(ops_processed[i]->m_dst,
+				const char *);
+
+		if (compare_buffers(buf1, strlen(buf1) + 1,
+				buf2, ops_processed[i]->produced) < 0)
+			goto err;
+	}
 
 	/* Free rest of resources */
-	rte_pktmbuf_free(uncomp_buf);
-	uncomp_buf = NULL;
+	for (i = 0; i < num_bufs; i++) {
+		priv_data = (struct priv_op_data *) (ops_processed[i] + 1);
+		rte_pktmbuf_free(uncomp_bufs[priv_data->orig_idx]);
+		uncomp_bufs[priv_data->orig_idx] = NULL;
 
-	rte_comp_op_free(op_processed);
-	op_processed = NULL;
+		rte_comp_op_free(ops_processed[i]);
+		ops_processed[i] = NULL;
+	}
 
 	return 0;
 
 err:
-	rte_pktmbuf_free(uncomp_buf);
-	rte_pktmbuf_free(comp_buf);
-	rte_comp_op_free(op);
-	rte_comp_op_free(op_processed);
-
+	for (i = 0; i < num_bufs; i++) {
+		rte_pktmbuf_free(uncomp_bufs[i]);
+		rte_pktmbuf_free(comp_bufs[i]);
+		rte_comp_op_free(ops[i]);
+		rte_comp_op_free(ops_processed[i]);
+	}
 	rte_compressdev_session_clear(0, sess);
 	rte_compressdev_session_terminate(sess);
 
@@ -625,7 +701,8 @@ test_compressdev_deflate_stateless_fixed(void)
 		test_buffer = compress_test_bufs[i];
 
 		/* Compress with compressdev, decompress with Zlib */
-		if (test_deflate_comp_decomp(test_buffer,
+		if (test_deflate_comp_decomp(&test_buffer, 1,
+				&i,
 				&compress_xform,
 				&ts_params->def_decomp_xform,
 				RTE_COMP_OP_STATELESS,
@@ -633,7 +710,8 @@ test_compressdev_deflate_stateless_fixed(void)
 			return TEST_FAILED;
 
 		/* Compress with Zlib, decompress with compressdev */
-		if (test_deflate_comp_decomp(test_buffer,
+		if (test_deflate_comp_decomp(&test_buffer, 1,
+				&i,
 				&compress_xform,
 				&ts_params->def_decomp_xform,
 				RTE_COMP_OP_STATELESS,
@@ -660,7 +738,8 @@ test_compressdev_deflate_stateless_dynamic(void)
 		test_buffer = compress_test_bufs[i];
 
 		/* Compress with compressdev, decompress with Zlib */
-		if (test_deflate_comp_decomp(test_buffer,
+		if (test_deflate_comp_decomp(&test_buffer, 1,
+				&i,
 				&compress_xform,
 				&ts_params->def_decomp_xform,
 				RTE_COMP_OP_STATELESS,
@@ -668,7 +747,8 @@ test_compressdev_deflate_stateless_dynamic(void)
 			return TEST_FAILED;
 
 		/* Compress with Zlib, decompress with compressdev */
-		if (test_deflate_comp_decomp(test_buffer,
+		if (test_deflate_comp_decomp(&test_buffer, 1,
+				&i,
 				&compress_xform,
 				&ts_params->def_decomp_xform,
 				RTE_COMP_OP_STATELESS,
@@ -679,6 +759,38 @@ test_compressdev_deflate_stateless_dynamic(void)
 	return TEST_SUCCESS;
 }
 
+static int
+test_compressdev_deflate_stateless_multi_op(void)
+{
+	struct comp_testsuite_params *ts_params = &testsuite_params;
+	uint16_t num_bufs = RTE_DIM(compress_test_bufs);
+	uint16_t buf_idx[num_bufs];
+	uint16_t i;
+
+	for (i = 0; i < num_bufs; i++)
+		buf_idx[i] = i;
+
+	/* Compress with compressdev, decompress with Zlib */
+	if (test_deflate_comp_decomp(compress_test_bufs, num_bufs,
+			buf_idx,
+			&ts_params->def_comp_xform,
+			&ts_params->def_decomp_xform,
+			RTE_COMP_OP_STATELESS,
+			ZLIB_DECOMPRESS) < 0)
+		return TEST_FAILED;
+
+	/* Compress with Zlib, decompress with compressdev */
+	if (test_deflate_comp_decomp(compress_test_bufs, num_bufs,
+			buf_idx,
+			&ts_params->def_comp_xform,
+			&ts_params->def_decomp_xform,
+			RTE_COMP_OP_STATELESS,
+			ZLIB_COMPRESS) < 0)
+		return TEST_FAILED;
+
+	return TEST_SUCCESS;
+}
+
 static struct unit_test_suite compressdev_testsuite  = {
 	.suite_name = "compressdev unit test suite",
 	.setup = testsuite_setup,
@@ -688,6 +800,8 @@ static struct unit_test_suite compressdev_testsuite  = {
 			test_compressdev_deflate_stateless_fixed),
 		TEST_CASE_ST(generic_ut_setup, generic_ut_teardown,
 			test_compressdev_deflate_stateless_dynamic),
+		TEST_CASE_ST(generic_ut_setup, generic_ut_teardown,
+			test_compressdev_deflate_stateless_multi_op),
 		TEST_CASES_END() /**< NULL terminate unit test array */
 	}
 };
-- 
2.14.3