[dpdk-dev] [PATCH v2] compress/isal: add chained mbuf support

[Lee, Daly]

This patch adds chained mbuf support for input or output buffers
during compression/decompression operations.

V2 - Fixed using chained mbufs with offsets.

Signed-off-by: Lee Daly <lee.daly at intel.com>
---
 doc/guides/compressdevs/features/isal.ini     |   1 +
 doc/guides/compressdevs/isal.rst              |   2 -
 drivers/compress/isal/isal_compress_pmd.c     | 397 ++++++++++++++++++++------
 drivers/compress/isal/isal_compress_pmd_ops.c |   5 +-
 4 files changed, 321 insertions(+), 84 deletions(-)

diff --git a/doc/guides/compressdevs/features/isal.ini b/doc/guides/compressdevs/features/isal.ini
index 7183d10..06a95fb 100644
--- a/doc/guides/compressdevs/features/isal.ini
+++ b/doc/guides/compressdevs/features/isal.ini
@@ -8,6 +8,7 @@ CPU SSE        = Y
 CPU AVX        = Y
 CPU AVX2       = Y
 CPU AVX512     = Y
+Chained mbufs  = Y
 Deflate        = Y
 Fixed          = Y
 Dynamic        = Y
diff --git a/doc/guides/compressdevs/isal.rst b/doc/guides/compressdevs/isal.rst
index 276ff06..3bc3022 100644
--- a/doc/guides/compressdevs/isal.rst
+++ b/doc/guides/compressdevs/isal.rst
@@ -78,8 +78,6 @@ As a result the level mappings from the API to the PMD are shown below.
 Limitations
 -----------
 
-* Chained mbufs will not be supported until a future release, meaning max data size being passed to PMD through a single mbuf is 64K - 1. If data is larger than this it will need to be split up and sent as multiple operations.
-
 * Compressdev level 0, no compression, is not supported.
 
 * Checksums will not be supported until future release.
diff --git a/drivers/compress/isal/isal_compress_pmd.c b/drivers/compress/isal/isal_compress_pmd.c
index 747ded1..70231c4 100644
--- a/drivers/compress/isal/isal_compress_pmd.c
+++ b/drivers/compress/isal/isal_compress_pmd.c
@@ -188,6 +188,211 @@ isal_comp_set_priv_xform_parameters(struct isal_priv_xform *priv_xform,
 	return 0;
 }
 
