Re: [PATCH v6 1/2] eventdev/eth_tx: add queue start stop API

[Jerin Jacob <jerinjacobk@gmail.com>]

On Mon, Sep 26, 2022 at 7:34 AM Naga Harish K S V
<s.v.naga.harish.k@intel.com> wrote:
>
> Add support to start or stop a particular queue
> that is associated with the adapter.
>
> Start function enables the Tx adapter to start enqueueing
> packets to the Tx queue.
>
> Stop function stops the Tx adapter from enqueueing any
> packets to the Tx queue. The stop API also frees any packets
> that may have been buffered for this queue. All inflight packets
> destined to the queue are freed by the adapter runtime until the
> queue is started again.
>
> Signed-off-by: Naga Harish K S V <s.v.naga.harish.k@intel.com>

I will wait for @Jayatheerthan, Jay Ack to merge to RC1.

> ---
> v6:
> * fix nitpicks
> v5:
> * fix build failure
> v4:
> * update programmer guide and doxygen comments
> v3:
> * fix documentation and address review comments
> ---
> ---
>  .../prog_guide/event_ethernet_tx_adapter.rst  |  25 ++++
>  doc/guides/rel_notes/release_22_11.rst        |   8 ++
>  lib/eventdev/eventdev_pmd.h                   |  41 +++++++
>  lib/eventdev/rte_event_eth_tx_adapter.c       | 112 +++++++++++++++++-
>  lib/eventdev/rte_event_eth_tx_adapter.h       |  54 +++++++++
>  lib/eventdev/version.map                      |   2 +
>  6 files changed, 238 insertions(+), 4 deletions(-)
>
> diff --git a/doc/guides/prog_guide/event_ethernet_tx_adapter.rst b/doc/guides/prog_guide/event_ethernet_tx_adapter.rst
> index 1a5b069d60..73022e307a 100644
> --- a/doc/guides/prog_guide/event_ethernet_tx_adapter.rst
> +++ b/doc/guides/prog_guide/event_ethernet_tx_adapter.rst
> @@ -182,3 +182,28 @@ mbufs are destined to the same ethernet port and queue by setting the bit
>  ``rte_event_vector::queue``.
>  If ``rte_event_vector::attr_valid`` is not set then the Tx adapter should peek
>  into each mbuf and transmit them to the requested ethernet port and queue pair.
> +
> +Queue start/stop
> +~~~~~~~~~~~~~~~~
> +
> +The adapter can be configured to start/stop enqueueing of packets to a
> +associated NIC queue using ``rte_event_eth_tx_adapter_queue_start()`` or
> +``rte_event_eth_tx_adapter_queue_stop()`` respectively. By default the queue
> +is in start state.
> +
> +These APIs help avoid some unexpected behavior with application stopping ethdev
> +Tx queues and adapter being unaware of it. With these APIs, the application can
> +call stop API to notify adapter that corresponding ethdev Tx queue is stopped
> +and any in-flight packets are freed by adapter dataplane code. Adapter queue
> +stop API is called before stopping the ethdev Tx queue. When ethdev Tx queue
> +is enabled, application can notify adapter to resume processing of the packets
> +for that queue by calling the start API. The ethdev Tx queue is started before
> +calling adapter start API.
> +
> +Start function enables the adapter runtime to start enqueueing of packets
> +to the Tx queue.
> +
> +Stop function stops the adapter runtime function from enqueueing any
> +packets to the associated Tx queue. This API also frees any packets that
> +may have been buffered for this queue. All inflight packets destined to the
> +queue are freed by the adapter runtime until the queue is started again.
> diff --git a/doc/guides/rel_notes/release_22_11.rst b/doc/guides/rel_notes/release_22_11.rst
> index 97720a78ee..a4a51598a1 100644
> --- a/doc/guides/rel_notes/release_22_11.rst
> +++ b/doc/guides/rel_notes/release_22_11.rst
> @@ -31,6 +31,14 @@ New Features
>    Added ``rte_event_eth_tx_adapter_instance_get`` to get the Tx adapter instance id for specified
>    ethernet device id and Tx queue index.
>
> +
> +* **Added Tx adapter queue start/stop API**
> +
> +  Added ``rte_event_eth_tx_adapter_queue_start`` to start enqueueing packets to the Tx queue by
> +  Tx adapter.
> +  Added ``rte_event_eth_tx_adapter_queue_stop`` to stop the Tx Adapter from enqueueing any
> +  packets to the Tx queue.
> +
>  .. This section should contain new features added in this release.
>     Sample format:
>
> diff --git a/lib/eventdev/eventdev_pmd.h b/lib/eventdev/eventdev_pmd.h
> index 1e65d096f1..27b936a60e 100644
> --- a/lib/eventdev/eventdev_pmd.h
> +++ b/lib/eventdev/eventdev_pmd.h
> @@ -1277,6 +1277,43 @@ typedef int (*eventdev_eth_tx_adapter_stats_reset_t)(uint8_t id,
>  typedef int (*eventdev_eth_tx_adapter_instance_get_t)
>         (uint16_t eth_dev_id, uint16_t tx_queue_id, uint8_t *txa_inst_id);
>
> +/**
> + * Start a Tx queue that is assigned to Tx adapter instance
> + *
> + * @param id
> + *  Adapter identifier
> + *
> + * @param eth_dev_id
> + *  Port identifier of Ethernet device
> + *
> + * @param tx_queue_id
> + *  Ethernet device Tx queue index
> + *
> + * @return
> + *  -  0: Success
> + *  - <0: Error code on failure
> + */
> +typedef int (*eventdev_eth_tx_adapter_queue_start)
> +       (uint8_t id, uint16_t eth_dev_id, uint16_t tx_queue_id);
> +
> +/**
> + * Stop a Tx queue that is assigned to Tx adapter instance
> + *
> + * @param id
> + *  Adapter identifier
> + *
> + * @param eth_dev_id
> + *  Port identifier of Ethernet device
> + *
> + * @param tx_queue_id
> + *  Ethernet device Tx queue index
> + *
> + * @return
> + *  -  0: Success
> + *  - <0: Error code on failure
> + */
> +typedef int (*eventdev_eth_tx_adapter_queue_stop)
> +       (uint8_t id, uint16_t eth_dev_id, uint16_t tx_queue_id);
>
>  /** Event device operations function pointer table */
>  struct eventdev_ops {
> @@ -1390,6 +1427,10 @@ struct eventdev_ops {
>         /**< Reset eth Tx adapter statistics */
>         eventdev_eth_tx_adapter_instance_get_t eth_tx_adapter_instance_get;
>         /**< Get Tx adapter instance id for Tx queue */
> +       eventdev_eth_tx_adapter_queue_start eth_tx_adapter_queue_start;
> +       /**< Start Tx queue assigned to Tx adapter instance */
> +       eventdev_eth_tx_adapter_queue_stop eth_tx_adapter_queue_stop;
> +       /**< Stop Tx queue assigned to Tx adapter instance */
>
>         eventdev_selftest dev_selftest;
>         /**< Start eventdev Selftest */
> diff --git a/lib/eventdev/rte_event_eth_tx_adapter.c b/lib/eventdev/rte_event_eth_tx_adapter.c
> index 7e82fe030c..8f12762201 100644
> --- a/lib/eventdev/rte_event_eth_tx_adapter.c
> +++ b/lib/eventdev/rte_event_eth_tx_adapter.c
> @@ -47,6 +47,12 @@
>  #define txa_dev_instance_get(id) \
>                         txa_evdev(id)->dev_ops->eth_tx_adapter_instance_get
>
> +#define txa_dev_queue_start(id) \
> +                       txa_evdev(id)->dev_ops->eth_tx_adapter_queue_start
> +
> +#define txa_dev_queue_stop(id) \
> +                       txa_evdev(id)->dev_ops->eth_tx_adapter_queue_stop
> +
>  #define RTE_EVENT_ETH_TX_ADAPTER_ID_VALID_OR_ERR_RET(id, retval) \
>  do { \
>         if (!txa_valid_id(id)) { \
> @@ -94,6 +100,8 @@ struct txa_retry {
>  struct txa_service_queue_info {
>         /* Queue has been added */
>         uint8_t added;
> +       /* Queue is stopped */
> +       bool stopped;
>         /* Retry callback argument */
>         struct txa_retry txa_retry;
>         /* Tx buffer */
> @@ -556,7 +564,7 @@ txa_process_event_vector(struct txa_service_data *txa,
>                 port = vec->port;
>                 queue = vec->queue;
>                 tqi = txa_service_queue(txa, port, queue);
> -               if (unlikely(tqi == NULL || !tqi->added)) {
> +               if (unlikely(tqi == NULL || !tqi->added || tqi->stopped)) {
>                         rte_pktmbuf_free_bulk(mbufs, vec->nb_elem);
>                         rte_mempool_put(rte_mempool_from_obj(vec), vec);
>                         return 0;
> @@ -570,7 +578,8 @@ txa_process_event_vector(struct txa_service_data *txa,
>                         port = mbufs[i]->port;
>                         queue = rte_event_eth_tx_adapter_txq_get(mbufs[i]);
>                         tqi = txa_service_queue(txa, port, queue);
> -                       if (unlikely(tqi == NULL || !tqi->added)) {
> +                       if (unlikely(tqi == NULL || !tqi->added ||
> +                                    tqi->stopped)) {
>                                 rte_pktmbuf_free(mbufs[i]);
>                                 continue;
>                         }
> @@ -607,7 +616,8 @@ txa_service_tx(struct txa_service_data *txa, struct rte_event *ev,
>                         queue = rte_event_eth_tx_adapter_txq_get(m);
>
>                         tqi = txa_service_queue(txa, port, queue);
> -                       if (unlikely(tqi == NULL || !tqi->added)) {
> +                       if (unlikely(tqi == NULL || !tqi->added ||
> +                                    tqi->stopped)) {
>                                 rte_pktmbuf_free(m);
>                                 continue;
>                         }
> @@ -671,7 +681,8 @@ txa_service_func(void *args)
>                         for (q = 0; q < dev->data->nb_tx_queues; q++) {
>
>                                 tqi = txa_service_queue(txa, i, q);
> -                               if (unlikely(tqi == NULL || !tqi->added))
> +                               if (unlikely(tqi == NULL || !tqi->added ||
> +                                            tqi->stopped))
>                                         continue;
>
>                                 nb_tx += rte_eth_tx_buffer_flush(i, q,
> @@ -866,6 +877,7 @@ txa_service_queue_add(uint8_t id,
>
>         tqi->tx_buf = tb;
>         tqi->added = 1;
> +       tqi->stopped = false;
>         tdi->nb_queues++;
>         txa->nb_queues++;
>
> @@ -884,6 +896,20 @@ txa_service_queue_add(uint8_t id,
>         return -1;
>  }
>
> +static inline void
> +txa_txq_buffer_drain(struct txa_service_queue_info *tqi)
> +{
> +       struct rte_eth_dev_tx_buffer *b;
> +       uint16_t i;
> +
> +       b = tqi->tx_buf;
> +
> +       for (i = 0; i < b->length; i++)
> +               rte_pktmbuf_free(b->pkts[i]);
> +
> +       b->length = 0;
> +}
> +
>  static int
>  txa_service_queue_del(uint8_t id,
>                 const struct rte_eth_dev *dev,
> @@ -929,6 +955,8 @@ txa_service_queue_del(uint8_t id,
>         if (tqi == NULL || !tqi->added)
>                 goto ret_unlock;
>
> +       /* Drain the buffered mbufs */
> +       txa_txq_buffer_drain(tqi);
>         tb = tqi->tx_buf;
>         tqi->added = 0;
>         tqi->tx_buf = NULL;
> @@ -1319,3 +1347,79 @@ rte_event_eth_tx_adapter_instance_get(uint16_t eth_dev_id,
>
>         return -EINVAL;
>  }
> +
> +static inline int
> +txa_sw_queue_start_state_set(uint16_t eth_dev_id, uint16_t tx_queue_id,
> +                            bool start_state, struct txa_service_data *txa)
> +{
> +       struct txa_service_queue_info *tqi = NULL;
> +
> +       rte_spinlock_lock(&txa->tx_lock);
> +       tqi = txa_service_queue(txa, eth_dev_id, tx_queue_id);
> +       if (unlikely(tqi == NULL || !tqi->added)) {
> +               rte_spinlock_unlock(&txa->tx_lock);
> +               return -EINVAL;
> +       }
> +       if (start_state == false)
> +               txa_txq_buffer_drain(tqi);
> +
> +       tqi->stopped = !start_state;
> +       rte_spinlock_unlock(&txa->tx_lock);
> +       return 0;
> +}
> +
> +static int
> +txa_queue_start_state_set(uint16_t eth_dev_id, uint16_t tx_queue_id,
> +                         bool start_state)
> +{
> +       struct txa_service_data *txa;
> +       uint8_t txa_inst_id;
> +       int ret;
> +       uint32_t caps = 0;
> +
> +       /* Below API already does validation of input parameters.
> +        * Hence skipping the validation here.
> +        */
> +       ret = rte_event_eth_tx_adapter_instance_get(eth_dev_id,
> +                                                   tx_queue_id,
> +                                                   &txa_inst_id);
> +       if (ret < 0)
> +               return -EINVAL;
> +
> +       txa = txa_service_id_to_data(txa_inst_id);
> +       ret = rte_event_eth_tx_adapter_caps_get(txa->eventdev_id,
> +                                               eth_dev_id,
> +                                               &caps);
> +       if (ret < 0)
> +               return -EINVAL;
> +
> +       if (caps & RTE_EVENT_ETH_TX_ADAPTER_CAP_INTERNAL_PORT) {
> +               if (start_state == true) {
> +                       ret = txa_dev_queue_start(txa_inst_id) ?
> +                             txa_dev_queue_start(txa_inst_id)(txa_inst_id,
> +                                                              eth_dev_id,
> +                                                              tx_queue_id) : 0;
> +               } else {
> +                       ret = txa_dev_queue_stop(txa_inst_id) ?
> +                             txa_dev_queue_stop(txa_inst_id)(txa_inst_id,
> +                                                             eth_dev_id,
> +                                                             tx_queue_id) : 0;
> +               }
> +               return ret;
> +       }
> +
> +       return txa_sw_queue_start_state_set(eth_dev_id, tx_queue_id,
> +                                           start_state, txa);
> +}
> +
> +int
> +rte_event_eth_tx_adapter_queue_start(uint16_t eth_dev_id, uint16_t tx_queue_id)
> +{
> +       return txa_queue_start_state_set(eth_dev_id, tx_queue_id, true);
> +}
> +
> +int
> +rte_event_eth_tx_adapter_queue_stop(uint16_t eth_dev_id, uint16_t tx_queue_id)
> +{
> +       return txa_queue_start_state_set(eth_dev_id, tx_queue_id, false);
> +}
> diff --git a/lib/eventdev/rte_event_eth_tx_adapter.h b/lib/eventdev/rte_event_eth_tx_adapter.h
> index 9432b740e8..74ded7b651 100644
> --- a/lib/eventdev/rte_event_eth_tx_adapter.h
> +++ b/lib/eventdev/rte_event_eth_tx_adapter.h
> @@ -35,6 +35,8 @@
>   *  - rte_event_eth_tx_adapter_event_port_get()
>   *  - rte_event_eth_tx_adapter_service_id_get()
>   *  - rte_event_eth_tx_adapter_instance_get()
> + *  - rte_event_eth_tx_adapter_queue_start()
> + *  - rte_event_eth_tx_adapter_queue_stop()
>   *
>   * The application creates the adapter using
>   * rte_event_eth_tx_adapter_create() or rte_event_eth_tx_adapter_create_ext().
> @@ -446,6 +448,58 @@ int
>  rte_event_eth_tx_adapter_instance_get(uint16_t eth_dev_id,
>                                       uint16_t tx_queue_id,
>                                       uint8_t *txa_inst_id);
> +/**
> + * Enables the adapter to start enqueueing of packets to the
> + * Tx queue.
> + *
> + * This function is provided so that the application can
> + * resume enqueueing packets that reference packets for
> + * <eth_dev_id, tx_queue_id> after calling
> + * rte_event_eth_tx_adapter_queue_stop().
> + * @see rte_event_eth_tx_adapter_queue_stop
> + *
> + * Use case:
> + * --------
> + * The queue start/stop APIs help avoid some unexpected behavior with
> + * application stopping ethdev Tx queues and adapter being unaware of it.
> + * With these APIs, the application can call stop API to notify adapter
> + * that corresponding ethdev Tx queue is stopped and any in-flight
> + * packets are freed by adapter dataplane code. Adapter queue stop API
> + * is called before stopping the ethdev Tx queue. When ethdev Tx queue
> + * is enabled, application can notify adapter to resume processing of
> + * the packets for that queue by calling the start API. The ethdev Tx
> + * queue is started before calling adapter start API.
> + *
> + * @param eth_dev_id
> + *  Port identifier of Ethernet device.
> + * @param tx_queue_id
> + *  Ethernet device transmit queue index.
> + * @return
> + *   - 0: Success
> + *   - <0: Error code on failure
> + */
> +__rte_experimental
> +int
> +rte_event_eth_tx_adapter_queue_start(uint16_t eth_dev_id, uint16_t tx_queue_id);
> +
> +/**
> + * Stops the adapter runtime function from enqueueing any packets
> + * to the associated Tx queue. This API also frees any packets that may
> + * have been buffered for this queue. All inflight packets destined to the
> + * queue are freed by the adapter runtime until the queue is started again.
> + * @see rte_event_eth_tx_adapter_queue_start
> + *
> + * @param eth_dev_id
> + *  Port identifier of Ethernet device.
> + * @param tx_queue_id
> + *  Ethernet device transmit queue index.
> + * @return
> + *   - 0: Success
> + *   - <0: Error code on failure
> + */
> +__rte_experimental
> +int
> +rte_event_eth_tx_adapter_queue_stop(uint16_t eth_dev_id, uint16_t tx_queue_id);
>
>  #ifdef __cplusplus
>  }
> diff --git a/lib/eventdev/version.map b/lib/eventdev/version.map
> index 9a71cf3f8f..dd63ec6f68 100644
> --- a/lib/eventdev/version.map
> +++ b/lib/eventdev/version.map
> @@ -116,6 +116,8 @@ EXPERIMENTAL {
>         # added in 22.11
>         rte_event_eth_rx_adapter_instance_get;
>         rte_event_eth_tx_adapter_instance_get;
> +       rte_event_eth_tx_adapter_queue_start;
> +       rte_event_eth_tx_adapter_queue_stop;
>  };
>
>  INTERNAL {
> --
> 2.25.1
>