Re: [dpdk-dev] [PATCH v1 5/6] raw/octeontx2_ep: add dequeue operation

[Mahipal Challa <mchalla@marvell.com>]

Hi Gavin,
Please see response inline.

________________________________
From: Gavin Hu (Arm Technology China) <Gavin.Hu@arm.com>
Sent: Saturday, December 14, 2019 9:48 PM
To: Mahipal Challa <mchalla@marvell.com>; dev@dpdk.org <dev@dpdk.org>
Cc: Jerin Jacob Kollanukkaran <jerinj@marvell.com>; Narayana Prasad Raju Athreya <pathreya@marvell.com>; Subrahmanyam Nilla <snilla@marvell.com>; Venkateshwarlu Nalla <venkatn@marvell.com>; nd <nd@arm.com>
Subject: [EXT] RE: [dpdk-dev] [PATCH v1 5/6] raw/octeontx2_ep: add dequeue operation

External Email

----------------------------------------------------------------------
Hi Mahipal,

> -----Original Message-----
> From: dev <dev-bounces@dpdk.org> On Behalf Of Mahipal Challa
> Sent: Friday, December 6, 2019 2:39 PM
> To: dev@dpdk.org
> Cc: jerinj@marvell.com; pathreya@marvell.com; snilla@marvell.com;
> venkatn@marvell.com
> Subject: [dpdk-dev] [PATCH v1 5/6] raw/octeontx2_ep: add dequeue
> operation
>
> Add rawdev dequeue operation for SDP VF devices.
>
> Signed-off-by: Mahipal Challa <mchalla@marvell.com>
> ---
>  drivers/raw/octeontx2_ep/otx2_ep_enqdeq.c | 199
> ++++++++++++++++++++++++++++++
>  drivers/raw/octeontx2_ep/otx2_ep_enqdeq.h |   2 +
>  drivers/raw/octeontx2_ep/otx2_ep_rawdev.c |   1 +
>  drivers/raw/octeontx2_ep/otx2_ep_rawdev.h |  18 ++-
>  4 files changed, 219 insertions(+), 1 deletion(-)
>
> diff --git a/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.c
> b/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.c
> index ebbacfb..451fcc0 100644
> --- a/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.c
> +++ b/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.c
> @@ -260,6 +260,7 @@
>                rte_mempool_get(sdpvf->enqdeq_mpool, &buf);
>                if (buf == NULL) {
>                        otx2_err("OQ buffer alloc failed");
> +                     droq->stats.rx_alloc_failure++;
>                        /* sdp_droq_destroy_ring_buffers(droq);*/
>                        return -ENOMEM;
>                }
> @@ -645,3 +646,201 @@
>        return SDP_IQ_SEND_FAILED;
>  }
>
> +static uint32_t
> +sdp_droq_refill(struct sdp_device *sdpvf, struct sdp_droq *droq)
> +{
> +     struct sdp_droq_desc *desc_ring;
> +     uint32_t desc_refilled = 0;
> +     void *buf = NULL;
> +
> +     desc_ring = droq->desc_ring;
> +
> +     while (droq->refill_count && (desc_refilled < droq->nb_desc)) {
> +             /* If a valid buffer exists (happens if there is no dispatch),
> +              * reuse the buffer, else allocate.
> +              */
> +             if (droq->recv_buf_list[droq->refill_idx].buffer != NULL)
> +                     break;
> +
> +             rte_mempool_get(sdpvf->enqdeq_mpool, &buf);
> +             /* If a buffer could not be allocated, no point in
> +              * continuing
> +              */
> +             if (buf == NULL) {
> +                     droq->stats.rx_alloc_failure++;
> +                     break;
> +             }
> +
> +             droq->recv_buf_list[droq->refill_idx].buffer = buf;
> +             desc_ring[droq->refill_idx].buffer_ptr =
> rte_mem_virt2iova(buf);
> +
> +             /* Reset any previous values in the length field. */
> +             droq->info_list[droq->refill_idx].length = 0;
> +
> +             droq->refill_idx = sdp_incr_index(droq->refill_idx, 1,
> +                             droq->nb_desc);
> +
> +             desc_refilled++;
> +             droq->refill_count--;
> +
> +     }
> +
> +     return desc_refilled;
> +}
> +
> +static int
> +sdp_droq_read_packet(struct sdp_device *sdpvf __rte_unused,
> +                  struct sdp_droq *droq,
> +                  struct sdp_droq_pkt *droq_pkt)
> +{
> +     struct sdp_droq_info *info;
> +     uint32_t total_len = 0;
> +     uint32_t pkt_len = 0;
> +
> +     info = &droq->info_list[droq->read_idx];
> +     sdp_swap_8B_data((uint64_t *)&info->length, 1);
> +     if (!info->length) {
> +             otx2_err("OQ info_list->length[%ld]", (long)info->length);
> +             goto oq_read_fail;
> +     }
> +
> +     /* Deduce the actual data size */
> +     info->length -= SDP_RH_SIZE;
> +     total_len += (uint32_t)info->length;
> +
> +     otx2_sdp_dbg("OQ: pkt_len[%ld], buffer_size %d",
> +                     (long)info->length, droq->buffer_size);
> +     if (info->length > droq->buffer_size) {
> +             otx2_err("This mode is not supported: pkt_len > buffer_size");
> +             goto oq_read_fail;
> +     }
> +
> +     if (info->length <= droq->buffer_size) {
> +             pkt_len = (uint32_t)info->length;
> +             droq_pkt->data = droq->recv_buf_list[droq->read_idx].buffer;
> +             droq_pkt->len  = pkt_len;
> +
> +             droq->recv_buf_list[droq->read_idx].buffer = NULL;
> +             droq->read_idx = sdp_incr_index(droq->read_idx, 1,/*
> count */
> +                                             droq->nb_desc /* max rd idx
> */);
> +             droq->refill_count++;
> +
> +     }
> +
> +     info->length = 0;
> +
> +     return SDP_OQ_RECV_SUCCESS;
> +
> +oq_read_fail:
> +     return SDP_OQ_RECV_FAILED;
> +}
> +
> +static inline uint32_t
> +sdp_check_droq_pkts(struct sdp_droq *droq, uint32_t burst_size)
> +{
> +     uint32_t min_pkts = 0;
> +     uint32_t new_pkts;
> +     uint32_t pkt_count;
> +
> +     /* Latest available OQ packets */
> +     pkt_count = rte_read32(droq->pkts_sent_reg);
> +
> +     /* Newly arrived packets */
> +     new_pkts = pkt_count - droq->last_pkt_count;
> +     otx2_sdp_dbg("Recvd [%d] new OQ pkts", new_pkts);
> +
> +     min_pkts = (new_pkts > burst_size) ? burst_size : new_pkts;
> +     if (min_pkts) {
> +             rte_atomic64_add(&droq->pkts_pending, min_pkts);
> +             /* Back up the aggregated packet count so far */
> +             droq->last_pkt_count += min_pkts;
> +     }
> +
> +     return min_pkts;
> +}
> +
> +/* Check for response arrival from OCTEON TX2
> + * returns number of requests completed
> + */
> +int
> +sdp_rawdev_dequeue(struct rte_rawdev *rawdev,
> +                struct rte_rawdev_buf **buffers, unsigned int count,
> +                rte_rawdev_obj_t context __rte_unused)
> +{
> +     struct sdp_droq_pkt *oq_pkt;
> +     struct sdp_device *sdpvf;
> +     struct sdp_droq *droq;
> +
> +     uint32_t q_no = 0, pkts;
> +     uint32_t new_pkts;
> +     uint32_t ret;
> +
> +     sdpvf = (struct sdp_device *)rawdev->dev_private;
> +
> +     droq = sdpvf->droq[q_no];
> +     if (!droq) {
> +             otx2_err("Invalid droq[%d]", q_no);
> +             goto deq_fail;
> +     }
> +
> +     /* Grab the lock */
> +     rte_spinlock_lock(&droq->lock);
> +
> +     new_pkts = sdp_check_droq_pkts(droq, count);
> +     if (!new_pkts) {
> +             otx2_sdp_dbg("Zero new_pkts:%d", new_pkts);
> +             goto deq_fail; /* No pkts at this moment */
> +     }
> +
> +     otx2_sdp_dbg("Received new_pkts = %d", new_pkts);
> +
> +     for (pkts = 0; pkts < new_pkts; pkts++) {
> +
> +             /* Push the received pkt to application */
> +             oq_pkt = (struct sdp_droq_pkt *)buffers[pkts];
> +
> +             ret = sdp_droq_read_packet(sdpvf, droq, oq_pkt);
> +             if (ret) {
> +                     otx2_err("DROQ read pakt failed.");
> +                     goto deq_fail;
> +             }
> +
> +             /* Stats */
> +             droq->stats.pkts_received++;
> +             droq->stats.bytes_received += oq_pkt->len;
> +     }
> +
> +     /* Ack the h/w with no# of pkts read by Host */
> +     rte_wmb();
The following rte_write32 call has a built-in rte_io_wmb, which is sufficient for IO device.
The above barrier can be saved.
[Mahipal]: Agreed, will remove it.

> +     rte_write32(pkts, droq->pkts_sent_reg);
> +     rte_wmb();
Rte_cio_wmb is sufficient to keep order between coherent memory domain between lcore and I/O device.
https://urldefense.proofpoint.com/v2/url?u=https-3A__code.dpdk.org_dpdk_latest_source_lib_librte-5Feal_common_include_generic_rte-5Fatomic.h-23L137&d=DwIFAg&c=nKjWec2b6R0mOyPaz7xtfQ&r=-s3WtB2eBJ7B88AJuqlVnFqXScHMHYESWbIBMvf7lwI&m=XF-4neNZaxQUEDvG1kq1bRFnn4v3X44D4se5yqm58Js&s=DWWym_IWt8oB3S86IKjl70drjid0ZHFyCtVfKVwMrsI&e=
[Mahipal]: Okay, will replace it with rte_cio_wmb() API.
> +
> +     droq->last_pkt_count -= pkts;
> +
> +     otx2_sdp_dbg("DROQ pkts[%d] pushed to application", pkts);
> +
> +     /* Refill DROQ buffers */
> +     if (droq->refill_count >= 2 /* droq->refill_threshold */) {
> +             int desc_refilled = sdp_droq_refill(sdpvf, droq);
> +
> +             /* Flush the droq descriptor data to memory to be sure
> +              * that when we update the credits the data in memory is
> +              * accurate.
> +              */
> +             rte_wmb();
Rte_cio_wmb should be the correct one, but the stronger rte_io_wmb within rte_write32 API suffice.
[Mahipal]: Agreed, will remove it.
> +             rte_write32(desc_refilled, droq->pkts_credit_reg);
> +
> +             /* Ensure mmio write completes */
> +             rte_wmb();
If the target to ensure the completeness of the above writes, this is ok, otherwise it is overkill.
[Mahipal]: Yes, the target is to ensure the completion of all the writes, so it is required.
> +             otx2_sdp_dbg("Refilled count = %d", desc_refilled);
> +     }
> +
> +     /* Release the spin lock */
> +     rte_spinlock_unlock(&droq->lock);
> +
> +     return pkts;
> +
> +deq_fail:
> +     rte_spinlock_unlock(&droq->lock);
> +     return SDP_OQ_RECV_FAILED;
> +}
> diff --git a/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.h
> b/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.h
> index b9b7c0b..172fdc5 100644
> --- a/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.h
> +++ b/drivers/raw/octeontx2_ep/otx2_ep_enqdeq.h
> @@ -11,6 +11,8 @@
>  #define SDP_IQ_SEND_FAILED      (-1)
>  #define SDP_IQ_SEND_SUCCESS     (0)
>
> +#define SDP_OQ_RECV_FAILED      (-1)
> +#define SDP_OQ_RECV_SUCCESS     (0)
>
>  static inline uint64_t
>  sdp_endian_swap_8B(uint64_t _d)
> diff --git a/drivers/raw/octeontx2_ep/otx2_ep_rawdev.c
> b/drivers/raw/octeontx2_ep/otx2_ep_rawdev.c
> index 4ba8473..ddb208d 100644
> --- a/drivers/raw/octeontx2_ep/otx2_ep_rawdev.c
> +++ b/drivers/raw/octeontx2_ep/otx2_ep_rawdev.c
> @@ -252,6 +252,7 @@
>        .dev_stop       = sdp_rawdev_stop,
>        .dev_close      = sdp_rawdev_close,
>        .enqueue_bufs   = sdp_rawdev_enqueue,
> +     .dequeue_bufs   = sdp_rawdev_dequeue,
>  };
>
>  static int
> diff --git a/drivers/raw/octeontx2_ep/otx2_ep_rawdev.h
> b/drivers/raw/octeontx2_ep/otx2_ep_rawdev.h
> index 8fd06fb..a77cbab 100644
> --- a/drivers/raw/octeontx2_ep/otx2_ep_rawdev.h
> +++ b/drivers/raw/octeontx2_ep/otx2_ep_rawdev.h
> @@ -279,6 +279,18 @@ struct sdp_recv_buffer {
>  };
>  #define SDP_DROQ_RECVBUF_SIZE        (sizeof(struct sdp_recv_buffer))
>
> +/* DROQ statistics. Each output queue has four stats fields. */
> +struct sdp_droq_stats {
> +     /* Number of packets received in this queue. */
> +     uint64_t pkts_received;
> +
> +     /* Bytes received by this queue. */
> +     uint64_t bytes_received;
> +
> +     /* Num of failures of rte_pktmbuf_alloc() */
> +     uint64_t rx_alloc_failure;
> +};
> +
>  /* Structure to define the configuration attributes for each Output queue. */
>  struct sdp_oq_config {
>        /* Max number of OQs available */
> @@ -345,6 +357,9 @@ struct sdp_droq {
>         */
>        void *pkts_sent_reg;
>
> +     /* Statistics for this DROQ. */
> +     struct sdp_droq_stats stats;
> +
>        /* DMA mapped address of the DROQ descriptor ring. */
>        size_t desc_ring_dma;
>
> @@ -476,6 +491,7 @@ struct sdp_device {
>
>  int sdp_rawdev_enqueue(struct rte_rawdev *dev, struct rte_rawdev_buf
> **buffers,
>                       unsigned int count, rte_rawdev_obj_t context);
> -
> +int sdp_rawdev_dequeue(struct rte_rawdev *dev, struct rte_rawdev_buf
> **buffers,
> +                    unsigned int count, rte_rawdev_obj_t context);
>
>  #endif /* _OTX2_EP_RAWDEV_H_ */
> --
> 1.8.3.1