Hi,
> -----Original Message-----
> From: Shally Verma [mailto:shally.ve...@caviumnetworks.com]
> Sent: Monday, July 2, 2018 5:57 PM
> To: De Lara Guarch, Pablo <pablo.de.lara.gua...@intel.com>
> Cc: dev@dpdk.org; pathr...@caviumnetworks.com;
> mcha...@caviumnetworks.com; Sunila Sahu <ss...@caviumnetworks.com>;
> Sunila Sahu <sunila.s...@caviumnetworks.com>; Ashish Gupta
> <ashish.gu...@caviumnetworks.com>
> Subject: [PATCH v2 4/5] compress/zlib: add enq deq apis
Better retitle to "support burst enqueue/dequeue".
>
> From: Sunila Sahu <ss...@caviumnetworks.com>
>
> implement enqueue and dequeue apis
This should start with capital letter, but this is probably not needed anyway.
>
> Signed-off-by: Sunila Sahu <sunila.s...@caviumnetworks.com>
> Signed-off-by: Shally Verma <shally.ve...@caviumnetworks.com>
> Signed-off-by: Ashish Gupta <ashish.gu...@caviumnetworks.com>
> ---
> drivers/compress/zlib/zlib_pmd.c | 238
> ++++++++++++++++++++++++++++++++++++++-
> 1 file changed, 237 insertions(+), 1 deletion(-)
>
> diff --git a/drivers/compress/zlib/zlib_pmd.c
> b/drivers/compress/zlib/zlib_pmd.c
> index 7c2614e..aef23de 100644
> --- a/drivers/compress/zlib/zlib_pmd.c
> +++ b/drivers/compress/zlib/zlib_pmd.c
> @@ -6,7 +6,198 @@
> #include <rte_common.h>
> #include "zlib_pmd_private.h"
>
> -/** Parse comp xform and set private xform/stream parameters */
> +/** Compute next mbuf in the list, assign data buffer and len */
> +#define COMPUTE_BUF(mbuf, data, len) \
> + ((mbuf = mbuf->next) ? \
> + (data = rte_pktmbuf_mtod(mbuf, uint8_t *)), \
> + (len = rte_pktmbuf_data_len(mbuf)) : 0)
> +
> +/** Set op->status to appropriate flag if we run out of mbuf */
> +#define COMPUTE_DST_BUF(mbuf, dst, dlen) \
> + ((COMPUTE_BUF(mbuf, dst, dlen)) ? 1 : \
> + (!(op->status = \
I see and build error here:
drivers/compress/zlib/zlib_pmd.c(81): error #187: use of "=" where "==" may
have been intended
COMPUTE_DST_BUF(mbuf_dst, strm->next_out,
It is a bit confusing macro, but afaik, you should pass op if you want to
modify the status
(I suggest making this more readable).
> + RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED)))
> +
> +static void
> +process_zlib_deflate(struct rte_comp_op *op, z_stream *strm) {
> + int ret, flush, fin_flush;
> + struct rte_mbuf *mbuf_src = op->m_src;
> + struct rte_mbuf *mbuf_dst = op->m_dst;
> +
> + switch (op->flush_flag) {
> + case RTE_COMP_FLUSH_FULL:
> + case RTE_COMP_FLUSH_FINAL:
> + fin_flush = Z_FINISH;
> + break;
> + default:
> + op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
> + ZLIB_PMD_ERR("\nInvalid flush value");
Better to have "\n" at the end for consistency.
> +
> + }
> +
> + if (unlikely(!strm)) {
> + op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
> + ZLIB_PMD_ERR("\nInvalid z_stream");
> + return;
> + }
> + /* Update z_stream with the inputs provided by application */
> + strm->next_in = rte_pktmbuf_mtod_offset(mbuf_src, uint8_t *,
> + op->src.offset);
> +
> + strm->avail_in = rte_pktmbuf_data_len(mbuf_src) - op->src.offset;
Shouldn't this be "op->src.length"?
> +
> + strm->next_out = rte_pktmbuf_mtod_offset(mbuf_dst, uint8_t *,
> + op->dst.offset);
> +
> + strm->avail_out = rte_pktmbuf_data_len(mbuf_dst) - op->dst.offset;
> +
> + /* Set flush value to NO_FLUSH unless it is last mbuf */
> + flush = Z_NO_FLUSH;
> + /* Initialize status to SUCCESS */
> + op->status = RTE_COMP_OP_STATUS_SUCCESS;
> +
...
> + /* Update source buffer to next mbuf
> + * Exit if op->status is not SUCCESS or input buffers are fully consumed
> + */
> + } while ((op->status == RTE_COMP_OP_STATUS_SUCCESS) &&
> + (COMPUTE_BUF(mbuf_src, strm->next_in, strm->avail_in)));
It looks like you support Scatter Gather here, but according to the
documentation, you don't support it...
> +
> +def_end:
> + /* Update op stats */
> + switch (op->status) {
> + case RTE_COMP_OP_STATUS_SUCCESS:
> + op->consumed += strm->total_in;
I see a compilation error with gcc:
drivers/compress/zlib/zlib_pmd.c:166:16: error:
this statement may fall through [-Werror=implicit-fallthrough=]
op->consumed += strm->total_in;
~~~~~~~~~~~~~^~~~~~~~~~~~~~~~~
drivers/compress/zlib/zlib_pmd.c:167:2: note: here
case RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED:
^~~~
If the intention is to fall-through, you should add a comment before the next
case.
> + case RTE_COMP_OP_STATUS_OUT_OF_SPACE_TERMINATED:
> + op->produced += strm->total_out;
> + break;
> + default:
> + ZLIB_PMD_ERR("stats not updated for status:%d\n",
> + op->status);
> + }
> +
> + deflateReset(strm);
> +}
> +
...
> +
> +/** Process comp operation for mbuf */
> +static inline int
> +process_zlib_op(struct zlib_qp *qp, struct rte_comp_op *op) {
> + struct zlib_stream *stream;
> +
> + if ((op->op_type == RTE_COMP_OP_STATEFUL) ||
> + op->src.offset > rte_pktmbuf_data_len(op->m_src) ||
> + op->dst.offset > rte_pktmbuf_data_len(op->m_dst)) {
An extra indentation is needed, so it is easy to distinguish between the if
statement and the body.
You should also check if (src.offset + src.length >
rte_pktmbuf_data_len(op->m_src))
> + op->status = RTE_COMP_OP_STATUS_INVALID_ARGS;
> + ZLIB_PMD_ERR("\nInvalid source or destination buffers or "
> + "invalid Operation requested");
Better to include the "\n" at the end.
> + } else {
> + stream = &((struct zlib_priv_xform *)op->private_xform)-
> >stream;
I think it is more readable to have this line split into two:
First get the private xform and then get zlib_stream pointer.
> + stream->comp(op, &stream->strm);
> + }
> + /* whatever is out of op, put it into completion queue with
> + * its status
> + */
> + return rte_ring_enqueue(qp->processed_pkts, (void *)op); }
> +
...
> +static uint16_t
> +zlib_pmd_enqueue_burst(void *queue_pair,
> + struct rte_comp_op **ops, uint16_t nb_ops) {
> + struct zlib_qp *qp = queue_pair;
> + int ret, i;
> + int enqd = 0;
"i" and "enqd" variables should be uint16_t.
> + for (i = 0; i < nb_ops; i++) {
> + ret = process_zlib_op(qp, ops[i]);
I think you should check if there is enough space in the ring for all the
operations, to save time.
If there is not, then the PMD would modify the operation unnecessarily and
waste a lot of time.
Then, with that check, you can just process the operations that can fit,
looping until minimum between
nb_ops and space available in the ring.
> + if (unlikely(ret < 0)) {
> + /* increment count if failed to push to completion
> + * queue
> + */
> + qp->qp_stats.enqueue_err_count++;
I think this variable should be used when there was an error in the operation
but
it was still be enqueued (because there was space in the ring).
So you can increase this count in process_zlib_op when there is an error.
> + } else {
> + qp->qp_stats.enqueued_count++;
I think this should be equal to all the operations enqueued (with and without
error).
> + enqd++;
> + }
> + }
> + return enqd;
> +}