+/* Compression using chained mbufs for input/output data */
+static int
+chained_mbuf_compression(struct rte_comp_op *op, struct isal_comp_qp *qp)
+{
+	int ret, i;
+	uint8_t segs;
+	uint32_t consumed_data;
+	uint32_t remaining_data = op->src.length;
+	struct rte_mbuf *src = op->m_src;
+	struct rte_mbuf *dst = op->m_dst;
+
+	/* check for offset passing multiple segments
+	 * and point compression stream to input/output buffer
+	 */
+	if (op->src.offset > src->data_len) {
+		segs = op->src.offset / src->data_len;
+		for (i = 0; i < segs; i++)
+			src = src->next;
+
+		qp->stream->avail_in = src->data_len -
+				(op->src.offset % src->data_len);
+		qp->stream->next_in = rte_pktmbuf_mtod_offset(src, uint8_t *,
+				(op->src.offset % src->data_len));
+	} else {
+		qp->stream->avail_in = src->data_len - op->src.offset;
+		qp->stream->next_in = rte_pktmbuf_mtod_offset(src, uint8_t *,
+				op->src.offset);
+	}
+
+	if (op->dst.offset > dst->data_len) {
+		segs = op->dst.offset / dst->data_len;
+		for (int i = 0; i < segs; i++)
+			dst = dst->next;
+
+		qp->stream->avail_out = dst->data_len -
+				(op->dst.offset % dst->data_len);
+		qp->stream->next_out = rte_pktmbuf_mtod_offset(dst, uint8_t *,
+				(op->dst.offset % dst->data_len));
+	} else {
+		qp->stream->avail_out = dst->data_len - op->dst.offset;
+		qp->stream->next_out = rte_pktmbuf_mtod_offset(dst, uint8_t *,
+				op->dst.offset);
+	}
+
+	if (unlikely(!qp->stream->next_in || !qp->stream->next_out)) {
+		ISAL_PMD_LOG(ERR, "Invalid source or destination buffer\n");
+		op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+		return -1;
+	}
+
+	while (qp->stream->internal_state.state != ZSTATE_END) {
+		/* Last segment of data */
+		if (remaining_data <= src->data_len)
+			qp->stream->end_of_stream = 1;
+
+		/* Execute compression operation */
+		ret = isal_deflate(qp->stream);
+
+		if (remaining_data == op->src.length) {
+			consumed_data = src->data_len - qp->stream->avail_in -
+					(op->src.offset % src->data_len);
+		} else
+			consumed_data = src->data_len - qp->stream->avail_in;
+
+		remaining_data -= consumed_data;
+
+		if (ret != COMP_OK) {
+			ISAL_PMD_LOG(ERR, "Compression operation failed\n");
+			op->status = RTE_COMP_OP_STATUS_ERROR;
+			return ret;
+		}
+
+		if (qp->stream->avail_in == 0 &&
+				qp->stream->total_in != op->src.length) {
+			if (src->next != NULL) {
+				src = src->next;
+				qp->stream->next_in =
+						rte_pktmbuf_mtod(src, uint8_t *);
+				qp->stream->avail_in =
+					RTE_MIN(remaining_data, src->data_len);
+			} else {
+				ISAL_PMD_LOG(ERR,
+				"Not enough input buffer segments\n"
+					"Output Produced: %d\n", op->produced);
+				op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+				return -1;
+			}
+		}
+
+		if (qp->stream->avail_out == 0 &&
+				qp->stream->internal_state.state != ZSTATE_END) {
+			if (dst->next != NULL) {
+				dst = dst->next;
+				qp->stream->next_out =
+						rte_pktmbuf_mtod(dst, uint8_t *);
+				qp->stream->avail_out = dst->data_len;
+			} else {
+				ISAL_PMD_LOG(ERR,
+				"Not enough output buffer segments\n"
+					"Output Produced: %d\n", op->produced);
+				op->status =
+				RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED;
+				return -1;
+			}
+		}
+	}
+	op->consumed = qp->stream->total_in;
+	op->produced = qp->stream->total_out;
+
+	return 0;
+}
+
+/* Decompression using chained mbufs for input/output data */
+static int
+chained_mbuf_decompression(struct rte_comp_op *op, struct isal_comp_qp *qp)
+{
+	int ret, i;
+	uint8_t segs;
+	uint32_t consumed_data;
+	uint32_t remaining_data = op->src.length;
+	struct rte_mbuf *src = op->m_src;
+	struct rte_mbuf *dst = op->m_dst;
+
+	/* check for offset passing multiple segments
+	 * and point decompression state to input/output buffer
+	 */
+	if (op->src.offset > src->data_len) {
+		segs = op->src.offset / src->data_len;
+		for (i = 0; i < segs; i++)
+			src = src->next;
+
+		qp->state->avail_in =
+				src->data_len - (op->src.offset % src->data_len);
+		qp->state->next_in = rte_pktmbuf_mtod_offset(src,
+				uint8_t *, (op->src.offset % src->data_len));
+	} else {
+		qp->state->avail_in = src->data_len - op->src.offset;
+		qp->state->next_in = rte_pktmbuf_mtod_offset(src, uint8_t *,
+				op->src.offset);
+	}
+	if (op->dst.offset > dst->data_len) {
+		segs = op->dst.offset / dst->data_len;
+		for (i = 0; i < segs; i++)
+			dst = dst->next;
+
+		qp->state->avail_out =
+				dst->data_len - (op->dst.offset % dst->data_len);
+		qp->state->next_out = rte_pktmbuf_mtod_offset(dst, uint8_t *,
+				(op->dst.offset % dst->data_len));
+	} else {
+		qp->state->avail_out = dst->data_len - op->dst.offset;
+		qp->state->next_out = rte_pktmbuf_mtod_offset(dst, uint8_t *,
+				op->dst.offset);
+	}
+	while (qp->state->block_state != ISAL_BLOCK_FINISH) {
+
+		ret = isal_inflate(qp->state);
+
+		if (remaining_data == op->src.length) {
+			consumed_data = src->data_len - qp->state->avail_in -
+					(op->src.offset % src->data_len);
+		} else
+			consumed_data = src->data_len - qp->state->avail_in;
+
+		op->consumed += consumed_data;
+		remaining_data -= consumed_data;
+
+		if (ret != ISAL_DECOMP_OK) {
+			ISAL_PMD_LOG(ERR, "Decompression operation failed\n");
+			op->status = RTE_COMP_OP_STATUS_ERROR;
+			return ret;
+		}
+
+		if (qp->state->avail_in == 0
+				&& op->consumed != op->src.length) {
+			if (src->next != NULL) {
+				src = src->next;
+				qp->state->next_in =
+						rte_pktmbuf_mtod(src, uint8_t *);
+				qp->state->avail_in =
+					RTE_MIN(remaining_data, src->data_len);
+			}
+		}
+
+		if (qp->state->avail_out == 0 &&
+				qp->state->block_state != ISAL_BLOCK_FINISH) {
+			if (dst->next != NULL) {
+				dst = dst->next;
+				qp->state->next_out =
+						rte_pktmbuf_mtod(dst, uint8_t *);
+				qp->state->avail_out = dst->data_len;
+			} else {
+				ISAL_PMD_LOG(ERR,
+				"Not enough output buffer segments\n");
+				op->status =
+				RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED;
+				return -1;
+			}
+		}
+	}
+	op->produced = qp->state->total_out;
+
+	return 0;
+}
+
 /* Stateless Compression Function */
 static int
 process_isal_deflate(struct rte_comp_op *op, struct isal_comp_qp *qp,
@@ -207,7 +412,7 @@ process_isal_deflate(struct rte_comp_op *op, struct isal_comp_qp *qp,
 	/* Stateless operation, input will be consumed in one go */
 	qp->stream->flush = NO_FLUSH;
 
-	/* set op level & intermediate level buffer */
+	/* set compression level & intermediate level buffer size */
 	qp->stream->level = priv_xform->compress.level;
 	qp->stream->level_buf_size = priv_xform->level_buffer_size;
 
@@ -225,57 +430,72 @@ process_isal_deflate(struct rte_comp_op *op, struct isal_comp_qp *qp,
 		isal_deflate_set_hufftables(qp->stream, NULL,
 				IGZIP_HUFFTABLE_DEFAULT);
 
-	qp->stream->end_of_stream = 1; /* All input consumed in one go */
-	if ((op->src.length + op->src.offset) > op->m_src->data_len) {
-		ISAL_PMD_LOG(ERR, "Input mbuf not big enough for offset.\n");
+	if (op->m_src->pkt_len < (op->src.length + op->src.offset)) {
+		ISAL_PMD_LOG(ERR, "Input mbuf(s) not big enough.\n");
 		op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
 		return -1;
 	}
-	/* Point compression stream to input buffer */
-	qp->stream->avail_in = op->src.length;
-	qp->stream->next_in = rte_pktmbuf_mtod_offset(op->m_src, uint8_t *,
-			op->src.offset);
 
-	if (op->dst.offset > op->m_dst->data_len) {
-		ISAL_PMD_LOG(ERR, "Output mbuf not big enough for offset.\n");
-		op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
-		return -1;
-	}
-	/*  Point compression stream to output buffer */
-	qp->stream->avail_out = op->m_dst->data_len;
-	qp->stream->next_out  = rte_pktmbuf_mtod_offset(op->m_dst, uint8_t *,
-		op->dst.offset);
+	/* Chained mbufs */
+	if (op->m_src->nb_segs > 1 || op->m_dst->nb_segs > 1) {
+		ret = chained_mbuf_compression(op, qp);
+		if (ret < 0) //check the returns... ensure op status changed.
+			return ret;
+	} else {
+	/* Linear buffer */
+		qp->stream->end_of_stream = 1; /* All input consumed in one go */
+		if ((op->src.length + op->src.offset) > op->m_src->data_len) {
+			ISAL_PMD_LOG(ERR, "Input mbuf not big enough for offset.\n");
+			op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+			return -1;
+		}
+		/* Point compression stream to input buffer */
+		qp->stream->avail_in = op->src.length;
+		qp->stream->next_in = rte_pktmbuf_mtod_offset(op->m_src, uint8_t *,
+				op->src.offset);
+
+		if (op->dst.offset > op->m_dst->data_len) {
+			ISAL_PMD_LOG(ERR, "Output mbuf not big enough for offset.\n");
+			op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+			return -1;
+		}
+		/*  Point compression stream to output buffer */
+		qp->stream->avail_out = op->m_dst->data_len;
+		qp->stream->next_out  = rte_pktmbuf_mtod_offset(op->m_dst, uint8_t *,
+			op->dst.offset);
 
-	if (unlikely(!qp->stream->next_in || !qp->stream->next_out)) {
-		ISAL_PMD_LOG(ERR, "Invalid source or destination buffers\n");
-		op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
-		return -1;
-	}
+		if (unlikely(!qp->stream->next_in || !qp->stream->next_out)) {
+			ISAL_PMD_LOG(ERR, "Invalid source or destination buffers\n");
+			op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+			return -1;
+		}
 
-	/* Execute compression operation */
-	ret =  isal_deflate_stateless(qp->stream);
+		/* Execute compression operation */
+		ret =  isal_deflate_stateless(qp->stream);
 
-	/* Check that output buffer did not run out of space */
-	if (ret == STATELESS_OVERFLOW) {
-		ISAL_PMD_LOG(ERR, "Output buffer not big enough\n");
-		op->status = RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED;
-		return ret;
-	}
+		/* Check that output buffer did not run out of space */
+		if (ret == STATELESS_OVERFLOW) {
+			ISAL_PMD_LOG(ERR, "Output buffer not big enough\n");
+			op->status = RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED;
+			return ret;
+		}
 
-	/* Check that input buffer has been fully consumed */
-	if (qp->stream->avail_in != (uint32_t)0) {
-		ISAL_PMD_LOG(ERR, "Input buffer could not be read entirely\n");
-		op->status = RTE_COMP_OP_STATUS_ERROR;
-		return -1;
-	}
+		/* Check that input buffer has been fully consumed */
+		if (qp->stream->avail_in != (uint32_t)0) {
+			ISAL_PMD_LOG(ERR, "Input buffer could not be read entirely\n");
+			op->status = RTE_COMP_OP_STATUS_ERROR;
+			return -1;
+		}
 
-	if (ret != COMP_OK) {
-		op->status = RTE_COMP_OP_STATUS_ERROR;
-		return ret;
-	}
+		if (ret != COMP_OK) {
+			ISAL_PMD_LOG(ERR, "Compression operation failed\n");
+			op->status = RTE_COMP_OP_STATUS_ERROR;
+			return ret;
+		}
 
-	op->consumed = qp->stream->total_in;
-	op->produced = qp->stream->total_out;
+		op->consumed = qp->stream->total_in;
+		op->produced = qp->stream->total_out;
+	}
 
 	return ret;
 }
@@ -291,57 +511,72 @@ process_isal_inflate(struct rte_comp_op *op, struct isal_comp_qp *qp)
 	/* Initialize decompression state */
 	isal_inflate_init(qp->state);
 
-	if ((op->src.length + op->src.offset) > op->m_src->data_len) {
-		ISAL_PMD_LOG(ERR, "Input mbuf not big enough for offset.\n");
+	if (op->m_src->pkt_len < (op->src.length + op->src.offset)) {
+		ISAL_PMD_LOG(ERR, "Input mbuf(s) not big enough.\n");
 		op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
 		return -1;
 	}
-	/* Point decompression state to input buffer */
-	qp->state->avail_in = op->src.length;
-	qp->state->next_in = rte_pktmbuf_mtod_offset(op->m_src, uint8_t *,
-			op->src.offset);
 
-	if (op->dst.offset > op->m_dst->data_len) {
-		ISAL_PMD_LOG(ERR, "Output mbuf not big enough for offset.\n");
-		op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+	/* Chained mbufs */
+	if (op->m_src->nb_segs > 1 || op->m_dst->nb_segs > 1) {
+		ret = chained_mbuf_decompression(op, qp);
+		if (ret !=  0)
+			return ret;
+	} else {
+		/* Linear buffer */
+		if ((op->src.length + op->src.offset) > op->m_src->data_len) {
+			ISAL_PMD_LOG(ERR, "Input mbuf not big enough for offset.\n");
+			op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
 		return -1;
-	}
-	/* Point decompression state to output buffer */
-	qp->state->avail_out = op->m_dst->data_len;
-	qp->state->next_out  = rte_pktmbuf_mtod_offset(op->m_dst, uint8_t *,
-			op->dst.offset);
+		}
+		/* Point decompression state to input buffer */
+		qp->state->avail_in = op->src.length;
+		qp->state->next_in = rte_pktmbuf_mtod_offset(op->m_src, uint8_t *,
+				op->src.offset);
+
+		if (op->dst.offset > op->m_dst->data_len) {
+			ISAL_PMD_LOG(ERR, "Output mbuf not big enough for offset.\n");
+			op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+			return -1;
+		}
+		/* Point decompression state to output buffer */
+		qp->state->avail_out = op->m_dst->data_len;
+		qp->state->next_out  = rte_pktmbuf_mtod_offset(op->m_dst, uint8_t *,
+				op->dst.offset);
+
+		if (unlikely(!qp->state->next_in || !qp->state->next_out)) {
+			ISAL_PMD_LOG(ERR, "Invalid source or destination buffers\n");
+			op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
+			return -1;
+		}
 
-	if (unlikely(!qp->state->next_in || !qp->state->next_out)) {
-		ISAL_PMD_LOG(ERR, "Invalid source or destination buffers\n");
-		op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
-		return -1;
-	}
+		/* Execute decompression operation */
+		ret = isal_inflate_stateless(qp->state);
 
-	/* Execute decompression operation */
-	ret = isal_inflate_stateless(qp->state);
+		if (ret == ISAL_OUT_OVERFLOW) {
+			ISAL_PMD_LOG(ERR, "Output buffer not big enough\n");
+			op->status = RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED;
+			return ret;
+		}
 
-	if (ret == ISAL_OUT_OVERFLOW) {
-		ISAL_PMD_LOG(ERR, "Output buffer not big enough\n");
-		op->status = RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED;
-		return ret;
-	}
+		/* Check that input buffer has been fully consumed */
+		if (qp->state->avail_in != (uint32_t)0) {
+			ISAL_PMD_LOG(ERR, "Input buffer could not be read entirely\n");
+			op->status = RTE_COMP_OP_STATUS_ERROR;
+			return -1;
+		}
 
-	/* Check that input buffer has been fully consumed */
-	if (qp->state->avail_in != (uint32_t)0) {
-		ISAL_PMD_LOG(ERR, "Input buffer could not be read entirely\n");
-		op->status = RTE_COMP_OP_STATUS_ERROR;
-		return -1;
-	}
+		if (ret != ISAL_DECOMP_OK) {
+			ISAL_PMD_LOG(ERR, "Decompression operation failed\n");
+			op->status = RTE_COMP_OP_STATUS_ERROR;
+			return ret;
+		}
 
-	if (ret != ISAL_DECOMP_OK) {
-		op->status = RTE_COMP_OP_STATUS_ERROR;
-		return ret;
+		op->consumed = op->src.length - qp->state->avail_in;
+		op->produced = qp->state->total_out;
 	}
 
-	op->consumed = op->src.length - qp->state->avail_in;
-	op->produced = qp->state->total_out;
-
-return ret;
+	return ret;
 }
 
 /* Process compression/decompression operation */
diff --git a/drivers/compress/isal/isal_compress_pmd_ops.c b/drivers/compress/isal/isal_compress_pmd_ops.c
index 585f228..b8c0ca1 100644
--- a/drivers/compress/isal/isal_compress_pmd_ops.c
+++ b/drivers/compress/isal/isal_compress_pmd_ops.c
@@ -12,7 +12,10 @@
 static const struct rte_compressdev_capabilities isal_pmd_capabilities[] = {
 	{
 		.algo = RTE_COMP_ALGO_DEFLATE,
-		.comp_feature_flags =	RTE_COMP_FF_SHAREABLE_PRIV_XFORM |
+		.comp_feature_flags =	RTE_COMP_FF_OOP_SGL_IN_SGL_OUT |
+					RTE_COMP_FF_OOP_SGL_IN_LB_OUT |
+					RTE_COMP_FF_OOP_LB_IN_SGL_OUT |
+					RTE_COMP_FF_SHAREABLE_PRIV_XFORM |
 					RTE_COMP_FF_HUFFMAN_FIXED |
 					RTE_COMP_FF_HUFFMAN_DYNAMIC,
 		.window_size = {
-- 
2.7.4