[dpdk-dev] [PATCH v6 02/14] ethdev: add support for hairpin queue

[Andrew Rybchenko]

Hi Ori,

On 10/27/19 3:24 PM, Ori Kam wrote:
> This commit introduce hairpin queue type.
>
> The hairpin queue in build from Rx queue binded to Tx queue.
> It is used to offload traffic coming from the wire and redirect it back
> to the wire.
>
> There are 3 new functions:
> - rte_eth_dev_hairpin_capability_get
> - rte_eth_rx_hairpin_queue_setup
> - rte_eth_tx_hairpin_queue_setup
>
> In order to use the queue, there is a need to create rte_flow
> with queue / RSS action that targets one or more of the Rx queues.
>
> Signed-off-by: Ori Kam <orika at mellanox.com>
> Reviewed-by: Andrew Rybchenko <arybchenko at solarflare.com>

LGTM, nothing critical may be except maximum number check
which I lost from my view before.
Plus few style suggestions which may be dropped, but I'd be
happier if applied.

Thanks.

[snip]

> diff --git a/lib/librte_ethdev/rte_ethdev.c b/lib/librte_ethdev/rte_ethdev.c
> index 7743205..68aca1f 100644
> --- a/lib/librte_ethdev/rte_ethdev.c
> +++ b/lib/librte_ethdev/rte_ethdev.c
> @@ -923,6 +923,13 @@ struct rte_eth_dev *
>   
>   	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->rx_queue_start, -ENOTSUP);
>   
> +	if (rte_eth_dev_is_rx_hairpin_queue(dev, rx_queue_id)) {
> +		RTE_ETHDEV_LOG(INFO,
> +			"Can't start Rx queue %"PRIu16" of device with port_id=%"PRIu16" is hairpin queue\n",

Log message looks a bit strange:
     Can't start Rx queue 5 of device with port_id=0 is hairpin queue
may be to put key information first:
     Can't start hairpin Rx queue 5 of device with port_id=0

> +			rx_queue_id, port_id);
> +		return -EINVAL;
> +	}
> +
>   	if (dev->data->rx_queue_state[rx_queue_id] != RTE_ETH_QUEUE_STATE_STOPPED) {
>   		RTE_ETHDEV_LOG(INFO,
>   			"Queue %"PRIu16" of device with port_id=%"PRIu16" already started\n",
> @@ -950,6 +957,13 @@ struct rte_eth_dev *
>   
>   	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->rx_queue_stop, -ENOTSUP);
>   
> +	if (rte_eth_dev_is_rx_hairpin_queue(dev, rx_queue_id)) {
> +		RTE_ETHDEV_LOG(INFO,
> +			"Can't stop Rx queue %"PRIu16" of device with port_id=%"PRIu16" is hairpin queue\n",

Same

> +			rx_queue_id, port_id);
> +		return -EINVAL;
> +	}
> +
>   	if (dev->data->rx_queue_state[rx_queue_id] == RTE_ETH_QUEUE_STATE_STOPPED) {
>   		RTE_ETHDEV_LOG(INFO,
>   			"Queue %"PRIu16" of device with port_id=%"PRIu16" already stopped\n",
> @@ -983,6 +997,13 @@ struct rte_eth_dev *
>   
>   	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->tx_queue_start, -ENOTSUP);
>   
> +	if (rte_eth_dev_is_tx_hairpin_queue(dev, tx_queue_id)) {
> +		RTE_ETHDEV_LOG(INFO,
> +			"Can't start Tx queue %"PRIu16" of device with port_id=%"PRIu16" is hairpin queue\n",

Same

> +			tx_queue_id, port_id);
> +		return -EINVAL;
> +	}
> +
>   	if (dev->data->tx_queue_state[tx_queue_id] != RTE_ETH_QUEUE_STATE_STOPPED) {
>   		RTE_ETHDEV_LOG(INFO,
>   			"Queue %"PRIu16" of device with port_id=%"PRIu16" already started\n",
> @@ -1008,6 +1029,13 @@ struct rte_eth_dev *
>   
>   	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->tx_queue_stop, -ENOTSUP);
>   
> +	if (rte_eth_dev_is_tx_hairpin_queue(dev, tx_queue_id)) {
> +		RTE_ETHDEV_LOG(INFO,
> +			"Can't stop Tx queue %"PRIu16" of device with port_id=%"PRIu16" is hairpin queue\n",

Same

> +			tx_queue_id, port_id);
> +		return -EINVAL;
> +	}
> +
>   	if (dev->data->tx_queue_state[tx_queue_id] == RTE_ETH_QUEUE_STATE_STOPPED) {
>   		RTE_ETHDEV_LOG(INFO,
>   			"Queue %"PRIu16" of device with port_id=%"PRIu16" already stopped\n",
> @@ -1780,6 +1808,79 @@ struct rte_eth_dev *
>   }
>   
>   int
> +rte_eth_rx_hairpin_queue_setup(uint16_t port_id, uint16_t rx_queue_id,
> +			       uint16_t nb_rx_desc,
> +			       const struct rte_eth_hairpin_conf *conf)
> +{
> +	int ret;
> +	struct rte_eth_dev *dev;
> +	struct rte_eth_hairpin_cap cap;
> +	void **rxq;
> +	int i;
> +	int count = 0;
> +
> +	RTE_ETH_VALID_PORTID_OR_ERR_RET(port_id, -EINVAL);
> +
> +	dev = &rte_eth_devices[port_id];
> +	if (rx_queue_id >= dev->data->nb_rx_queues) {
> +		RTE_ETHDEV_LOG(ERR, "Invalid RX queue_id=%u\n", rx_queue_id);
> +		return -EINVAL;
> +	}
> +	ret = rte_eth_dev_hairpin_capability_get(port_id, &cap);
> +	if (ret != 0)
> +		return ret;
> +	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->rx_hairpin_queue_setup,
> +				-ENOTSUP);

Most likely unsupported hairpin is caught by capability get above.
So, may be it is better to move the check just before usage far below.
Also, if line length is sufficient I think it would better to put -ENOTSUP
to the previous line just to follow port_id check style.

> +	/* Use default specified by driver, if nb_rx_desc is zero */
> +	if (nb_rx_desc == 0)
> +		nb_rx_desc = cap.max_nb_desc;

Function description and comment above mentions PMD default, but
there is no default. It uses just maximum. I have no strong opinion
if default is really required or it is OK to say that maximum is used.
The only concern is: why maximum?

> +	if (nb_rx_desc > cap.max_nb_desc) {
> +		RTE_ETHDEV_LOG(ERR,
> +			"Invalid value for nb_rx_desc(=%hu), should be: <= %hu",
> +			nb_rx_desc, cap.max_nb_desc);
> +		return -EINVAL;
> +	}
> +	if (conf->peer_count > cap.max_rx_2_tx) {
> +		RTE_ETHDEV_LOG(ERR,
> +			"Invalid value for number of peers for Rx queue(=%hu), should be: <= %hu",
> +			conf->peer_count, cap.max_rx_2_tx);
> +		return -EINVAL;
> +	}
> +	if (conf->peer_count == 0) {
> +		RTE_ETHDEV_LOG(ERR,
> +			"Invalid value for number of peers for Rx queue(=%hu), should be: > 0",
> +			conf->peer_count);
> +		return -EINVAL;
> +	}
> +	if (cap.max_nb_queues != UINT16_MAX) {

I'm not sure that we need to handle it separately. Code below
should handle it and it and there is no point to optimize it.

> +		for (i = 0; i < dev->data->nb_rx_queues; i++) {

May I suggest to assign count = 0 to make it a bit easier to read and
more robust against future changes.

> +			if (rte_eth_dev_is_rx_hairpin_queue(dev, i))

The condition should be more tricky if we resetup hairpin queue.
I.e. we should check if i is rx_queue_id and count it anyway.

> +				count++;
> +		}
> +		if (count > cap.max_nb_queues) {
> +			RTE_ETHDEV_LOG(ERR, "To many Rx hairpin queues %d",

I think it would be useful to log max here as well to catch
unset max cases easier.

> +			count);
> +			return -EINVAL;
> +		}
> +	}
> +	if (dev->data->dev_started)
> +		return -EBUSY;
> +	rxq = dev->data->rx_queues;
> +	if (rxq[rx_queue_id] != NULL) {
> +		RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->rx_queue_release,
> +					-ENOTSUP);
> +		(*dev->dev_ops->rx_queue_release)(rxq[rx_queue_id]);
> +		rxq[rx_queue_id] = NULL;
> +	}
> +	ret = (*dev->dev_ops->rx_hairpin_queue_setup)(dev, rx_queue_id,
> +						      nb_rx_desc, conf);
> +	if (ret == 0)
> +		dev->data->rx_queue_state[rx_queue_id] =
> +			RTE_ETH_QUEUE_STATE_HAIRPIN;
> +	return eth_err(port_id, ret);
> +}
> +
> +int
>   rte_eth_tx_queue_setup(uint16_t port_id, uint16_t tx_queue_id,
>   		       uint16_t nb_tx_desc, unsigned int socket_id,
>   		       const struct rte_eth_txconf *tx_conf)
> @@ -1878,6 +1979,78 @@ struct rte_eth_dev *
>   		       tx_queue_id, nb_tx_desc, socket_id, &local_conf));
>   }
>   
> +int
> +rte_eth_tx_hairpin_queue_setup(uint16_t port_id, uint16_t tx_queue_id,
> +			       uint16_t nb_tx_desc,
> +			       const struct rte_eth_hairpin_conf *conf)

