[PATCH] app/testpmd: skip stopped queues when forwarding

[PATCH] app/testpmd: skip stopped queues when forwarding

[Dmitry Kozlyuk]

After "port <port_id> rxq|txq <queue_id> stop"
the stopped queue was used in forwarding nonetheless,
which may cause undefined behavior in the PMD.

Record the configured queue state
and account for it when launching forwarding as follows:
+--------+---------+-----------------+---------------+
|RxQ     |TxQ      |Configured mode  |Launch routine |
+--------+---------+-----------------+---------------+
|stopped |stopped  |*                |-              |
|stopped |started  |txonly           |(configured)   |
|stopped |started  |*                |-              |
|started |stopped  |*                |rxonly         |
|started |started  |*                |(configured)   |
+--------+---------+-----------------+---------------+
Display stopped queues on "show port config rxtx".

Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
Cc: jing.d.chen@intel.com
Cc: stable@dpdk.org

Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
Reviewed-by: Raslan Darawsheh <rasland@nvidia.com>
---
 app/test-pmd/cmdline.c |  8 ++++++++
 app/test-pmd/config.c  |  6 ++++++
 app/test-pmd/testpmd.c | 18 ++++++++++++++++--
 app/test-pmd/testpmd.h | 10 ++++++++++
 4 files changed, 40 insertions(+), 2 deletions(-)

diff --git a/app/test-pmd/cmdline.c b/app/test-pmd/cmdline.c
index e626b1c7d9..8b0920e23d 100644
--- a/app/test-pmd/cmdline.c
+++ b/app/test-pmd/cmdline.c
@@ -2702,6 +2702,14 @@ cmd_config_rxtx_queue_parsed(void *parsed_result,
 
 	if (ret == -ENOTSUP)
 		fprintf(stderr, "Function not supported in PMD\n");
+	if (ret == 0) {
+		struct rte_port *port;
+		struct queue_state *states;
+
+		port = &ports[res->portid];
+		states = isrx ? port->rxq_state : port->txq_state;
+		states[res->qid].stopped = !isstart;
+	}
 }
 
 cmdline_parse_token_string_t cmd_config_rxtx_queue_port =
diff --git a/app/test-pmd/config.c b/app/test-pmd/config.c
index 1722d6c8f8..7ce9cb483a 100644
--- a/app/test-pmd/config.c
+++ b/app/test-pmd/config.c
@@ -2817,6 +2817,9 @@ rxtx_config_display(void)
 				       rx_conf->share_qid);
 			printf("\n");
 		}
+		for (qid = 0; qid < nb_rxq; qid++)
+			if (ports[pid].rxq_state[qid].stopped)
+				printf("    RX queue %d is stopped\n", qid);
 
 		/* per tx queue config only for first queue to be less verbose */
 		for (qid = 0; qid < 1; qid++) {
@@ -2850,6 +2853,9 @@ rxtx_config_display(void)
 			printf("      TX offloads=0x%"PRIx64" - TX RS bit threshold=%d\n",
 				offloads_tmp, tx_rs_thresh_tmp);
 		}
+		for (qid = 0; qid < nb_txq; qid++)
+			if (ports[pid].txq_state[qid].stopped)
+				printf("    TX queue %d is stopped\n", qid);
 	}
 }
 
diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
index 6c387bde84..36ff845181 100644
--- a/app/test-pmd/testpmd.c
+++ b/app/test-pmd/testpmd.c
@@ -2152,6 +2152,8 @@ flush_fwd_rx_queues(void)
 		for (rxp = 0; rxp < cur_fwd_config.nb_fwd_ports; rxp++) {
 			for (rxq = 0; rxq < nb_rxq; rxq++) {
 				port_id = fwd_ports_ids[rxp];
+				if (ports[port_id].rxq_state[rxq].stopped)
+					continue;
 				/**
 				* testpmd can stuck in the below do while loop
 				* if rte_eth_rx_burst() always returns nonzero
@@ -2223,8 +2225,20 @@ run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
 static int
 start_pkt_forward_on_core(void *fwd_arg)
 {
-	run_pkt_fwd_on_lcore((struct fwd_lcore *) fwd_arg,
-			     cur_fwd_config.fwd_eng->packet_fwd);
+	struct fwd_lcore *fc = fwd_arg;
+	struct fwd_stream *fsm = fwd_streams[fc->stream_idx];
+	struct queue_state *rxq = &ports[fsm->rx_port].rxq_state[fsm->rx_queue];
+	struct queue_state *txq = &ports[fsm->tx_port].txq_state[fsm->tx_queue];
+	struct fwd_engine *fwd_engine = cur_fwd_config.fwd_eng;
+	packet_fwd_t packet_fwd;
+
+	/* Check if there will ever be any packets to send. */
+	if (rxq->stopped && (txq->stopped || fwd_engine != &tx_only_engine))
+		return 0;
+	/* Force rxonly mode if RxQ is started, but TxQ is stopped. */
+	packet_fwd = !rxq->stopped && txq->stopped ? rx_only_engine.packet_fwd
+						   : fwd_engine->packet_fwd;
+	run_pkt_fwd_on_lcore(fc, packet_fwd);
 	return 0;
 }
 
diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
index 2149ecd93a..2744fa4d76 100644
--- a/app/test-pmd/testpmd.h
+++ b/app/test-pmd/testpmd.h
@@ -216,6 +216,12 @@ struct xstat_display_info {
 	bool	 allocated;
 };
 
+/** Application state of a queue. */
+struct queue_state {
+	/** The queue is stopped and should not be used. */
+	bool stopped;
+};
+
 /**
  * The data structure associated with each port.
  */
@@ -256,6 +262,10 @@ struct rte_port {
 	uint64_t		mbuf_dynf;
 	const struct rte_eth_rxtx_callback *tx_set_dynf_cb[RTE_MAX_QUEUES_PER_PORT+1];
 	struct xstat_display_info xstats_info;
+	/** Per-Rx-queue state. */
+	struct queue_state rxq_state[RTE_MAX_QUEUES_PER_PORT];
+	/** Per-Tx-queue state. */
+	struct queue_state txq_state[RTE_MAX_QUEUES_PER_PORT];
 };
 
 /**
-- 
2.25.1

RE: [PATCH] app/testpmd: skip stopped queues when forwarding

[Dmitry Kozlyuk]

> From: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> Sent: Thursday, January 13, 2022 12:21 PM

Hi Aman, Xiaoyun, Yuying,

Any comments on the proposed behavior or the code?

Re: [PATCH] app/testpmd: skip stopped queues when forwarding

[Singh, Aman Deep]

Hi Dmitry,

Thanks for the patch.

On 1/13/2022 2:51 PM, Dmitry Kozlyuk wrote:
> After "port <port_id> rxq|txq <queue_id> stop"
> the stopped queue was used in forwarding nonetheless,
> which may cause undefined behavior in the PMD.
>
> Record the configured queue state
> and account for it when launching forwarding as follows:
> +--------+---------+-----------------+---------------+
> |RxQ     |TxQ      |Configured mode  |Launch routine |
> +--------+---------+-----------------+---------------+
> |stopped |stopped  |*                |-              |
> |stopped |started  |txonly           |(configured)   |
> |stopped |started  |*                |-              |
> |started |stopped  |*                |rxonly         |
> |started |started  |*                |(configured)   |
> +--------+---------+-----------------+---------------+
> Display stopped queues on "show port config rxtx".
>
> Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
> Cc: jing.d.chen@intel.com
> Cc: stable@dpdk.org
>
> Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> Reviewed-by: Raslan Darawsheh <rasland@nvidia.com>
> ---
>   app/test-pmd/cmdline.c |  8 ++++++++
>   app/test-pmd/config.c  |  6 ++++++
>   app/test-pmd/testpmd.c | 18 ++++++++++++++++--
>   app/test-pmd/testpmd.h | 10 ++++++++++
>   4 files changed, 40 insertions(+), 2 deletions(-)
>
> diff --git a/app/test-pmd/cmdline.c b/app/test-pmd/cmdline.c
> index e626b1c7d9..8b0920e23d 100644
> --- a/app/test-pmd/cmdline.c
> +++ b/app/test-pmd/cmdline.c
> @@ -2702,6 +2702,14 @@ cmd_config_rxtx_queue_parsed(void *parsed_result,
>   
>   	if (ret == -ENOTSUP)
>   		fprintf(stderr, "Function not supported in PMD\n");
> +	if (ret == 0) {
> +		struct rte_port *port;
> +		struct queue_state *states;
> +
> +		port = &ports[res->portid];
> +		states = isrx ? port->rxq_state : port->txq_state;
> +		states[res->qid].stopped = !isstart;
> +	}
>   }
>   
>   cmdline_parse_token_string_t cmd_config_rxtx_queue_port =
> diff --git a/app/test-pmd/config.c b/app/test-pmd/config.c
> index 1722d6c8f8..7ce9cb483a 100644
> --- a/app/test-pmd/config.c
> +++ b/app/test-pmd/config.c
> @@ -2817,6 +2817,9 @@ rxtx_config_display(void)
>   				       rx_conf->share_qid);
>   			printf("\n");
>   		}
> +		for (qid = 0; qid < nb_rxq; qid++)
> +			if (ports[pid].rxq_state[qid].stopped)
> +				printf("    RX queue %d is stopped\n", qid);
>   
>   		/* per tx queue config only for first queue to be less verbose */
>   		for (qid = 0; qid < 1; qid++) {
> @@ -2850,6 +2853,9 @@ rxtx_config_display(void)
>   			printf("      TX offloads=0x%"PRIx64" - TX RS bit threshold=%d\n",
>   				offloads_tmp, tx_rs_thresh_tmp);
>   		}
> +		for (qid = 0; qid < nb_txq; qid++)
> +			if (ports[pid].txq_state[qid].stopped)
> +				printf("    TX queue %d is stopped\n", qid);
>   	}
>   }
>   
> diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
> index 6c387bde84..36ff845181 100644
> --- a/app/test-pmd/testpmd.c
> +++ b/app/test-pmd/testpmd.c
> @@ -2152,6 +2152,8 @@ flush_fwd_rx_queues(void)
>   		for (rxp = 0; rxp < cur_fwd_config.nb_fwd_ports; rxp++) {
>   			for (rxq = 0; rxq < nb_rxq; rxq++) {
>   				port_id = fwd_ports_ids[rxp];
> +				if (ports[port_id].rxq_state[rxq].stopped)
> +					continue;
>   				/**
>   				* testpmd can stuck in the below do while loop
>   				* if rte_eth_rx_burst() always returns nonzero
> @@ -2223,8 +2225,20 @@ run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
>   static int
>   start_pkt_forward_on_core(void *fwd_arg)
>   {
> -	run_pkt_fwd_on_lcore((struct fwd_lcore *) fwd_arg,
> -			     cur_fwd_config.fwd_eng->packet_fwd);
> +	struct fwd_lcore *fc = fwd_arg;
> +	struct fwd_stream *fsm = fwd_streams[fc->stream_idx];
> +	struct queue_state *rxq = &ports[fsm->rx_port].rxq_state[fsm->rx_queue];
> +	struct queue_state *txq = &ports[fsm->tx_port].txq_state[fsm->tx_queue];
> +	struct fwd_engine *fwd_engine = cur_fwd_config.fwd_eng;
> +	packet_fwd_t packet_fwd;
> +
> +	/* Check if there will ever be any packets to send. */
> +	if (rxq->stopped && (txq->stopped || fwd_engine != &tx_only_engine))
> +		return 0;
> +	/* Force rxonly mode if RxQ is started, but TxQ is stopped. */
> +	packet_fwd = !rxq->stopped && txq->stopped ? rx_only_engine.packet_fwd
> +						   : fwd_engine->packet_fwd;
Should we have a print here for user info, that mode has been changed or 
ignored.
> +	run_pkt_fwd_on_lcore(fc, packet_fwd);
>   	return 0;
>   }
>   
> diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
> index 2149ecd93a..2744fa4d76 100644
> --- a/app/test-pmd/testpmd.h
> +++ b/app/test-pmd/testpmd.h
> @@ -216,6 +216,12 @@ struct xstat_display_info {
>   	bool	 allocated;
>   };
>   
> +/** Application state of a queue. */
> +struct queue_state {
> +	/** The queue is stopped and should not be used. */
> +	bool stopped;
> +};
> +
>   /**
>    * The data structure associated with each port.
>    */
> @@ -256,6 +262,10 @@ struct rte_port {
>   	uint64_t		mbuf_dynf;
>   	const struct rte_eth_rxtx_callback *tx_set_dynf_cb[RTE_MAX_QUEUES_PER_PORT+1];
>   	struct xstat_display_info xstats_info;
> +	/** Per-Rx-queue state. */
> +	struct queue_state rxq_state[RTE_MAX_QUEUES_PER_PORT];
> +	/** Per-Tx-queue state. */
> +	struct queue_state txq_state[RTE_MAX_QUEUES_PER_PORT];
Can we think of adding rxq_state/txq_state as part of existing 
structures under rte_port->rte_eth_rxconf/rte_eth_txconf.
And if it helps, rather than bool can we use u8 with eth_dev defines-
#define RTE_ETH_QUEUE_STATE_STOPPED 0 /**< Queue stopped. */
#define RTE_ETH_QUEUE_STATE_STARTED 1 /**< Queue started. */
>   };
>   
>   /**

RE: [PATCH] app/testpmd: skip stopped queues when forwarding

[Zhang, Yuying]

Hi Dmitry,

> -----Original Message-----
> From: Singh, Aman Deep <aman.deep.singh@intel.com>
> Sent: Thursday, February 3, 2022 9:52 PM
> To: Dmitry Kozlyuk <dkozlyuk@nvidia.com>; dev@dpdk.org
> Cc: Li, Xiaoyun <xiaoyun.li@intel.com>; Zhang, Yuying
> <yuying.zhang@intel.com>; jing.d.chen@intel.com; stable@dpdk.org; Raslan
> Darawsheh <rasland@nvidia.com>
> Subject: Re: [PATCH] app/testpmd: skip stopped queues when forwarding
> 
> Hi Dmitry,
> 
> Thanks for the patch.
> 
> On 1/13/2022 2:51 PM, Dmitry Kozlyuk wrote:
> > After "port <port_id> rxq|txq <queue_id> stop"
> > the stopped queue was used in forwarding nonetheless, which may cause
> > undefined behavior in the PMD.
> >
> > Record the configured queue state
> > and account for it when launching forwarding as follows:
> > +--------+---------+-----------------+---------------+
> > |RxQ     |TxQ      |Configured mode  |Launch routine |
> > +--------+---------+-----------------+---------------+
> > |stopped |stopped  |*                |-              |
> > |stopped |started  |txonly           |(configured)   |
> > |stopped |started  |*                |-              |
> > |started |stopped  |*                |rxonly         |
> > |started |started  |*                |(configured)   |
> > +--------+---------+-----------------+---------------+
> > Display stopped queues on "show port config rxtx".
> >
> > Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
> > Cc: jing.d.chen@intel.com
> > Cc: stable@dpdk.org
> >
> > Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> > Reviewed-by: Raslan Darawsheh <rasland@nvidia.com>
> > ---
> >   app/test-pmd/cmdline.c |  8 ++++++++
> >   app/test-pmd/config.c  |  6 ++++++
> >   app/test-pmd/testpmd.c | 18 ++++++++++++++++--
> >   app/test-pmd/testpmd.h | 10 ++++++++++
> >   4 files changed, 40 insertions(+), 2 deletions(-)
> >
> > diff --git a/app/test-pmd/cmdline.c b/app/test-pmd/cmdline.c index
> > e626b1c7d9..8b0920e23d 100644
> > --- a/app/test-pmd/cmdline.c
> > +++ b/app/test-pmd/cmdline.c
> > @@ -2702,6 +2702,14 @@ cmd_config_rxtx_queue_parsed(void
> > *parsed_result,
> >
> >   	if (ret == -ENOTSUP)
> >   		fprintf(stderr, "Function not supported in PMD\n");
> > +	if (ret == 0) {
> > +		struct rte_port *port;
> > +		struct queue_state *states;
> > +
> > +		port = &ports[res->portid];
> > +		states = isrx ? port->rxq_state : port->txq_state;
> > +		states[res->qid].stopped = !isstart;
> > +	}
> >   }
> >
> >   cmdline_parse_token_string_t cmd_config_rxtx_queue_port = diff --git
> > a/app/test-pmd/config.c b/app/test-pmd/config.c index
> > 1722d6c8f8..7ce9cb483a 100644
> > --- a/app/test-pmd/config.c
> > +++ b/app/test-pmd/config.c
> > @@ -2817,6 +2817,9 @@ rxtx_config_display(void)
> >   				       rx_conf->share_qid);
> >   			printf("\n");
> >   		}
> > +		for (qid = 0; qid < nb_rxq; qid++)
> > +			if (ports[pid].rxq_state[qid].stopped)
> > +				printf("    RX queue %d is stopped\n", qid);
> >
> >   		/* per tx queue config only for first queue to be less verbose
> */
> >   		for (qid = 0; qid < 1; qid++) {
> > @@ -2850,6 +2853,9 @@ rxtx_config_display(void)
> >   			printf("      TX offloads=0x%"PRIx64" - TX RS bit
> threshold=%d\n",
> >   				offloads_tmp, tx_rs_thresh_tmp);
> >   		}
> > +		for (qid = 0; qid < nb_txq; qid++)
> > +			if (ports[pid].txq_state[qid].stopped)
> > +				printf("    TX queue %d is stopped\n", qid);
> >   	}
> >   }
> >
> > diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c index
> > 6c387bde84..36ff845181 100644
> > --- a/app/test-pmd/testpmd.c
> > +++ b/app/test-pmd/testpmd.c
> > @@ -2152,6 +2152,8 @@ flush_fwd_rx_queues(void)
> >   		for (rxp = 0; rxp < cur_fwd_config.nb_fwd_ports; rxp++) {
> >   			for (rxq = 0; rxq < nb_rxq; rxq++) {
> >   				port_id = fwd_ports_ids[rxp];
> > +				if (ports[port_id].rxq_state[rxq].stopped)
> > +					continue;
> >   				/**
> >   				* testpmd can stuck in the below do while
> loop
> >   				* if rte_eth_rx_burst() always returns
> nonzero @@ -2223,8
> > +2225,20 @@ run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t
> pkt_fwd)
> >   static int
> >   start_pkt_forward_on_core(void *fwd_arg)
> >   {
> > -	run_pkt_fwd_on_lcore((struct fwd_lcore *) fwd_arg,
> > -			     cur_fwd_config.fwd_eng->packet_fwd);
> > +	struct fwd_lcore *fc = fwd_arg;
> > +	struct fwd_stream *fsm = fwd_streams[fc->stream_idx];
> > +	struct queue_state *rxq = &ports[fsm->rx_port].rxq_state[fsm-
> >rx_queue];
> > +	struct queue_state *txq = &ports[fsm->tx_port].txq_state[fsm-
> >tx_queue];
> > +	struct fwd_engine *fwd_engine = cur_fwd_config.fwd_eng;
> > +	packet_fwd_t packet_fwd;
> > +
> > +	/* Check if there will ever be any packets to send. */
> > +	if (rxq->stopped && (txq->stopped || fwd_engine !=
> &tx_only_engine))
> > +		return 0;
Have you considered other fwd_engines such as io_fwd_engine and mac_fwd_engine?
> > +	/* Force rxonly mode if RxQ is started, but TxQ is stopped. */
> > +	packet_fwd = !rxq->stopped && txq->stopped ?
> rx_only_engine.packet_fwd
> > +						   : fwd_engine->packet_fwd;
> Should we have a print here for user info, that mode has been changed or
> ignored.
Why need to force rxonly mode for this situation? BTW, the value of cur_fwd_eng
hasn't been updated after you changed forward mode.
> > +	run_pkt_fwd_on_lcore(fc, packet_fwd);
> >   	return 0;
> >   }
> >
> > diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h index
> > 2149ecd93a..2744fa4d76 100644
> > --- a/app/test-pmd/testpmd.h
> > +++ b/app/test-pmd/testpmd.h
> > @@ -216,6 +216,12 @@ struct xstat_display_info {
> >   	bool	 allocated;
> >   };
> >
> > +/** Application state of a queue. */
> > +struct queue_state {
> > +	/** The queue is stopped and should not be used. */
> > +	bool stopped;
> > +};
> > +
> >   /**
> >    * The data structure associated with each port.
> >    */
> > @@ -256,6 +262,10 @@ struct rte_port {
> >   	uint64_t		mbuf_dynf;
> >   	const struct rte_eth_rxtx_callback
> *tx_set_dynf_cb[RTE_MAX_QUEUES_PER_PORT+1];
> >   	struct xstat_display_info xstats_info;
> > +	/** Per-Rx-queue state. */
> > +	struct queue_state rxq_state[RTE_MAX_QUEUES_PER_PORT];
> > +	/** Per-Tx-queue state. */
> > +	struct queue_state txq_state[RTE_MAX_QUEUES_PER_PORT];
> Can we think of adding rxq_state/txq_state as part of existing structures
> under rte_port->rte_eth_rxconf/rte_eth_txconf.
> And if it helps, rather than bool can we use u8 with eth_dev defines- #define
> RTE_ETH_QUEUE_STATE_STOPPED 0 /**< Queue stopped. */ #define
> RTE_ETH_QUEUE_STATE_STARTED 1 /**< Queue started. */
The same.
> >   };
> >
> >   /**

RE: [PATCH] app/testpmd: skip stopped queues when forwarding

[Dmitry Kozlyuk]

Hi Aman, Yuying,

You share some concerns, so I'm answering in one thread.

> From: Zhang, Yuying <yuying.zhang@intel.com>
> Sent: Wednesday, February 9, 2022 12:00 PM
[...]
> > > +2225,20 @@ run_pkt_fwd_on_lcore(struct fwd_lcore *fc,
> packet_fwd_t
> > pkt_fwd)
> > >   static int
> > >   start_pkt_forward_on_core(void *fwd_arg)
> > >   {
> > > -   run_pkt_fwd_on_lcore((struct fwd_lcore *) fwd_arg,
> > > -                        cur_fwd_config.fwd_eng->packet_fwd);
> > > +   struct fwd_lcore *fc = fwd_arg;
> > > +   struct fwd_stream *fsm = fwd_streams[fc->stream_idx];
> > > +   struct queue_state *rxq = &ports[fsm->rx_port].rxq_state[fsm-
> > >rx_queue];
> > > +   struct queue_state *txq = &ports[fsm->tx_port].txq_state[fsm-
> > >tx_queue];
> > > +   struct fwd_engine *fwd_engine = cur_fwd_config.fwd_eng;
> > > +   packet_fwd_t packet_fwd;
> > > +
> > > +   /* Check if there will ever be any packets to send. */
> > > +   if (rxq->stopped && (txq->stopped || fwd_engine !=
> > &tx_only_engine))
> > > +           return 0;
> Have you considered other fwd_engines such as io_fwd_engine and
> mac_fwd_engine?

The only engine that can send packets without receiving them is "txonly".
All other engines call rte_eth_rx_burst(),
which is illegal for a stopped RxQ even if the packets are discarded.

> > > +   /* Force rxonly mode if RxQ is started, but TxQ is stopped. */
> > > +   packet_fwd = !rxq->stopped && txq->stopped ?
> > rx_only_engine.packet_fwd
> > > +                                              : fwd_engine-
> >packet_fwd;
> > Should we have a print here for user info, that mode has been
> > changed or ignored.

Good idea.

> Why need to force rxonly mode for this situation? BTW, the value of
> cur_fwd_eng hasn't been updated after you changed forward mode.

It is logical to preserve as much workload as possible
so that stopping a TxQ does not reduce the Rx from NIC's perspective.

> > > +   run_pkt_fwd_on_lcore(fc, packet_fwd);
> > >     return 0;
> > >   }
> > >
> > > diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h index
> > > 2149ecd93a..2744fa4d76 100644
> > > --- a/app/test-pmd/testpmd.h
> > > +++ b/app/test-pmd/testpmd.h
> > > @@ -216,6 +216,12 @@ struct xstat_display_info {
> > >     bool     allocated;
> > >   };
> > >
> > > +/** Application state of a queue. */
> > > +struct queue_state {
> > > +   /** The queue is stopped and should not be used. */
> > > +   bool stopped;
> > > +};
> > > +
> > >   /**
> > >    * The data structure associated with each port.
> > >    */
> > > @@ -256,6 +262,10 @@ struct rte_port {
> > >     uint64_t                mbuf_dynf;
> > >     const struct rte_eth_rxtx_callback
> > *tx_set_dynf_cb[RTE_MAX_QUEUES_PER_PORT+1];
> > >     struct xstat_display_info xstats_info;
> > > +   /** Per-Rx-queue state. */
> > > +   struct queue_state rxq_state[RTE_MAX_QUEUES_PER_PORT];
> > > +   /** Per-Tx-queue state. */
> > > +   struct queue_state txq_state[RTE_MAX_QUEUES_PER_PORT];
> > Can we think of adding rxq_state/txq_state as part of existing
> structures
> > under rte_port->rte_eth_rxconf/rte_eth_txconf.
> > And if it helps, rather than bool can we use u8 with eth_dev
> defines- #define
> > RTE_ETH_QUEUE_STATE_STOPPED 0 /**< Queue stopped. */ #define
> > RTE_ETH_QUEUE_STATE_STARTED 1 /**< Queue started. */
> The same.

Will change to constants (even enum maybe?).

Struct rte_eth_{rx,tx}conf cannot be changed, because it's ethdev API,
and should not be changed, because it reflects queue configuration,
while the new arrays reflect the queue state.
What do you think of the following?

struct port_rxqueue {
	struct rte_eth_rxconf conf;
	uint8_t state; 
};

struct port_txqueue {
	struct rte_eth_txconf conf;
	uint8_t state;
};

struct rte_port {
	/* ... */
	struct port_rxqueue rxq[RTE_MAX_QUEUES_PER_PORT];
	struct port_txqueue txq[RTE_MAX_QUEUES_PER_PORT];
};

RE: [PATCH] app/testpmd: skip stopped queues when forwarding

[Li, Xiaoyun]

Hi Dmitry

Sorry to be direct, but I don't think this patch makes sense.
I need the code to clarify the problems so I'll use this thread not the one you're answering.

> -----Original Message-----
> From: Zhang, Yuying <yuying.zhang@intel.com>
> Sent: Wednesday, February 9, 2022 09:00
> To: Singh, Aman Deep <aman.deep.singh@intel.com>; Dmitry Kozlyuk
> <dkozlyuk@nvidia.com>; dev@dpdk.org
> Cc: Li, Xiaoyun <xiaoyun.li@intel.com>; jing.d.chen@intel.com;
> stable@dpdk.org; Raslan Darawsheh <rasland@nvidia.com>
> Subject: RE: [PATCH] app/testpmd: skip stopped queues when forwarding
> 
> Hi Dmitry,
> 
> > -----Original Message-----
> > From: Singh, Aman Deep <aman.deep.singh@intel.com>
> > Sent: Thursday, February 3, 2022 9:52 PM
> > To: Dmitry Kozlyuk <dkozlyuk@nvidia.com>; dev@dpdk.org
> > Cc: Li, Xiaoyun <xiaoyun.li@intel.com>; Zhang, Yuying
> > <yuying.zhang@intel.com>; jing.d.chen@intel.com; stable@dpdk.org;
> > Raslan Darawsheh <rasland@nvidia.com>
> > Subject: Re: [PATCH] app/testpmd: skip stopped queues when forwarding
> >
> > Hi Dmitry,
> >
> > Thanks for the patch.
> >
> > On 1/13/2022 2:51 PM, Dmitry Kozlyuk wrote:
> > > After "port <port_id> rxq|txq <queue_id> stop"
> > > the stopped queue was used in forwarding nonetheless, which may
> > > cause undefined behavior in the PMD.
> > >
> > > Record the configured queue state
> > > and account for it when launching forwarding as follows:
> > > +--------+---------+-----------------+---------------+
> > > |RxQ     |TxQ      |Configured mode  |Launch routine |
> > > +--------+---------+-----------------+---------------+
> > > |stopped |stopped  |*                |-              |
> > > |stopped |started  |txonly           |(configured)   |
> > > |stopped |started  |*                |-              |
> > > |started |stopped  |*                |rxonly         |
> > > |started |started  |*                |(configured)   |
> > > +--------+---------+-----------------+---------------+
> > > Display stopped queues on "show port config rxtx".
> > >
> > > Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
> > > Cc: jing.d.chen@intel.com
> > > Cc: stable@dpdk.org
> > >
> > > Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> > > Reviewed-by: Raslan Darawsheh <rasland@nvidia.com>
> > > ---
> > >   app/test-pmd/cmdline.c |  8 ++++++++
> > >   app/test-pmd/config.c  |  6 ++++++
> > >   app/test-pmd/testpmd.c | 18 ++++++++++++++++--
> > >   app/test-pmd/testpmd.h | 10 ++++++++++
> > >   4 files changed, 40 insertions(+), 2 deletions(-)
> > >
> > > diff --git a/app/test-pmd/cmdline.c b/app/test-pmd/cmdline.c index
> > > e626b1c7d9..8b0920e23d 100644
> > > --- a/app/test-pmd/cmdline.c
> > > +++ b/app/test-pmd/cmdline.c

<snip>

> > > +2225,20 @@ run_pkt_fwd_on_lcore(struct fwd_lcore *fc,
> packet_fwd_t
> > pkt_fwd)
> > >   static int
> > >   start_pkt_forward_on_core(void *fwd_arg)
> > >   {
> > > -	run_pkt_fwd_on_lcore((struct fwd_lcore *) fwd_arg,
> > > -			     cur_fwd_config.fwd_eng->packet_fwd);
> > > +	struct fwd_lcore *fc = fwd_arg;
> > > +	struct fwd_stream *fsm = fwd_streams[fc->stream_idx];
> > > +	struct queue_state *rxq = &ports[fsm->rx_port].rxq_state[fsm-
> > >rx_queue];
> > > +	struct queue_state *txq = &ports[fsm->tx_port].txq_state[fsm-
> > >tx_queue];
> > > +	struct fwd_engine *fwd_engine = cur_fwd_config.fwd_eng;
> > > +	packet_fwd_t packet_fwd;

1. For each lcore, there may be many fwd streams which are different queue pair even different port pair.
Just imagine the simplest case, you run iofwd with 4 ports single queue with single forwarding core.
But you're only dealing with the first fwd stream here. Each fc actually includes stream from fc->stream_idx to fc->stream_idx+fc->stream_nb-1.

What are you going to do with this?
If you stop port 0 rxq 0, you shouldn't change the whole lcore behavior to rxonly. Because the pair of port 2 rxq 0-> port 3 txq 0 and port 3 rxq0 -> port 2 txq0 should be normal iofwd.
And this is only the basic case.
And I don't think it's that harmful to just let those fwd running even if you stop some rxq. Because their pair won't have pkts to do tx if there's no rx.

2. Even if you're going to narrow down for the case which fc->nb_stream=1.
You keep the value of cur_fwd_config. There's a potential issue here since cur_fwd_config is not really cur_fwd_config anymore.
In stop_packet_forwarding(), It sets fc->stopped=1 and uses cur_fwd_config.fwd_eng->port_fwd_end to stop packeting forward.
Right now, in most of the fwd mode, port_fwd_end=NULL. But txonly and noisy_vnf and icmp_echo are not.
What if someday rxonly engine needs to have a port_fwd_end? Should you call port_fwd_end for your lcore (rxonly) too while others are still cur_fwd_engine?

Also, in this case, shouldn't you init "stopped" (or "state" following other comments) to be 1 for txonly? Because the default value will be 0, following your code, txonly engine will change to rxonly if the txq is stopped.

Also, even if you init "stopped" for txonly, have you considered flow_gen engine? Flow_gen is also kind of a txonly engine (all received pkts will be dropped) to generate multi-flow packets.


Anyway, I still think it's not worth it. And there may be other issues too. The patch looks dangerous to me.

BRs
Xiaoyun

> > > +
> > > +	/* Check if there will ever be any packets to send. */
> > > +	if (rxq->stopped && (txq->stopped || fwd_engine !=
> > &tx_only_engine))
> > > +		return 0;
> Have you considered other fwd_engines such as io_fwd_engine and
> mac_fwd_engine?
> > > +	/* Force rxonly mode if RxQ is started, but TxQ is stopped. */
> > > +	packet_fwd = !rxq->stopped && txq->stopped ?
> > rx_only_engine.packet_fwd
> > > +						   : fwd_engine->packet_fwd;
> > Should we have a print here for user info, that mode has been changed
> > or ignored.
> Why need to force rxonly mode for this situation? BTW, the value of
> cur_fwd_eng hasn't been updated after you changed forward mode.
> > > +	run_pkt_fwd_on_lcore(fc, packet_fwd);
> > >   	return 0;
> > >   }
> > >
> > > diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h index
> > > 2149ecd93a..2744fa4d76 100644
> > > --- a/app/test-pmd/testpmd.h
> > > +++ b/app/test-pmd/testpmd.h
> > > @@ -216,6 +216,12 @@ struct xstat_display_info {
> > >   	bool	 allocated;
> > >   };
> > >
> > > +/** Application state of a queue. */ struct queue_state {
> > > +	/** The queue is stopped and should not be used. */
> > > +	bool stopped;
> > > +};
> > > +
> > >   /**
> > >    * The data structure associated with each port.
> > >    */
> > > @@ -256,6 +262,10 @@ struct rte_port {
> > >   	uint64_t		mbuf_dynf;
> > >   	const struct rte_eth_rxtx_callback
> > *tx_set_dynf_cb[RTE_MAX_QUEUES_PER_PORT+1];
> > >   	struct xstat_display_info xstats_info;
> > > +	/** Per-Rx-queue state. */
> > > +	struct queue_state rxq_state[RTE_MAX_QUEUES_PER_PORT];
> > > +	/** Per-Tx-queue state. */
> > > +	struct queue_state txq_state[RTE_MAX_QUEUES_PER_PORT];
> > Can we think of adding rxq_state/txq_state as part of existing
> > structures under rte_port->rte_eth_rxconf/rte_eth_txconf.
> > And if it helps, rather than bool can we use u8 with eth_dev defines-
> > #define RTE_ETH_QUEUE_STATE_STOPPED 0 /**< Queue stopped. */
> #define
> > RTE_ETH_QUEUE_STATE_STARTED 1 /**< Queue started. */
> The same.
> > >   };
> > >
> > >   /**

RE: [PATCH] app/testpmd: skip stopped queues when forwarding

[Dmitry Kozlyuk]

- jing.d.chen@intel.com (invalid address?), +ethdev maintainers

> From: Li, Xiaoyun <xiaoyun.li@intel.com>
> [...] 
> Sorry to be direct, but I don't think this patch makes sense.
> I need the code to clarify the problems so I'll use this thread not
> the one you're answering.

Thank you for explaining the issues in great detail!
I'd like to understand one crucial point below
to decide whether testpmd needs any changes at all.

> [...]
> And I don't think it's that harmful to just let those fwd running even
> if you stop some rxq. Because their pair won't have pkts to do tx if
> there's no rx.

This patch was created with the assumption
that stopped queues may not be used for Rx/Tx.
No-op behavior of rte_eth_rx/tx_burst()
for a stopped queue is not documented.
Yes, at least some PMDs implement it this way.
But is this behavior intended?

It is the application that stops the queue or starts it deferred.
Stopping is non-atomic, so polling the queue is not allowed during it.
Hence, if the application intends to stop queues, it must either:

a) Know the queue is not polled before stopping it.
   Use case: a secondary that was polling the queue has crashed,
   the primary is doing a recovery to free all mbufs.
   There is no issue since there is no poller to touch the queue.

b) Tell the poller to skip the queue for the time of stopping it.
   Use case: deferred queue start or queue reconfiguration.
   But if the poller is cooperating anyway,
   what prevents it from not touching the queue for longer?

The same considerations apply to starting a queue.

No-op behavior is convenient from the application point of view.
But it also means that pollers of stopped queues
will go all the way down to PMD Rx/Tx routines, wasting cycles,
and some PMDs will do a check for the queue state,
even though it may never be needed for a particular application.

RE: [PATCH] app/testpmd: skip stopped queues when forwarding

[Dmitry Kozlyuk]

Andrew, Ferruh, Thomas, which behavior does ethdev assume (see below)?

> This patch was created with the assumption
> that stopped queues may not be used for Rx/Tx.
> No-op behavior of rte_eth_rx/tx_burst()
> for a stopped queue is not documented.
> Yes, at least some PMDs implement it this way.
> But is this behavior intended?
> 
> It is the application that stops the queue or starts it deferred.
> Stopping is non-atomic, so polling the queue is not allowed during it.
> Hence, if the application intends to stop queues, it must either:
> 
> a) Know the queue is not polled before stopping it.
>    Use case: a secondary that was polling the queue has crashed,
>    the primary is doing a recovery to free all mbufs.
>    There is no issue since there is no poller to touch the queue.
> 
> b) Tell the poller to skip the queue for the time of stopping it.
>    Use case: deferred queue start or queue reconfiguration.
>    But if the poller is cooperating anyway,
>    what prevents it from not touching the queue for longer?
> 
> The same considerations apply to starting a queue.
> 
> No-op behavior is convenient from the application point of view.
> But it also means that pollers of stopped queues
> will go all the way down to PMD Rx/Tx routines, wasting cycles,
> and some PMDs will do a check for the queue state,
> even though it may never be needed for a particular application.

Re: [PATCH] app/testpmd: skip stopped queues when forwarding

[Thomas Monjalon]

21/02/2022 09:58, Dmitry Kozlyuk:
> Andrew, Ferruh, Thomas, which behavior does ethdev assume (see below)?

For the whole device stop, this is the documentation:
"
  The transmit and receive functions should not be invoked
  when the device is stopped.
"

There is also this comment on rte_eth_dev_reset:
"
  Note: To avoid unexpected behavior, the application should stop calling
  Tx and Rx functions before calling rte_eth_dev_reset( ). For thread
  safety, all these controlling functions should be called from the same
  thread.
"

For queue stop, there is no documented expectation.

There is this comment for queue callback removal:
"
  The memory for the callback can be
  subsequently freed back by the application by calling rte_free():
 
  - Immediately - if the port is stopped, or the user knows that no
    callbacks are in flight e.g. if called from the thread doing Rx/Tx
    on that queue.
 
  - After a short delay - where the delay is sufficient to allow any
    in-flight callbacks to complete. Alternately, the RCU mechanism can be
    used to detect when data plane threads have ceased referencing the
    callback memory.
"

> > This patch was created with the assumption
> > that stopped queues may not be used for Rx/Tx.
> > No-op behavior of rte_eth_rx/tx_burst()
> > for a stopped queue is not documented.

Indeed, it is not documented.
I suggest working on this documentation first.
testpmd could be adjusted later if needed.

> > Yes, at least some PMDs implement it this way.
> > But is this behavior intended?
> > 
> > It is the application that stops the queue or starts it deferred.
> > Stopping is non-atomic, so polling the queue is not allowed during it.
> > Hence, if the application intends to stop queues, it must either:
> > 
> > a) Know the queue is not polled before stopping it.
> >    Use case: a secondary that was polling the queue has crashed,
> >    the primary is doing a recovery to free all mbufs.
> >    There is no issue since there is no poller to touch the queue.
> > 
> > b) Tell the poller to skip the queue for the time of stopping it.
> >    Use case: deferred queue start or queue reconfiguration.
> >    But if the poller is cooperating anyway,
> >    what prevents it from not touching the queue for longer?
> > 
> > The same considerations apply to starting a queue.
> > 
> > No-op behavior is convenient from the application point of view.
> > But it also means that pollers of stopped queues
> > will go all the way down to PMD Rx/Tx routines, wasting cycles,
> > and some PMDs will do a check for the queue state,
> > even though it may never be needed for a particular application.

Yes that's the question: Do we want
1/ to allow apps to poll stopped queues, adding checks in PMDs,
or do we prefer
2/ saving check cycles and expect apps to not poll?

[PATCH v2 0/2] app/testpmd: skip stopped queues when forwarding

[Dmitry Kozlyuk]

It was unspecified what happens of a stopped queue is polled.
Declare that polling is prohibited and modify testpmd to follow this.
Rationale is described i nthe commit log; it follows the discussion:
http://inbox.dpdk.org/dev/BL1PR12MB594551A71273709E1C04A8BAB9309@BL1PR12MB5945.namprd12.prod.outlook.com/

v2:
  * Do not try to change forwarding mode partially,
    disable individual flows that need stopped queues instead.
  * Do not store queue status in testpmd,
    query it as needed instead.

Dmitry Kozlyuk (2):
  ethdev: prohibit polling of a stopped queue
  app/testpmd: do not poll stopped queues

 app/test-pmd/5tswap.c         | 13 ++++++++
 app/test-pmd/csumonly.c       | 13 ++++++++
 app/test-pmd/flowgen.c        | 13 ++++++++
 app/test-pmd/icmpecho.c       | 13 ++++++++
 app/test-pmd/ieee1588fwd.c    | 13 ++++++++
 app/test-pmd/iofwd.c          | 13 ++++++++
 app/test-pmd/macfwd.c         | 13 ++++++++
 app/test-pmd/noisy_vnf.c      | 13 ++++++++
 app/test-pmd/rxonly.c         | 13 ++++++++
 app/test-pmd/shared_rxq_fwd.c | 13 ++++++++
 app/test-pmd/testpmd.c        | 57 ++++++++++++++++++++++++++++++++++-
 app/test-pmd/testpmd.h        |  4 +++
 app/test-pmd/txonly.c         | 13 ++++++++
 lib/ethdev/rte_ethdev.h       |  2 +-
 14 files changed, 204 insertions(+), 2 deletions(-)

-- 
2.25.1

[PATCH v2 1/2] ethdev: prohibit polling of a stopped queue

[Dmitry Kozlyuk]

Whether it is allowed to call Rx/Tx functions for a stopped queue
was undocumented. Some PMDs make this behavior a no-op
either by explicitly checking the queue state
or by the way how their routines are implemented or HW works.

No-op behavior may be convenient for application developers.
But it also means that pollers of stopped queues
would go all the way down to PMD Rx/Tx routines, wasting cycles.
Some PMDs would do a check for the queue state on data path,
even though it may never be needed for a particular application.
Also, use cases for stopping queues or starting them deferred
do not logically require polling stopped queues.

Use case 1: a secondary that was polling the queue has crashed,
the primary is doing a recovery to free all mbufs.
By definition the queue to be restarted is not polled.

Use case 2: deferred queue start or queue reconfiguration.
The polling thread must be synchronized anyway,
because queue start and stop are non-atomic.

Prohibit calling Rx/Tx functions on stopped queues.

Fixes: 0748be2cf9a2 ("ethdev: queue start and stop")
Cc: stable@dpdk.org

Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
Acked-by: Matan Azrad <matan@nvidia.com>
---
 lib/ethdev/rte_ethdev.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/ethdev/rte_ethdev.h b/lib/ethdev/rte_ethdev.h
index c2d1f9a972..9f12a6043c 100644
--- a/lib/ethdev/rte_ethdev.h
+++ b/lib/ethdev/rte_ethdev.h
@@ -74,7 +74,7 @@
  * rte_eth_rx_queue_setup()), it must call rte_eth_dev_stop() first to stop the
  * device and then do the reconfiguration before calling rte_eth_dev_start()
  * again. The transmit and receive functions should not be invoked when the
- * device is stopped.
+ * device is stopped or when the queue is stopped (for that queue).
  *
  * Please note that some configuration is not stored between calls to
  * rte_eth_dev_stop()/rte_eth_dev_start(). The following configuration will
-- 
2.25.1

[PATCH v2 2/2] app/testpmd: do not poll stopped queues

[Dmitry Kozlyuk]

Calling Rx/Tx functions on a stopped queue is not supported.
Do not run packet forwarding for streams that use stopped queues.

Each stream has a read-only "disabled" field,
so that lcore function can skip such streams.
Forwarding engines can set this field
using a new "stream_init" callback function
by checking relevant queue states.
A helper function is provided to check if a given
Rx queue, Tx queue, or both of them are stopped.

Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
Cc: stable@dpdk.org

Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
Acked-by: Matan Azrad <matan@nvidia.com>
---
 app/test-pmd/5tswap.c         | 13 ++++++++
 app/test-pmd/csumonly.c       | 13 ++++++++
 app/test-pmd/flowgen.c        | 13 ++++++++
 app/test-pmd/icmpecho.c       | 13 ++++++++
 app/test-pmd/ieee1588fwd.c    | 13 ++++++++
 app/test-pmd/iofwd.c          | 13 ++++++++
 app/test-pmd/macfwd.c         | 13 ++++++++
 app/test-pmd/noisy_vnf.c      | 13 ++++++++
 app/test-pmd/rxonly.c         | 13 ++++++++
 app/test-pmd/shared_rxq_fwd.c | 13 ++++++++
 app/test-pmd/testpmd.c        | 57 ++++++++++++++++++++++++++++++++++-
 app/test-pmd/testpmd.h        |  4 +++
 app/test-pmd/txonly.c         | 13 ++++++++
 13 files changed, 203 insertions(+), 1 deletion(-)

diff --git a/app/test-pmd/5tswap.c b/app/test-pmd/5tswap.c
index 629d3e0d31..2aa1f1843b 100644
--- a/app/test-pmd/5tswap.c
+++ b/app/test-pmd/5tswap.c
@@ -185,9 +185,22 @@ pkt_burst_5tuple_swap(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static int
+stream_init_5tuple_swap(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine five_tuple_swap_fwd_engine = {
 	.fwd_mode_name  = "5tswap",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_5tuple_swap,
 	.packet_fwd     = pkt_burst_5tuple_swap,
 };
diff --git a/app/test-pmd/csumonly.c b/app/test-pmd/csumonly.c
index 5274d498ee..a031cae2ca 100644
--- a/app/test-pmd/csumonly.c
+++ b/app/test-pmd/csumonly.c
@@ -1178,9 +1178,22 @@ pkt_burst_checksum_forward(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static int
+stream_init_checksum_forward(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine csum_fwd_engine = {
 	.fwd_mode_name  = "csum",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_checksum_forward,
 	.packet_fwd     = pkt_burst_checksum_forward,
 };
diff --git a/app/test-pmd/flowgen.c b/app/test-pmd/flowgen.c
index 9ceef3b54a..e2c1bfd82c 100644
--- a/app/test-pmd/flowgen.c
+++ b/app/test-pmd/flowgen.c
@@ -207,9 +207,22 @@ flowgen_begin(portid_t pi)
 	return 0;
 }
 
+static int
+flowgen_stream_init(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine flow_gen_engine = {
 	.fwd_mode_name  = "flowgen",
 	.port_fwd_begin = flowgen_begin,
 	.port_fwd_end   = NULL,
+	.stream_init    = flowgen_stream_init,
 	.packet_fwd     = pkt_burst_flow_gen,
 };
diff --git a/app/test-pmd/icmpecho.c b/app/test-pmd/icmpecho.c
index 99c94cb282..dd3699ff3b 100644
--- a/app/test-pmd/icmpecho.c
+++ b/app/test-pmd/icmpecho.c
@@ -512,9 +512,22 @@ reply_to_icmp_echo_rqsts(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static int
+icmpecho_stream_init(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine icmp_echo_engine = {
 	.fwd_mode_name  = "icmpecho",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = icmpecho_stream_init,
 	.packet_fwd     = reply_to_icmp_echo_rqsts,
 };
diff --git a/app/test-pmd/ieee1588fwd.c b/app/test-pmd/ieee1588fwd.c
index 9ff817aa68..f9f73f2c14 100644
--- a/app/test-pmd/ieee1588fwd.c
+++ b/app/test-pmd/ieee1588fwd.c
@@ -211,9 +211,22 @@ port_ieee1588_fwd_end(portid_t pi)
 	rte_eth_timesync_disable(pi);
 }
 
+static int
+port_ieee1588_stream_init(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine ieee1588_fwd_engine = {
 	.fwd_mode_name  = "ieee1588",
 	.port_fwd_begin = port_ieee1588_fwd_begin,
 	.port_fwd_end   = port_ieee1588_fwd_end,
+	.stream_init    = port_ieee1588_stream_init,
 	.packet_fwd     = ieee1588_packet_fwd,
 };
diff --git a/app/test-pmd/iofwd.c b/app/test-pmd/iofwd.c
index 19cd920f70..b736a2a3bc 100644
--- a/app/test-pmd/iofwd.c
+++ b/app/test-pmd/iofwd.c
@@ -88,9 +88,22 @@ pkt_burst_io_forward(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static int
+stream_init_forward(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine io_fwd_engine = {
 	.fwd_mode_name  = "io",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_forward,
 	.packet_fwd     = pkt_burst_io_forward,
 };
diff --git a/app/test-pmd/macfwd.c b/app/test-pmd/macfwd.c
index 812a0c721f..64b65c8c51 100644
--- a/app/test-pmd/macfwd.c
+++ b/app/test-pmd/macfwd.c
@@ -119,9 +119,22 @@ pkt_burst_mac_forward(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static int
+stream_init_mac_forward(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine mac_fwd_engine = {
 	.fwd_mode_name  = "mac",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_mac_forward,
 	.packet_fwd     = pkt_burst_mac_forward,
 };
diff --git a/app/test-pmd/noisy_vnf.c b/app/test-pmd/noisy_vnf.c
index e4434bea95..58f53212a4 100644
--- a/app/test-pmd/noisy_vnf.c
+++ b/app/test-pmd/noisy_vnf.c
@@ -277,9 +277,22 @@ noisy_fwd_begin(portid_t pi)
 	return 0;
 }
 
+static int
+stream_init_noisy_vnf(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = rx_stopped || tx_stopped;
+	return ret;
+}
+
 struct fwd_engine noisy_vnf_engine = {
 	.fwd_mode_name  = "noisy",
 	.port_fwd_begin = noisy_fwd_begin,
 	.port_fwd_end   = noisy_fwd_end,
+	.stream_init    = stream_init_noisy_vnf,
 	.packet_fwd     = pkt_burst_noisy_vnf,
 };
diff --git a/app/test-pmd/rxonly.c b/app/test-pmd/rxonly.c
index d1a579d8d8..945ea2d27a 100644
--- a/app/test-pmd/rxonly.c
+++ b/app/test-pmd/rxonly.c
@@ -68,9 +68,22 @@ pkt_burst_receive(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static int
+stream_init_receive(struct fwd_stream *fs)
+{
+	bool rx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, NULL);
+	if (ret == 0)
+		fs->disabled = rx_stopped;
+	return ret;
+}
+
 struct fwd_engine rx_only_engine = {
 	.fwd_mode_name  = "rxonly",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_receive,
 	.packet_fwd     = pkt_burst_receive,
 };
diff --git a/app/test-pmd/shared_rxq_fwd.c b/app/test-pmd/shared_rxq_fwd.c
index da54a383fd..9389df2627 100644
--- a/app/test-pmd/shared_rxq_fwd.c
+++ b/app/test-pmd/shared_rxq_fwd.c
@@ -107,9 +107,22 @@ shared_rxq_fwd(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static int
+shared_rxq_stream_init(struct fwd_stream *fs)
+{
+	bool rx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, &rx_stopped, NULL);
+	if (ret == 0)
+		fs->disabled = rx_stopped;
+	return ret;
+}
+
 struct fwd_engine shared_rxq_engine = {
 	.fwd_mode_name  = "shared_rxq",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = shared_rxq_stream_init,
 	.packet_fwd     = shared_rxq_fwd,
 };
diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
index fe2ce19f99..b3e360121a 100644
--- a/app/test-pmd/testpmd.c
+++ b/app/test-pmd/testpmd.c
@@ -1763,6 +1763,37 @@ reconfig(portid_t new_port_id, unsigned socket_id)
 	init_port_config();
 }
 
+int
+fwd_stream_get_stopped_queues(struct fwd_stream *fs, bool *rx, bool *tx)
+{
+	struct rte_eth_rxq_info rx_qinfo;
+	struct rte_eth_txq_info tx_qinfo;
+	int ret;
+
+	if (rx != NULL) {
+		ret = rte_eth_rx_queue_info_get(fs->rx_port, fs->rx_queue,
+						&rx_qinfo);
+		if (ret < 0) {
+			RTE_LOG(ERR, USER1, "Cannot get port %d RX queue %d info: %s\n",
+				fs->rx_port, fs->rx_queue,
+				rte_strerror(rte_errno));
+			return ret;
+		}
+		*rx = rx_qinfo.queue_state == RTE_ETH_QUEUE_STATE_STOPPED;
+	}
+	if (tx != NULL) {
+		ret = rte_eth_tx_queue_info_get(fs->tx_port, fs->tx_queue,
+						&tx_qinfo);
+		if (ret < 0) {
+			TESTPMD_LOG(ERR, "Cannot get port %d TX queue %d info: %s\n",
+				    fs->tx_port, fs->tx_queue,
+				    rte_strerror(rte_errno));
+			return ret;
+		}
+		*tx = tx_qinfo.queue_state == RTE_ETH_QUEUE_STATE_STOPPED;
+	}
+	return 0;
+}
 
 int
 init_fwd_streams(void)
@@ -2155,6 +2186,21 @@ flush_fwd_rx_queues(void)
 	for (j = 0; j < 2; j++) {
 		for (rxp = 0; rxp < cur_fwd_config.nb_fwd_ports; rxp++) {
 			for (rxq = 0; rxq < nb_rxq; rxq++) {
+				struct rte_eth_rxq_info rx_qinfo;
+				int ret;
+
+				ret = rte_eth_rx_queue_info_get(rxp, rxq,
+								&rx_qinfo);
+				if (ret < 0) {
+					TESTPMD_LOG(ERR, "Cannot get port %d RX queue %d info: %s\n",
+						    rxp, rxq,
+						    rte_strerror(rte_errno));
+					return;
+				}
+				if (rx_qinfo.queue_state ==
+				    RTE_ETH_QUEUE_STATE_STOPPED)
+					continue;
+
 				port_id = fwd_ports_ids[rxp];
 				/**
 				* testpmd can stuck in the below do while loop
@@ -2201,7 +2247,8 @@ run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
 	nb_fs = fc->stream_nb;
 	do {
 		for (sm_id = 0; sm_id < nb_fs; sm_id++)
-			(*pkt_fwd)(fsm[sm_id]);
+			if (!fsm[sm_id]->disabled)
+				(*pkt_fwd)(fsm[sm_id]);
 #ifdef RTE_LIB_BITRATESTATS
 		if (bitrate_enabled != 0 &&
 				bitrate_lcore_id == rte_lcore_id()) {
@@ -2283,6 +2330,7 @@ start_packet_forwarding(int with_tx_first)
 {
 	port_fwd_begin_t port_fwd_begin;
 	port_fwd_end_t  port_fwd_end;
+	stream_init_t stream_init = cur_fwd_eng->stream_init;
 	unsigned int i;
 
 	if (strcmp(cur_fwd_eng->fwd_mode_name, "rxonly") == 0 && !nb_rxq)
@@ -2313,6 +2361,13 @@ start_packet_forwarding(int with_tx_first)
 	if (!pkt_fwd_shared_rxq_check())
 		return;
 
+	if (stream_init != NULL)
+		for (i = 0; i < cur_fwd_config.nb_fwd_streams; i++)
+			if (stream_init(fwd_streams[i]) < 0) {
+				TESTPMD_LOG(ERR, "Cannot init stream\n");
+				return;
+			}
+
 	port_fwd_begin = cur_fwd_config.fwd_eng->port_fwd_begin;
 	if (port_fwd_begin != NULL) {
 		for (i = 0; i < cur_fwd_config.nb_fwd_ports; i++) {
diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
index 31f766c965..59edae645e 100644
--- a/app/test-pmd/testpmd.h
+++ b/app/test-pmd/testpmd.h
@@ -134,6 +134,7 @@ struct fwd_stream {
 	portid_t   tx_port;   /**< forwarding port of received packets */
 	queueid_t  tx_queue;  /**< TX queue to send forwarded packets */
 	streamid_t peer_addr; /**< index of peer ethernet address of packets */
+	bool       disabled;  /**< the stream is disabled and should not run */
 
 	unsigned int retry_enabled;
 
@@ -323,12 +324,14 @@ struct fwd_lcore {
  */
 typedef int (*port_fwd_begin_t)(portid_t pi);
 typedef void (*port_fwd_end_t)(portid_t pi);
+typedef int (*stream_init_t)(struct fwd_stream *fs);
 typedef void (*packet_fwd_t)(struct fwd_stream *fs);
 
 struct fwd_engine {
 	const char       *fwd_mode_name; /**< Forwarding mode name. */
 	port_fwd_begin_t port_fwd_begin; /**< NULL if nothing special to do. */
 	port_fwd_end_t   port_fwd_end;   /**< NULL if nothing special to do. */
+	stream_init_t    stream_init;    /**< NULL if nothing special to do. */
 	packet_fwd_t     packet_fwd;     /**< Mandatory. */
 };
 
@@ -887,6 +890,7 @@ void rxtx_config_display(void);
 void fwd_config_setup(void);
 void set_def_fwd_config(void);
 void reconfig(portid_t new_port_id, unsigned socket_id);
+int fwd_stream_get_stopped_queues(struct fwd_stream *fs, bool *rx, bool *tx);
 int init_fwd_streams(void);
 void update_fwd_ports(portid_t new_pid);
 
diff --git a/app/test-pmd/txonly.c b/app/test-pmd/txonly.c
index fc039a622c..1fa5238896 100644
--- a/app/test-pmd/txonly.c
+++ b/app/test-pmd/txonly.c
@@ -504,9 +504,22 @@ tx_only_begin(portid_t pi)
 	return 0;
 }
 
+static int
+tx_only_stream_init(struct fwd_stream *fs)
+{
+	bool tx_stopped;
+	int ret;
+
+	ret = fwd_stream_get_stopped_queues(fs, NULL, &tx_stopped);
+	if (ret == 0)
+		fs->disabled = tx_stopped;
+	return ret;
+}
+
 struct fwd_engine tx_only_engine = {
 	.fwd_mode_name  = "txonly",
 	.port_fwd_begin = tx_only_begin,
 	.port_fwd_end   = NULL,
+	.stream_init    = tx_only_stream_init,
 	.packet_fwd     = pkt_burst_transmit,
 };
-- 
2.25.1

[PATCH v3 0/2] app/testpmd: skip stopped queues when forwarding

[Dmitry Kozlyuk]

It was unspecified what happens of a stopped queue is polled.
Declare that polling is prohibited and modify testpmd to follow this.
Rationale is described in the commit log; it follows the discussion:
http://inbox.dpdk.org/dev/BL1PR12MB594551A71273709E1C04A8BAB9309@BL1PR12MB5945.namprd12.prod.outlook.com/

v3:
  * Revert to storing queue status in testpmd to handle PMDs
    that don't implement rte_eth_rx/tx_queue_info_get()
    (Daxue Gao).
  * Add missed "macswap" support.

v2:
  * Do not try to change forwarding mode partially,
    disable individual flows that need stopped queues instead.
  * Do not store queue status in testpmd,
    query it as needed instead.

Dmitry Kozlyuk (2):
  app/testpmd: do not poll stopped queues
  ethdev: prohibit polling of a stopped queue

 app/test-pmd/5tswap.c         | 13 +++++
 app/test-pmd/cmdline.c        | 45 +++++++++-------
 app/test-pmd/config.c         |  8 +--
 app/test-pmd/csumonly.c       | 13 +++++
 app/test-pmd/flowgen.c        | 13 +++++
 app/test-pmd/icmpecho.c       | 13 +++++
 app/test-pmd/ieee1588fwd.c    | 13 +++++
 app/test-pmd/iofwd.c          | 13 +++++
 app/test-pmd/macfwd.c         | 13 +++++
 app/test-pmd/macswap.c        | 13 +++++
 app/test-pmd/noisy_vnf.c      | 13 +++++
 app/test-pmd/rxonly.c         |  8 +++
 app/test-pmd/shared_rxq_fwd.c |  8 +++
 app/test-pmd/testpmd.c        | 96 +++++++++++++++++++++++------------
 app/test-pmd/testpmd.h        | 19 ++++++-
 app/test-pmd/txonly.c         |  8 +++
 lib/ethdev/rte_ethdev.h       |  2 +-
 17 files changed, 254 insertions(+), 57 deletions(-)

-- 
2.25.1

[PATCH v3 1/2] app/testpmd: do not poll stopped queues

[Dmitry Kozlyuk]

Calling Rx/Tx functions on a stopped queue is not supported.
Do not run packet forwarding for streams that use stopped queues.

Each stream has a read-only "disabled" field,
so that lcore function can skip such streams.
Forwarding engines can set this field
using a new "stream_init" callback function
by checking relevant queue states,
which are stored along with queue configurations
(not all PMDs implement rte_eth_rx/tx_queue_info_get()
to query the state from there).

Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
Cc: stable@dpdk.org

Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
Acked-by: Matan Azrad <matan@nvidia.com>
---
 app/test-pmd/5tswap.c         | 13 ++++++
 app/test-pmd/cmdline.c        | 45 ++++++++++--------
 app/test-pmd/config.c         |  8 ++--
 app/test-pmd/csumonly.c       | 13 ++++++
 app/test-pmd/flowgen.c        | 13 ++++++
 app/test-pmd/icmpecho.c       | 13 ++++++
 app/test-pmd/ieee1588fwd.c    | 13 ++++++
 app/test-pmd/iofwd.c          | 13 ++++++
 app/test-pmd/macfwd.c         | 13 ++++++
 app/test-pmd/macswap.c        | 13 ++++++
 app/test-pmd/noisy_vnf.c      | 13 ++++++
 app/test-pmd/rxonly.c         |  8 ++++
 app/test-pmd/shared_rxq_fwd.c |  8 ++++
 app/test-pmd/testpmd.c        | 87 ++++++++++++++++++++++-------------
 app/test-pmd/testpmd.h        | 19 +++++++-
 app/test-pmd/txonly.c         |  8 ++++
 16 files changed, 244 insertions(+), 56 deletions(-)

diff --git a/app/test-pmd/5tswap.c b/app/test-pmd/5tswap.c
index 629d3e0d31..f041a5e1d5 100644
--- a/app/test-pmd/5tswap.c
+++ b/app/test-pmd/5tswap.c
@@ -185,9 +185,22 @@ pkt_burst_5tuple_swap(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+stream_init_5tuple_swap(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine five_tuple_swap_fwd_engine = {
 	.fwd_mode_name  = "5tswap",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_5tuple_swap,
 	.packet_fwd     = pkt_burst_5tuple_swap,
 };
diff --git a/app/test-pmd/cmdline.c b/app/test-pmd/cmdline.c
index 7ab0575e64..2ca935f086 100644
--- a/app/test-pmd/cmdline.c
+++ b/app/test-pmd/cmdline.c
@@ -2658,8 +2658,10 @@ cmd_config_rxtx_queue_parsed(void *parsed_result,
 			__rte_unused void *data)
 {
 	struct cmd_config_rxtx_queue *res = parsed_result;
+	struct rte_port *port;
 	uint8_t isrx;
 	uint8_t isstart;
+	uint8_t *state;
 	int ret = 0;
 
 	if (test_done == 0) {
@@ -2707,8 +2709,15 @@ cmd_config_rxtx_queue_parsed(void *parsed_result,
 	else
 		ret = rte_eth_dev_tx_queue_stop(res->portid, res->qid);
 
-	if (ret == -ENOTSUP)
+	if (ret == -ENOTSUP) {
 		fprintf(stderr, "Function not supported in PMD\n");
+		return;
+	}
+
+	port = &ports[res->portid];
+	state = isrx ? &port->rxq[res->qid].state : &port->txq[res->qid].state;
+	*state = isstart ? RTE_ETH_QUEUE_STATE_STARTED :
+			   RTE_ETH_QUEUE_STATE_STOPPED;
 }
 
 cmdline_parse_token_string_t cmd_config_rxtx_queue_port =
@@ -2777,11 +2786,11 @@ cmd_config_deferred_start_rxtx_queue_parsed(void *parsed_result,
 
 	ison = !strcmp(res->state, "on");
 
-	if (isrx && port->rx_conf[res->qid].rx_deferred_start != ison) {
-		port->rx_conf[res->qid].rx_deferred_start = ison;
+	if (isrx && port->rxq[res->qid].conf.rx_deferred_start != ison) {
+		port->rxq[res->qid].conf.rx_deferred_start = ison;
 		needreconfig = 1;
-	} else if (!isrx && port->tx_conf[res->qid].tx_deferred_start != ison) {
-		port->tx_conf[res->qid].tx_deferred_start = ison;
+	} else if (!isrx && port->txq[res->qid].conf.tx_deferred_start != ison) {
+		port->txq[res->qid].conf.tx_deferred_start = ison;
 		needreconfig = 1;
 	}
 
@@ -2899,7 +2908,7 @@ cmd_setup_rxtx_queue_parsed(
 				     res->qid,
 				     port->nb_rx_desc[res->qid],
 				     socket_id,
-				     &port->rx_conf[res->qid],
+				     &port->rxq[res->qid].conf,
 				     mp);
 		if (ret)
 			fprintf(stderr, "Failed to setup RX queue\n");
@@ -2917,7 +2926,7 @@ cmd_setup_rxtx_queue_parsed(
 					     res->qid,
 					     port->nb_tx_desc[res->qid],
 					     socket_id,
-					     &port->tx_conf[res->qid]);
+					     &port->txq[res->qid].conf);
 		if (ret)
 			fprintf(stderr, "Failed to setup TX queue\n");
 	}
@@ -4693,7 +4702,7 @@ cmd_config_queue_tx_offloads(struct rte_port *port)
 
 	/* Apply queue tx offloads configuration */
 	for (k = 0; k < port->dev_info.max_tx_queues; k++)
-		port->tx_conf[k].offloads =
+		port->txq[k].conf.offloads =
 			port->dev_conf.txmode.offloads;
 }
 
@@ -16204,7 +16213,7 @@ cmd_rx_offload_get_configuration_parsed(
 
 	nb_rx_queues = dev_info.nb_rx_queues;
 	for (q = 0; q < nb_rx_queues; q++) {
-		queue_offloads = port->rx_conf[q].offloads;
+		queue_offloads = port->rxq[q].conf.offloads;
 		printf("  Queue[%2d] :", q);
 		print_rx_offloads(queue_offloads);
 		printf("\n");
@@ -16324,11 +16333,11 @@ cmd_config_per_port_rx_offload_parsed(void *parsed_result,
 	if (!strcmp(res->on_off, "on")) {
 		port->dev_conf.rxmode.offloads |= single_offload;
 		for (q = 0; q < nb_rx_queues; q++)
-			port->rx_conf[q].offloads |= single_offload;
+			port->rxq[q].conf.offloads |= single_offload;
 	} else {
 		port->dev_conf.rxmode.offloads &= ~single_offload;
 		for (q = 0; q < nb_rx_queues; q++)
-			port->rx_conf[q].offloads &= ~single_offload;
+			port->rxq[q].conf.offloads &= ~single_offload;
 	}
 
 	cmd_reconfig_device_queue(port_id, 1, 1);
@@ -16434,9 +16443,9 @@ cmd_config_per_queue_rx_offload_parsed(void *parsed_result,
 	}
 
 	if (!strcmp(res->on_off, "on"))
-		port->rx_conf[queue_id].offloads |= single_offload;
+		port->rxq[queue_id].conf.offloads |= single_offload;
 	else
-		port->rx_conf[queue_id].offloads &= ~single_offload;
+		port->rxq[queue_id].conf.offloads &= ~single_offload;
 
 	cmd_reconfig_device_queue(port_id, 1, 1);
 }
@@ -16623,7 +16632,7 @@ cmd_tx_offload_get_configuration_parsed(
 
 	nb_tx_queues = dev_info.nb_tx_queues;
 	for (q = 0; q < nb_tx_queues; q++) {
-		queue_offloads = port->tx_conf[q].offloads;
+		queue_offloads = port->txq[q].conf.offloads;
 		printf("  Queue[%2d] :", q);
 		print_tx_offloads(queue_offloads);
 		printf("\n");
@@ -16747,11 +16756,11 @@ cmd_config_per_port_tx_offload_parsed(void *parsed_result,
 	if (!strcmp(res->on_off, "on")) {
 		port->dev_conf.txmode.offloads |= single_offload;
 		for (q = 0; q < nb_tx_queues; q++)
-			port->tx_conf[q].offloads |= single_offload;
+			port->txq[q].conf.offloads |= single_offload;
 	} else {
 		port->dev_conf.txmode.offloads &= ~single_offload;
 		for (q = 0; q < nb_tx_queues; q++)
-			port->tx_conf[q].offloads &= ~single_offload;
+			port->txq[q].conf.offloads &= ~single_offload;
 	}
 
 	cmd_reconfig_device_queue(port_id, 1, 1);
@@ -16860,9 +16869,9 @@ cmd_config_per_queue_tx_offload_parsed(void *parsed_result,
 	}
 
 	if (!strcmp(res->on_off, "on"))
-		port->tx_conf[queue_id].offloads |= single_offload;
+		port->txq[queue_id].conf.offloads |= single_offload;
 	else
-		port->tx_conf[queue_id].offloads &= ~single_offload;
+		port->txq[queue_id].conf.offloads &= ~single_offload;
 
 	cmd_reconfig_device_queue(port_id, 1, 1);
 }
diff --git a/app/test-pmd/config.c b/app/test-pmd/config.c
index cc8e7aa138..c4ab3f8cae 100644
--- a/app/test-pmd/config.c
+++ b/app/test-pmd/config.c
@@ -3551,8 +3551,8 @@ rxtx_config_display(void)
 	       nb_fwd_lcores, nb_fwd_ports);
 
 	RTE_ETH_FOREACH_DEV(pid) {
-		struct rte_eth_rxconf *rx_conf = &ports[pid].rx_conf[0];
-		struct rte_eth_txconf *tx_conf = &ports[pid].tx_conf[0];
+		struct rte_eth_rxconf *rx_conf = &ports[pid].rxq[0].conf;
+		struct rte_eth_txconf *tx_conf = &ports[pid].txq[0].conf;
 		uint16_t *nb_rx_desc = &ports[pid].nb_rx_desc[0];
 		uint16_t *nb_tx_desc = &ports[pid].nb_tx_desc[0];
 		struct rte_eth_rxq_info rx_qinfo;
@@ -3810,7 +3810,7 @@ fwd_stream_on_other_lcores(uint16_t domain_id, lcoreid_t src_lc,
 			fs = fwd_streams[sm_id];
 			port = &ports[fs->rx_port];
 			dev_info = &port->dev_info;
-			rxq_conf = &port->rx_conf[fs->rx_queue];
+			rxq_conf = &port->rxq[fs->rx_queue].conf;
 			if ((dev_info->dev_capa & RTE_ETH_DEV_CAPA_RXQ_SHARE)
 			    == 0 || rxq_conf->share_group == 0)
 				/* Not shared rxq. */
@@ -3870,7 +3870,7 @@ pkt_fwd_shared_rxq_check(void)
 			fs->lcore = fwd_lcores[lc_id];
 			port = &ports[fs->rx_port];
 			dev_info = &port->dev_info;
-			rxq_conf = &port->rx_conf[fs->rx_queue];
+			rxq_conf = &port->rxq[fs->rx_queue].conf;
 			if ((dev_info->dev_capa & RTE_ETH_DEV_CAPA_RXQ_SHARE)
 			    == 0 || rxq_conf->share_group == 0)
 				/* Not shared rxq. */
diff --git a/app/test-pmd/csumonly.c b/app/test-pmd/csumonly.c
index 5274d498ee..eb58ca1906 100644
--- a/app/test-pmd/csumonly.c
+++ b/app/test-pmd/csumonly.c
@@ -1178,9 +1178,22 @@ pkt_burst_checksum_forward(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+stream_init_checksum_forward(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine csum_fwd_engine = {
 	.fwd_mode_name  = "csum",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_checksum_forward,
 	.packet_fwd     = pkt_burst_checksum_forward,
 };
diff --git a/app/test-pmd/flowgen.c b/app/test-pmd/flowgen.c
index 9ceef3b54a..1e01120ae9 100644
--- a/app/test-pmd/flowgen.c
+++ b/app/test-pmd/flowgen.c
@@ -207,9 +207,22 @@ flowgen_begin(portid_t pi)
 	return 0;
 }
 
+static void
+flowgen_stream_init(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine flow_gen_engine = {
 	.fwd_mode_name  = "flowgen",
 	.port_fwd_begin = flowgen_begin,
 	.port_fwd_end   = NULL,
+	.stream_init    = flowgen_stream_init,
 	.packet_fwd     = pkt_burst_flow_gen,
 };
diff --git a/app/test-pmd/icmpecho.c b/app/test-pmd/icmpecho.c
index 99c94cb282..066f2a3ab7 100644
--- a/app/test-pmd/icmpecho.c
+++ b/app/test-pmd/icmpecho.c
@@ -512,9 +512,22 @@ reply_to_icmp_echo_rqsts(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+icmpecho_stream_init(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine icmp_echo_engine = {
 	.fwd_mode_name  = "icmpecho",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = icmpecho_stream_init,
 	.packet_fwd     = reply_to_icmp_echo_rqsts,
 };
diff --git a/app/test-pmd/ieee1588fwd.c b/app/test-pmd/ieee1588fwd.c
index 9ff817aa68..fc4e2d014c 100644
--- a/app/test-pmd/ieee1588fwd.c
+++ b/app/test-pmd/ieee1588fwd.c
@@ -211,9 +211,22 @@ port_ieee1588_fwd_end(portid_t pi)
 	rte_eth_timesync_disable(pi);
 }
 
+static void
+port_ieee1588_stream_init(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine ieee1588_fwd_engine = {
 	.fwd_mode_name  = "ieee1588",
 	.port_fwd_begin = port_ieee1588_fwd_begin,
 	.port_fwd_end   = port_ieee1588_fwd_end,
+	.stream_init    = port_ieee1588_stream_init,
 	.packet_fwd     = ieee1588_packet_fwd,
 };
diff --git a/app/test-pmd/iofwd.c b/app/test-pmd/iofwd.c
index 19cd920f70..71849aaf96 100644
--- a/app/test-pmd/iofwd.c
+++ b/app/test-pmd/iofwd.c
@@ -88,9 +88,22 @@ pkt_burst_io_forward(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+stream_init_forward(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine io_fwd_engine = {
 	.fwd_mode_name  = "io",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_forward,
 	.packet_fwd     = pkt_burst_io_forward,
 };
diff --git a/app/test-pmd/macfwd.c b/app/test-pmd/macfwd.c
index 812a0c721f..79c9241d00 100644
--- a/app/test-pmd/macfwd.c
+++ b/app/test-pmd/macfwd.c
@@ -119,9 +119,22 @@ pkt_burst_mac_forward(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+stream_init_mac_forward(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine mac_fwd_engine = {
 	.fwd_mode_name  = "mac",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_mac_forward,
 	.packet_fwd     = pkt_burst_mac_forward,
 };
diff --git a/app/test-pmd/macswap.c b/app/test-pmd/macswap.c
index 4627ff83e9..acb0fd7fb4 100644
--- a/app/test-pmd/macswap.c
+++ b/app/test-pmd/macswap.c
@@ -97,9 +97,22 @@ pkt_burst_mac_swap(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+stream_init_mac_swap(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine mac_swap_engine = {
 	.fwd_mode_name  = "macswap",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_mac_swap,
 	.packet_fwd     = pkt_burst_mac_swap,
 };
diff --git a/app/test-pmd/noisy_vnf.c b/app/test-pmd/noisy_vnf.c
index e4434bea95..a92e810190 100644
--- a/app/test-pmd/noisy_vnf.c
+++ b/app/test-pmd/noisy_vnf.c
@@ -277,9 +277,22 @@ noisy_fwd_begin(portid_t pi)
 	return 0;
 }
 
+static void
+stream_init_noisy_vnf(struct fwd_stream *fs)
+{
+	bool rx_stopped, tx_stopped;
+
+	rx_stopped = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	tx_stopped = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+	fs->disabled = rx_stopped || tx_stopped;
+}
+
 struct fwd_engine noisy_vnf_engine = {
 	.fwd_mode_name  = "noisy",
 	.port_fwd_begin = noisy_fwd_begin,
 	.port_fwd_end   = noisy_fwd_end,
+	.stream_init    = stream_init_noisy_vnf,
 	.packet_fwd     = pkt_burst_noisy_vnf,
 };
diff --git a/app/test-pmd/rxonly.c b/app/test-pmd/rxonly.c
index d1a579d8d8..04457010f4 100644
--- a/app/test-pmd/rxonly.c
+++ b/app/test-pmd/rxonly.c
@@ -68,9 +68,17 @@ pkt_burst_receive(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+stream_init_receive(struct fwd_stream *fs)
+{
+	fs->disabled = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+}
+
 struct fwd_engine rx_only_engine = {
 	.fwd_mode_name  = "rxonly",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = stream_init_receive,
 	.packet_fwd     = pkt_burst_receive,
 };
diff --git a/app/test-pmd/shared_rxq_fwd.c b/app/test-pmd/shared_rxq_fwd.c
index da54a383fd..2e9047804b 100644
--- a/app/test-pmd/shared_rxq_fwd.c
+++ b/app/test-pmd/shared_rxq_fwd.c
@@ -107,9 +107,17 @@ shared_rxq_fwd(struct fwd_stream *fs)
 	get_end_cycles(fs, start_tsc);
 }
 
+static void
+shared_rxq_stream_init(struct fwd_stream *fs)
+{
+	fs->disabled = ports[fs->rx_port].rxq[fs->rx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+}
+
 struct fwd_engine shared_rxq_engine = {
 	.fwd_mode_name  = "shared_rxq",
 	.port_fwd_begin = NULL,
 	.port_fwd_end   = NULL,
+	.stream_init    = shared_rxq_stream_init,
 	.packet_fwd     = shared_rxq_fwd,
 };
diff --git a/app/test-pmd/testpmd.c b/app/test-pmd/testpmd.c
index fe2ce19f99..52175a6cd2 100644
--- a/app/test-pmd/testpmd.c
+++ b/app/test-pmd/testpmd.c
@@ -1573,10 +1573,10 @@ init_config_port_offloads(portid_t pid, uint32_t socket_id)
 
 	/* Apply Rx offloads configuration */
 	for (i = 0; i < port->dev_info.max_rx_queues; i++)
-		port->rx_conf[i].offloads = port->dev_conf.rxmode.offloads;
+		port->rxq[i].conf.offloads = port->dev_conf.rxmode.offloads;
 	/* Apply Tx offloads configuration */
 	for (i = 0; i < port->dev_info.max_tx_queues; i++)
-		port->tx_conf[i].offloads = port->dev_conf.txmode.offloads;
+		port->txq[i].conf.offloads = port->dev_conf.txmode.offloads;
 
 	if (eth_link_speed)
 		port->dev_conf.link_speeds = eth_link_speed;
@@ -1763,7 +1763,6 @@ reconfig(portid_t new_port_id, unsigned socket_id)
 	init_port_config();
 }
 
-
 int
 init_fwd_streams(void)
 {
@@ -2156,6 +2155,12 @@ flush_fwd_rx_queues(void)
 		for (rxp = 0; rxp < cur_fwd_config.nb_fwd_ports; rxp++) {
 			for (rxq = 0; rxq < nb_rxq; rxq++) {
 				port_id = fwd_ports_ids[rxp];
+
+				/* Polling stopped queues is prohibited. */
+				if (ports[port_id].rxq[rxq].state ==
+				    RTE_ETH_QUEUE_STATE_STOPPED)
+					continue;
+
 				/**
 				* testpmd can stuck in the below do while loop
 				* if rte_eth_rx_burst() always returns nonzero
@@ -2201,7 +2206,8 @@ run_pkt_fwd_on_lcore(struct fwd_lcore *fc, packet_fwd_t pkt_fwd)
 	nb_fs = fc->stream_nb;
 	do {
 		for (sm_id = 0; sm_id < nb_fs; sm_id++)
-			(*pkt_fwd)(fsm[sm_id]);
+			if (!fsm[sm_id]->disabled)
+				(*pkt_fwd)(fsm[sm_id]);
 #ifdef RTE_LIB_BITRATESTATS
 		if (bitrate_enabled != 0 &&
 				bitrate_lcore_id == rte_lcore_id()) {
@@ -2283,6 +2289,7 @@ start_packet_forwarding(int with_tx_first)
 {
 	port_fwd_begin_t port_fwd_begin;
 	port_fwd_end_t  port_fwd_end;
+	stream_init_t stream_init = cur_fwd_eng->stream_init;
 	unsigned int i;
 
 	if (strcmp(cur_fwd_eng->fwd_mode_name, "rxonly") == 0 && !nb_rxq)
@@ -2313,6 +2320,10 @@ start_packet_forwarding(int with_tx_first)
 	if (!pkt_fwd_shared_rxq_check())
 		return;
 
+	if (stream_init != NULL)
+		for (i = 0; i < cur_fwd_config.nb_fwd_streams; i++)
+			stream_init(fwd_streams[i]);
+
 	port_fwd_begin = cur_fwd_config.fwd_eng->port_fwd_begin;
 	if (port_fwd_begin != NULL) {
 		for (i = 0; i < cur_fwd_config.nb_fwd_ports; i++) {
@@ -2574,7 +2585,7 @@ rx_queue_setup(uint16_t port_id, uint16_t rx_queue_id,
 		ret = rte_eth_rx_queue_setup(port_id, rx_queue_id,
 					     nb_rx_desc, socket_id,
 					     rx_conf, mp);
-		return ret;
+		goto exit;
 	}
 	for (i = 0; i < rx_pkt_nb_segs; i++) {
 		struct rte_eth_rxseg_split *rx_seg = &rx_useg[i].split;
@@ -2599,6 +2610,10 @@ rx_queue_setup(uint16_t port_id, uint16_t rx_queue_id,
 				    socket_id, rx_conf, NULL);
 	rx_conf->rx_seg = NULL;
 	rx_conf->rx_nseg = 0;
+exit:
+	ports[port_id].rxq[rx_queue_id].state = rx_conf->rx_deferred_start ?
+						RTE_ETH_QUEUE_STATE_STOPPED :
+						RTE_ETH_QUEUE_STATE_STARTED;
 	return ret;
 }
 
@@ -2801,7 +2816,7 @@ start_port(portid_t pid)
 				for (k = 0;
 				     k < port->dev_info.max_rx_queues;
 				     k++)
-					port->rx_conf[k].offloads |=
+					port->rxq[k].conf.offloads |=
 						dev_conf.rxmode.offloads;
 			}
 			/* Apply Tx offloads configuration */
@@ -2812,7 +2827,7 @@ start_port(portid_t pid)
 				for (k = 0;
 				     k < port->dev_info.max_tx_queues;
 				     k++)
-					port->tx_conf[k].offloads |=
+					port->txq[k].conf.offloads |=
 						dev_conf.txmode.offloads;
 			}
 		}
@@ -2820,20 +2835,28 @@ start_port(portid_t pid)
 			port->need_reconfig_queues = 0;
 			/* setup tx queues */
 			for (qi = 0; qi < nb_txq; qi++) {
+				struct rte_eth_txconf *conf =
+							&port->txq[qi].conf;
+
 				if ((numa_support) &&
 					(txring_numa[pi] != NUMA_NO_CONFIG))
 					diag = rte_eth_tx_queue_setup(pi, qi,
 						port->nb_tx_desc[qi],
 						txring_numa[pi],
-						&(port->tx_conf[qi]));
+						&(port->txq[qi].conf));
 				else
 					diag = rte_eth_tx_queue_setup(pi, qi,
 						port->nb_tx_desc[qi],
 						port->socket_id,
-						&(port->tx_conf[qi]));
+						&(port->txq[qi].conf));
 
-				if (diag == 0)
+				if (diag == 0) {
+					port->txq[qi].state =
+						conf->tx_deferred_start ?
+						RTE_ETH_QUEUE_STATE_STOPPED :
+						RTE_ETH_QUEUE_STATE_STARTED;
 					continue;
+				}
 
 				/* Fail to setup tx queue, return */
 				if (port->port_status == RTE_PORT_HANDLING)
@@ -2866,7 +2889,7 @@ start_port(portid_t pid)
 					diag = rx_queue_setup(pi, qi,
 					     port->nb_rx_desc[qi],
 					     rxring_numa[pi],
-					     &(port->rx_conf[qi]),
+					     &(port->rxq[qi].conf),
 					     mp);
 				} else {
 					struct rte_mempool *mp =
@@ -2881,7 +2904,7 @@ start_port(portid_t pid)
 					diag = rx_queue_setup(pi, qi,
 					     port->nb_rx_desc[qi],
 					     port->socket_id,
-					     &(port->rx_conf[qi]),
+					     &(port->rxq[qi].conf),
 					     mp);
 				}
 				if (diag == 0)
@@ -3656,59 +3679,59 @@ rxtx_port_config(portid_t pid)
 	struct rte_port *port = &ports[pid];
 
 	for (qid = 0; qid < nb_rxq; qid++) {
-		offloads = port->rx_conf[qid].offloads;
-		port->rx_conf[qid] = port->dev_info.default_rxconf;
+		offloads = port->rxq[qid].conf.offloads;
+		port->rxq[qid].conf = port->dev_info.default_rxconf;
 
 		if (rxq_share > 0 &&
 		    (port->dev_info.dev_capa & RTE_ETH_DEV_CAPA_RXQ_SHARE)) {
 			/* Non-zero share group to enable RxQ share. */
-			port->rx_conf[qid].share_group = pid / rxq_share + 1;
-			port->rx_conf[qid].share_qid = qid; /* Equal mapping. */
+			port->rxq[qid].conf.share_group = pid / rxq_share + 1;
+			port->rxq[qid].conf.share_qid = qid; /* Equal mapping. */
 		}
 
 		if (offloads != 0)
-			port->rx_conf[qid].offloads = offloads;
+			port->rxq[qid].conf.offloads = offloads;
 
 		/* Check if any Rx parameters have been passed */
 		if (rx_pthresh != RTE_PMD_PARAM_UNSET)
-			port->rx_conf[qid].rx_thresh.pthresh = rx_pthresh;
+			port->rxq[qid].conf.rx_thresh.pthresh = rx_pthresh;
 
 		if (rx_hthresh != RTE_PMD_PARAM_UNSET)
-			port->rx_conf[qid].rx_thresh.hthresh = rx_hthresh;
+			port->rxq[qid].conf.rx_thresh.hthresh = rx_hthresh;
 
 		if (rx_wthresh != RTE_PMD_PARAM_UNSET)
-			port->rx_conf[qid].rx_thresh.wthresh = rx_wthresh;
+			port->rxq[qid].conf.rx_thresh.wthresh = rx_wthresh;
 
 		if (rx_free_thresh != RTE_PMD_PARAM_UNSET)
-			port->rx_conf[qid].rx_free_thresh = rx_free_thresh;
+			port->rxq[qid].conf.rx_free_thresh = rx_free_thresh;
 
 		if (rx_drop_en != RTE_PMD_PARAM_UNSET)
-			port->rx_conf[qid].rx_drop_en = rx_drop_en;
+			port->rxq[qid].conf.rx_drop_en = rx_drop_en;
 
 		port->nb_rx_desc[qid] = nb_rxd;
 	}
 
 	for (qid = 0; qid < nb_txq; qid++) {
-		offloads = port->tx_conf[qid].offloads;
-		port->tx_conf[qid] = port->dev_info.default_txconf;
+		offloads = port->txq[qid].conf.offloads;
+		port->txq[qid].conf = port->dev_info.default_txconf;
 		if (offloads != 0)
-			port->tx_conf[qid].offloads = offloads;
+			port->txq[qid].conf.offloads = offloads;
 
 		/* Check if any Tx parameters have been passed */
 		if (tx_pthresh != RTE_PMD_PARAM_UNSET)
-			port->tx_conf[qid].tx_thresh.pthresh = tx_pthresh;
+			port->txq[qid].conf.tx_thresh.pthresh = tx_pthresh;
 
 		if (tx_hthresh != RTE_PMD_PARAM_UNSET)
-			port->tx_conf[qid].tx_thresh.hthresh = tx_hthresh;
+			port->txq[qid].conf.tx_thresh.hthresh = tx_hthresh;
 
 		if (tx_wthresh != RTE_PMD_PARAM_UNSET)
-			port->tx_conf[qid].tx_thresh.wthresh = tx_wthresh;
+			port->txq[qid].conf.tx_thresh.wthresh = tx_wthresh;
 
 		if (tx_rs_thresh != RTE_PMD_PARAM_UNSET)
-			port->tx_conf[qid].tx_rs_thresh = tx_rs_thresh;
+			port->txq[qid].conf.tx_rs_thresh = tx_rs_thresh;
 
 		if (tx_free_thresh != RTE_PMD_PARAM_UNSET)
-			port->tx_conf[qid].tx_free_thresh = tx_free_thresh;
+			port->txq[qid].conf.tx_free_thresh = tx_free_thresh;
 
 		port->nb_tx_desc[qid] = nb_txd;
 	}
@@ -3789,7 +3812,7 @@ init_port_config(void)
 				for (i = 0;
 				     i < port->dev_info.nb_rx_queues;
 				     i++)
-					port->rx_conf[i].offloads &=
+					port->rxq[i].conf.offloads &=
 						~RTE_ETH_RX_OFFLOAD_RSS_HASH;
 			}
 		}
@@ -3963,7 +3986,7 @@ init_port_dcb_config(portid_t pid,
 	if (port_conf.rxmode.mq_mode == RTE_ETH_MQ_RX_VMDQ_DCB) {
 		port_conf.rxmode.offloads &= ~RTE_ETH_RX_OFFLOAD_RSS_HASH;
 		for (i = 0; i < nb_rxq; i++)
-			rte_port->rx_conf[i].offloads &=
+			rte_port->rxq[i].conf.offloads &=
 				~RTE_ETH_RX_OFFLOAD_RSS_HASH;
 	}
 
diff --git a/app/test-pmd/testpmd.h b/app/test-pmd/testpmd.h
index 31f766c965..1eda7b97ab 100644
--- a/app/test-pmd/testpmd.h
+++ b/app/test-pmd/testpmd.h
@@ -134,6 +134,7 @@ struct fwd_stream {
 	portid_t   tx_port;   /**< forwarding port of received packets */
 	queueid_t  tx_queue;  /**< TX queue to send forwarded packets */
 	streamid_t peer_addr; /**< index of peer ethernet address of packets */
+	bool       disabled;  /**< the stream is disabled and should not run */
 
 	unsigned int retry_enabled;
 
@@ -238,6 +239,18 @@ struct xstat_display_info {
 	bool	 allocated;
 };
 
+/** RX queue configuration and state. */
+struct port_rxqueue {
+	struct rte_eth_rxconf conf;
+	uint8_t state; /**< RTE_ETH_QUEUE_STATE_* value. */
+};
+
+/** TX queue configuration and state. */
+struct port_txqueue {
+	struct rte_eth_txconf conf;
+	uint8_t state; /**< RTE_ETH_QUEUE_STATE_* value. */
+};
+
 /**
  * The data structure associated with each port.
  */
@@ -260,8 +273,8 @@ struct rte_port {
 	uint8_t                 dcb_flag;   /**< enable dcb */
 	uint16_t                nb_rx_desc[RTE_MAX_QUEUES_PER_PORT+1]; /**< per queue rx desc number */
 	uint16_t                nb_tx_desc[RTE_MAX_QUEUES_PER_PORT+1]; /**< per queue tx desc number */
-	struct rte_eth_rxconf   rx_conf[RTE_MAX_QUEUES_PER_PORT+1]; /**< per queue rx configuration */
-	struct rte_eth_txconf   tx_conf[RTE_MAX_QUEUES_PER_PORT+1]; /**< per queue tx configuration */
+	struct port_rxqueue     rxq[RTE_MAX_QUEUES_PER_PORT+1]; /**< per queue rx configuration and state */
+	struct port_txqueue     txq[RTE_MAX_QUEUES_PER_PORT+1]; /**< per queue tx configuration and state */
 	struct rte_ether_addr   *mc_addr_pool; /**< pool of multicast addrs */
 	uint32_t                mc_addr_nb; /**< nb. of addr. in mc_addr_pool */
 	queueid_t               queue_nb; /**< nb. of queues for flow rules */
@@ -323,12 +336,14 @@ struct fwd_lcore {
  */
 typedef int (*port_fwd_begin_t)(portid_t pi);
 typedef void (*port_fwd_end_t)(portid_t pi);
+typedef void (*stream_init_t)(struct fwd_stream *fs);
 typedef void (*packet_fwd_t)(struct fwd_stream *fs);
 
 struct fwd_engine {
 	const char       *fwd_mode_name; /**< Forwarding mode name. */
 	port_fwd_begin_t port_fwd_begin; /**< NULL if nothing special to do. */
 	port_fwd_end_t   port_fwd_end;   /**< NULL if nothing special to do. */
+	stream_init_t    stream_init;    /**< NULL if nothing special to do. */
 	packet_fwd_t     packet_fwd;     /**< Mandatory. */
 };
 
diff --git a/app/test-pmd/txonly.c b/app/test-pmd/txonly.c
index fc039a622c..e1bc78b73d 100644
--- a/app/test-pmd/txonly.c
+++ b/app/test-pmd/txonly.c
@@ -504,9 +504,17 @@ tx_only_begin(portid_t pi)
 	return 0;
 }
 
+static void
+tx_only_stream_init(struct fwd_stream *fs)
+{
+	fs->disabled = ports[fs->tx_port].txq[fs->tx_queue].state ==
+						RTE_ETH_QUEUE_STATE_STOPPED;
+}
+
 struct fwd_engine tx_only_engine = {
 	.fwd_mode_name  = "txonly",
 	.port_fwd_begin = tx_only_begin,
 	.port_fwd_end   = NULL,
+	.stream_init    = tx_only_stream_init,
 	.packet_fwd     = pkt_burst_transmit,
 };
-- 
2.25.1

RE: [PATCH v3 1/2] app/testpmd: do not poll stopped queues

[Dmitry Kozlyuk]

> Calling Rx/Tx functions on a stopped queue is not supported.
> Do not run packet forwarding for streams that use stopped queues.
> 
> Each stream has a read-only "disabled" field,
> so that lcore function can skip such streams.
> Forwarding engines can set this field
> using a new "stream_init" callback function
> by checking relevant queue states,
> which are stored along with queue configurations
> (not all PMDs implement rte_eth_rx/tx_queue_info_get()
> to query the state from there).
> 
> Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
> Cc: stable@dpdk.org
> 
> Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> Acked-by: Matan Azrad <matan@nvidia.com>
> ---

Pinging testpmd maintainers for this bugfix.
The sibling ethdev patch will have to wait until the next release cycle.

Re: [PATCH v3 1/2] app/testpmd: do not poll stopped queues

[Stephen Hemminger]

On Wed, 9 Mar 2022 10:36:15 +0000
Dmitry Kozlyuk <dkozlyuk@nvidia.com> wrote:

> > Calling Rx/Tx functions on a stopped queue is not supported.
> > Do not run packet forwarding for streams that use stopped queues.
> > 
> > Each stream has a read-only "disabled" field,
> > so that lcore function can skip such streams.
> > Forwarding engines can set this field
> > using a new "stream_init" callback function
> > by checking relevant queue states,
> > which are stored along with queue configurations
> > (not all PMDs implement rte_eth_rx/tx_queue_info_get()
> > to query the state from there).
> > 
> > Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
> > Cc: stable@dpdk.org
> > 
> > Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> > Acked-by: Matan Azrad <matan@nvidia.com>
> > ---  
> 
> Pinging testpmd maintainers for this bugfix.
> The sibling ethdev patch will have to wait until the next release cycle.

So you broke a bunch of drivers now that is merged???
Sound like it needs to be reverted. Certainly not something that should
go into stable tree if you expect all drivers to correctly set queue state.

Re: [PATCH v3 1/2] app/testpmd: do not poll stopped queues

[Thomas Monjalon]

07/03/2022 13:53, Dmitry Kozlyuk:
> Calling Rx/Tx functions on a stopped queue is not supported.
> Do not run packet forwarding for streams that use stopped queues.
> 
> Each stream has a read-only "disabled" field,
> so that lcore function can skip such streams.
> Forwarding engines can set this field
> using a new "stream_init" callback function
> by checking relevant queue states,
> which are stored along with queue configurations
> (not all PMDs implement rte_eth_rx/tx_queue_info_get()
> to query the state from there).
> 
> Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
> Cc: stable@dpdk.org
> 
> Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> Acked-by: Matan Azrad <matan@nvidia.com>

This patch is waiting for long without any comment.

Applied in next-net, thanks.

RE: [PATCH v3 1/2] app/testpmd: do not poll stopped queues

[Jiang, YuX]

Hi Kozlyuk,

Here is bugzilla ticket https://bugs.dpdk.org/show_bug.cgi?id=1028, its bad commit id is this patch.
Could you please have a look? Thanks so much.

Best regards,
Yu Jiang

> -----Original Message-----
> From: Thomas Monjalon <thomas@monjalon.net>
> Sent: Wednesday, May 25, 2022 11:46 PM
> To: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> Cc: dev@dpdk.org; stable@dpdk.org; Matan Azrad <matan@nvidia.com>; Li,
> Xiaoyun <xiaoyun.li@intel.com>; Singh, Aman Deep
> <aman.deep.singh@intel.com>; Zhang, Yuying <yuying.zhang@intel.com>
> Subject: Re: [PATCH v3 1/2] app/testpmd: do not poll stopped queues
> 
> 07/03/2022 13:53, Dmitry Kozlyuk:
> > Calling Rx/Tx functions on a stopped queue is not supported.
> > Do not run packet forwarding for streams that use stopped queues.
> >
> > Each stream has a read-only "disabled" field, so that lcore function
> > can skip such streams.
> > Forwarding engines can set this field
> > using a new "stream_init" callback function by checking relevant queue
> > states, which are stored along with queue configurations (not all PMDs
> > implement rte_eth_rx/tx_queue_info_get() to query the state from
> > there).
> >
> > Fixes: 5f4ec54f1d16 ("testpmd: queue start and stop")
> > Cc: stable@dpdk.org
> >
> > Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
> > Acked-by: Matan Azrad <matan@nvidia.com>
> 
> This patch is waiting for long without any comment.
> 
> Applied in next-net, thanks.
> 

[PATCH v3 2/2] ethdev: prohibit polling of a stopped queue

[Dmitry Kozlyuk]

Whether it is allowed to call Rx/Tx functions for a stopped queue
was undocumented. Some PMDs make this behavior a no-op
either by explicitly checking the queue state
or by the way how their routines are implemented or HW works.

No-op behavior may be convenient for application developers.
But it also means that pollers of stopped queues
would go all the way down to PMD Rx/Tx routines, wasting cycles.
Some PMDs would do a check for the queue state on data path,
even though it may never be needed for a particular application.
Also, use cases for stopping queues or starting them deferred
do not logically require polling stopped queues.

Use case 1: a secondary that was polling the queue has crashed,
the primary is doing a recovery to free all mbufs.
By definition the queue to be restarted is not polled.

Use case 2: deferred queue start or queue reconfiguration.
The polling thread must be synchronized anyway,
because queue start and stop are non-atomic.

Prohibit calling Rx/Tx functions on stopped queues.

Fixes: 0748be2cf9a2 ("ethdev: queue start and stop")
Cc: stable@dpdk.org

Signed-off-by: Dmitry Kozlyuk <dkozlyuk@nvidia.com>
Acked-by: Matan Azrad <matan@nvidia.com>
---
 lib/ethdev/rte_ethdev.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/ethdev/rte_ethdev.h b/lib/ethdev/rte_ethdev.h
index c2d1f9a972..9f12a6043c 100644
--- a/lib/ethdev/rte_ethdev.h
+++ b/lib/ethdev/rte_ethdev.h
@@ -74,7 +74,7 @@
  * rte_eth_rx_queue_setup()), it must call rte_eth_dev_stop() first to stop the
  * device and then do the reconfiguration before calling rte_eth_dev_start()
  * again. The transmit and receive functions should not be invoked when the
- * device is stopped.
+ * device is stopped or when the queue is stopped (for that queue).
  *
  * Please note that some configuration is not stored between calls to
  * rte_eth_dev_stop()/rte_eth_dev_start(). The following configuration will
-- 
2.25.1