Same notes as for Rx queue above.

> +{
> +	struct rte_eth_dev *dev;
> +	struct rte_eth_hairpin_cap cap;
> +	void **txq;
> +	int i;
> +	int count = 0;
> +	int ret;
> +
> +	RTE_ETH_VALID_PORTID_OR_ERR_RET(port_id, -EINVAL);
> +	dev = &rte_eth_devices[port_id];
> +	if (tx_queue_id >= dev->data->nb_tx_queues) {
> +		RTE_ETHDEV_LOG(ERR, "Invalid TX queue_id=%u\n", tx_queue_id);
> +		return -EINVAL;
> +	}
> +	ret = rte_eth_dev_hairpin_capability_get(port_id, &cap);
> +	if (ret != 0)
> +		return ret;
> +	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->tx_hairpin_queue_setup,
> +				-ENOTSUP);
> +	/* Use default specified by driver, if nb_tx_desc is zero */
> +	if (nb_tx_desc == 0)
> +		nb_tx_desc = cap.max_nb_desc;
> +	if (nb_tx_desc > cap.max_nb_desc) {
> +		RTE_ETHDEV_LOG(ERR,
> +			"Invalid value for nb_tx_desc(=%hu), should be: <= %hu",
> +			nb_tx_desc, cap.max_nb_desc);
> +		return -EINVAL;
> +	}
> +	if (conf->peer_count > cap.max_tx_2_rx) {
> +		RTE_ETHDEV_LOG(ERR,
> +			"Invalid value for number of peers for Tx queue(=%hu), should be: <= %hu",
> +			conf->peer_count, cap.max_tx_2_rx);
> +		return -EINVAL;
> +	}
> +	if (conf->peer_count == 0) {
> +		RTE_ETHDEV_LOG(ERR,
> +			"Invalid value for number of peers for Tx queue(=%hu), should be: > 0",
> +			conf->peer_count);
> +		return -EINVAL;
> +	}
> +	if (cap.max_nb_queues != UINT16_MAX) {
> +		for (i = 0; i < dev->data->nb_tx_queues; i++) {
> +			if (rte_eth_dev_is_tx_hairpin_queue(dev, i))
> +				count++;
> +		}
> +		if (count > cap.max_nb_queues) {
> +			RTE_ETHDEV_LOG(ERR,
> +				       "To many Tx hairpin queues %d", count);
> +			return -EINVAL;
> +		}
> +	}
> +	if (dev->data->dev_started)
> +		return -EBUSY;
> +	txq = dev->data->tx_queues;
> +	if (txq[tx_queue_id] != NULL) {
> +		RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->tx_queue_release,
> +					-ENOTSUP);
> +		(*dev->dev_ops->tx_queue_release)(txq[tx_queue_id]);
> +		txq[tx_queue_id] = NULL;
> +	}
> +	ret = (*dev->dev_ops->tx_hairpin_queue_setup)
> +		(dev, tx_queue_id, nb_tx_desc, conf);
> +	if (ret == 0)
> +		dev->data->tx_queue_state[tx_queue_id] =
> +			RTE_ETH_QUEUE_STATE_HAIRPIN;
> +	return eth_err(port_id, ret);
> +}
> +
>   void
>   rte_eth_tx_buffer_drop_callback(struct rte_mbuf **pkts, uint16_t unsent,
>   		void *userdata __rte_unused)
> @@ -4007,12 +4180,19 @@ int rte_eth_set_queue_rate_limit(uint16_t port_id, uint16_t queue_idx,
>   	rte_errno = ENOTSUP;
>   	return NULL;
>   #endif
> +	struct rte_eth_dev *dev;
> +
>   	/* check input parameters */
>   	if (!rte_eth_dev_is_valid_port(port_id) || fn == NULL ||
>   		    queue_id >= rte_eth_devices[port_id].data->nb_rx_queues) {
>   		rte_errno = EINVAL;
>   		return NULL;
>   	}
> +	dev = &rte_eth_devices[port_id];
> +	if (rte_eth_dev_is_rx_hairpin_queue(dev, queue_id)) {
> +		rte_errno = EINVAL;
> +		return NULL;
> +	}
>   	struct rte_eth_rxtx_callback *cb = rte_zmalloc(NULL, sizeof(*cb), 0);
>   
>   	if (cb == NULL) {
> @@ -4084,6 +4264,8 @@ int rte_eth_set_queue_rate_limit(uint16_t port_id, uint16_t queue_idx,
>   	rte_errno = ENOTSUP;
>   	return NULL;
>   #endif
> +	struct rte_eth_dev *dev;
> +
>   	/* check input parameters */
>   	if (!rte_eth_dev_is_valid_port(port_id) || fn == NULL ||
>   		    queue_id >= rte_eth_devices[port_id].data->nb_tx_queues) {
> @@ -4091,6 +4273,12 @@ int rte_eth_set_queue_rate_limit(uint16_t port_id, uint16_t queue_idx,
>   		return NULL;
>   	}
>   
> +	dev = &rte_eth_devices[port_id];
> +	if (rte_eth_dev_is_tx_hairpin_queue(dev, queue_id)) {
> +		rte_errno = EINVAL;
> +		return NULL;
> +	}
> +
>   	struct rte_eth_rxtx_callback *cb = rte_zmalloc(NULL, sizeof(*cb), 0);
>   
>   	if (cb == NULL) {
> @@ -4204,6 +4392,13 @@ int rte_eth_set_queue_rate_limit(uint16_t port_id, uint16_t queue_idx,
>   		return -EINVAL;
>   	}
>   
> +	if (rte_eth_dev_is_rx_hairpin_queue(dev, queue_id)) {
> +		RTE_ETHDEV_LOG(INFO,
> +			"Can't get queue info for Rx queue %"PRIu16" of device with port_id=%"PRIu16" is hairpin queue\n",

"queue" is repeated 3 times above ;) I'm afraid it is too much, may be:
"Can't get hairpin Rx queue %" PRIu16 " port %" PRIu16 " info\n"
or
"Can't get hairpin Rx queue %" PRIu16 " info of device with port_id=%" 
PRIu16 "\n"
Anyway up to you.

> +			queue_id, port_id);
> +		return -EINVAL;
> +	}
> +
>   	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->rxq_info_get, -ENOTSUP);
>   
>   	memset(qinfo, 0, sizeof(*qinfo));
> @@ -4228,6 +4423,13 @@ int rte_eth_set_queue_rate_limit(uint16_t port_id, uint16_t queue_idx,
>   		return -EINVAL;
>   	}
>   
> +	if (rte_eth_dev_is_tx_hairpin_queue(dev, queue_id)) {
> +		RTE_ETHDEV_LOG(INFO,
> +			"Can't get queue info for Tx queue %"PRIu16" of device with port_id=%"PRIu16" is hairpin queue\n",

Same

> +			queue_id, port_id);
> +		return -EINVAL;
> +	}
> +
>   	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->txq_info_get, -ENOTSUP);
>   
>   	memset(qinfo, 0, sizeof(*qinfo));
> @@ -4600,6 +4802,21 @@ int rte_eth_set_queue_rate_limit(uint16_t port_id, uint16_t queue_idx,
>   }
>   
>   int
> +rte_eth_dev_hairpin_capability_get(uint16_t port_id,
> +				   struct rte_eth_hairpin_cap *cap)
> +{
> +	struct rte_eth_dev *dev;
> +
> +	RTE_ETH_VALID_PORTID_OR_ERR_RET(port_id, -EINVAL);
> +
> +	dev = &rte_eth_devices[port_id];
> +	RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->hairpin_cap_get,
> +				-ENOTSUP);

Please, move -ENOTSUP to the previous line since line length is sufficient
and make it similar to port_id check above.

> +	memset(cap, 0, sizeof(*cap));
> +	return eth_err(port_id, (*dev->dev_ops->hairpin_cap_get)(dev, cap));
> +}
> +
> +int
>   rte_eth_dev_pool_ops_supported(uint16_t port_id, const char *pool)
>   {
>   	struct rte_eth_dev *dev;
> diff --git a/lib/librte_ethdev/rte_ethdev.h b/lib/librte_ethdev/rte_ethdev.h
> index 9e1f9ae..9b69255 100644
> --- a/lib/librte_ethdev/rte_ethdev.h
> +++ b/lib/librte_ethdev/rte_ethdev.h
> @@ -839,6 +839,46 @@ struct rte_eth_txconf {
>   };
>   
>   /**
> + * @warning
> + * @b EXPERIMENTAL: this API may change, or be removed, without prior notice
> + *
> + * A structure used to return the hairpin capabilities that are supported.
> + */
> +struct rte_eth_hairpin_cap {
> +	/** The max number of hairpin queues (different bindings). */
> +	uint16_t max_nb_queues;
> +	/**< Max number of Rx queues to be connected to one Tx queue. */

Should be /**

> +	uint16_t max_rx_2_tx;
> +	/**< Max number of Tx queues to be connected to one Rx queue. */

Should be /**

[snip]